from pydantic import BaseModel, Field
from typing import List
class WebSearchItem:
    reason: str = Field
    query: str = Field
class WebSearchPlan:
    searches: List = Field
class ReportData:
    short_summary: str = Field
    follow_up_questions: List = Field
    content: str = Field

Pydantic 会在运行时校验字段类型，一旦 LLM 输出不符合 schema，就会抛异常，我们随后会捕获并走兜底路径。

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
    model="qwen-plus",
    temperature=0,
    api_key=os.getenv,
    base_url=os.getenv,
)
planner_prompt = ChatPromptTemplate.from_messages()
planner_chain = planner_prompt | llm.with_structured_output

这里Zui关键的是 `with_structured_output` —— 它直接让模型返回一个Yi经实例化好的 Pydantic 对象，无需自行解析 JSON 字符串。

from langchain_community.tools import DuckDuckGoSearchResults
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
search_tool = DuckDuckGoSearchResults
search_chain = (
    RunnablePassthrough.assign(
        search_results=RunnableLambda(
            lambda x: search_tool.invoke
        )
    )
    | ChatPromptTemplate.from_messages()
      | llm
)

`RunnablePassthrough.assign` Neng在保持原始输入字段不变的情况下为字典额外添加 `search_results`，实现「输入‑ ‑输出」的一站式流水线。配合 `abatch` 与 `max_concurrency` 参数，我们只需一行代码就Neng实现Zui高五路并发，而不会因为单条失败导致整体中断。

writer_prompt = ChatPromptTemplate.from_messages()
writer_chain = writer_prompt | llm.with_structured_output
# 降级方案：当 JSON 被截断时用普通文本模式继续生成。
writer_chain_plain = writer_prompt | llm

`ReportData` 中除了正文，还包括简短概述和后续问题，让Zui终产出geng具可读性与延展性。Ru果结构化输出因 token 超限报错，我们会捕获异常并回退到 `writer_chain_plain`，确保任务始终有返回值。

class ResearchManager:
   def __init__:
       self.planner = planner_chain
       self.searcher = search_chain
       self.writer   = writer_chain
       self.writer_fallback = writer_chain_plain
   async def run:
       # 1️⃣ 策划搜索计划
       try:
           plan:WebSearchPlan = await self.planner.ainvoke
       except Exception as e:
           print
           # 简单兜底：使用原始 query + 常规补充词构造计划
           plan = WebSearchPlan(searches=)
       # 2️⃣ 并发执行所有搜索任务
       inputs=
       raw_results=await self.searcher.abatch(
           inputs,
           config={"max_concurrency":5},
           return_exceptions=True,
       )
       summaries=
       # 3️⃣ 合成Zui终报告
       payload={"query":query,"research_results":"\
---\
".join}
       try:
           report:ReportData=await self.writer.ainvoke
       except Exception as e:
           print
           report_text=await self.writer_fallback.ainvoke
           report=ReportData(short_summary=f"关于 {query} 的简要报告",
                            follow_up_questions=,
                            content=report_text)
       # 保存到本地文件系统
       filename=f"research_reports/{self._slug}.md"
       os.makedirs, exist_ok=True)
       with open as f:
            f.write
       print
   def _slug:
        return "".join else "_" for c in text)
# 示例调用：
# manager=ResearchManager
# asyncio.run)

uv run test_setup.py   # 若显示 All checks passed，即可继续。

uv run main_langchain.py --query "ChatGPT 与企业知识库融合方案"
# 稍等片刻，你会kan到类似：
# Planning... ✅ 
# Searching... ✅ 
# Writing... ✅ 
# Report saved to research_reports/ChatGPT_与_企业_知识库融合方案.md

cat research_reports/ChatGPT_与_企业_知识库融合方案.md | less -R