第 8 章 · 02 open_deep_research 与实战案例

第 8 章 · 02 open_deep_research 与实战案例

本节摘要:本节攀到 ★★★ 层——把全书机制组装成一个完整系统。主角是 examples/open_deep_research:smolagents 最大的完整示例工程,对 OpenAI Deep Research 的开源复现,在 GAIA 验证集拿到 55% pass@1(原版 67%)。目录含 2 个 notebook、3 个入口脚本(run.py/run_gaia.py/app.py)与 scripts/ 下 8 个辅助脚本;架构是典型两级多智能体:manager CodeAgent(挂 visualizer 视觉问答与文件审查工具)+ search_agent ToolCallingAgent(挂 8 个网页浏览工具)。之后速览五个实战案例:text_to_sql(错误观察驱动的自我纠错)、rag(BM25 检索增强)、async_agent(Starlette+anyio 集成 Web 服务)、plan_customization(human-in-the-loop 人工改计划)、structured_output_tool(MCP 结构化输出),以及 multiple_tools.py 多工具组合。

内容来源:原项目 examples/open_deep_research/(README.md、run.py、run_gaia.py、scripts/)及 examples 下 text_to_sql.py、rag.py、async_agent/、plan_customization/、structured_output_tool.py、multiple_tools.py,精读并套用体系化模板。

⚠️ 注意:open_deep_research 默认用 OpenAI o1(需 tier-3 权限)且依赖 SerpApi/Serper 搜索 key;GAIA 复现还要求申请 smolagents/GAIA-annotated 数据集访问(即时批准)。跑通它不是本节目的,读懂它的分工与组装方式才是。

学习目标

阅读完本节,你应当能够:

1. 画出 open_deep_research 的两级架构:manager 与 search_agent 各挂哪些工具、各自 max_steps 与 planning_interval。
2. 说出 scripts/ 八个脚本按"浏览/审查/评分/改写/基建"的分类。
3. 解释 text_to_sql 为什么天然自我纠错(错误如何回流成观察)。
4. 用 step_callbacks 实现 human-in-the-loop 的计划审核并说清 reset=False 续跑;说明同步 agent 塞进异步 Web 服务的标准姿势(anyio.to_thread)。

一、open_deep_research:GAIA 基准复现

1.1 目录结构与架构

examples/open_deep_research/
├── README.md                    # 55% pass@1 说明与复现步骤
├── run.py                       # 单问入口:python run.py --model-id "o1" "问题"
├── run_gaia.py                  # GAIA 全量评测:--concurrency 32 线程池并发
├── app.py / analysis.ipynb / visual_vs_text_browser.ipynb   # Space 入口与两份分析
└── scripts/
    ├── text_web_browser.py      # SimpleTextBrowser + 6 个浏览工具(567 行)
    ├── text_inspector_tool.py   # 把文件/PDF/YouTube 交给 VLM 审查(124 行)
    ├── visual_qa.py             # visualizer 工具:截图交给视觉模型(189 行)
    ├── mdconvert.py             # 任意格式(html/pdf/xlsx/docx)转 markdown(1002 行)
    ├── reformulator.py          # 终答改写器:基于完整对话重述答案(借自 Autogen)
    ├── gaia_scorer.py / run_agents.py   # GAIA 评分与问题描述加工
    └── cookies.py               # 浏览器会话 cookie 持久化(715 行)
​

核心装配在 run.py:60-111,逐段读:

60   def create_agent(model_id="o1"):
61       model_params = {
62           "model_id": model_id,
63           "custom_role_conversions": custom_role_conversions,
64           "max_completion_tokens": 8192,
65       }
66       if model_id == "o1":
67           model_params["reasoning_effort"] = "high"
68       model = LiteLLMModel(**model_params)
70       text_limit = 100000
71       browser = SimpleTextBrowser(**BROWSER_CONFIG)
72       WEB_TOOLS = [                                # 搜索+浏览+审查共 8 件
73           GoogleSearchTool(provider="serper"),
74           VisitTool(browser), PageUpTool(browser), PageDownTool(browser),
75           FinderTool(browser), FindNextTool(browser), ArchiveSearchTool(browser),
80           TextInspectorTool(model, text_limit),
81       ]
82       text_webbrowser_agent = ToolCallingAgent(
83           model=model,
84           tools=WEB_TOOLS,
85           max_steps=20,
86           verbosity_level=2,
87           planning_interval=4,
88           name="search_agent",
89           description="""A team member that will search the internet to answer your question. ...""",
95           provide_run_summary=True,
96       )
97       text_webbrowser_agent.prompt_templates["managed_agent"]["task"] += """You can navigate to .txt online files. ..."""
101      manager_agent = CodeAgent(
102          model=model,
103          tools=[visualizer, TextInspectorTool(model, text_limit)],
104          max_steps=12,
105          verbosity_level=2,
106          additional_authorized_imports=["*"],
107          planning_interval=4,
108          managed_agents=[text_webbrowser_agent],
109      )
111      return manager_agent
​

这份 50 行装配是全书机制的集大成:

• 两级分工(第 8 章第 1 节):manager 是 CodeAgent、负责推理与计算,additional_authorized_imports=["*"] 放开全部导入做数据处理;search_agent 是 ToolCallingAgent、负责"单时间线"的网页翻找,max_steps 给到 20。
• 工具生态(第 5 章):GoogleSearchTool 内置工具、SimpleTextBrowser 系列是自定义 Tool 子类(共享一个 browser 有状态对象)、TextInspectorTool 甚至把 model 本身当参数传进工具——工具内部再调 VLM 看文件。
• 规划(第 4 章):两级都开 planning_interval=4,每 4 步重新审视事实与计划。
• 提示词定制(第 3 章):L97 直接对 yaml 模板追加内容,教 search_agent 遇到 pdf/YouTube 转用 inspect_file_as_text、信息不足时用 final_answer 反向请求澄清——不改库源码,运行时微调提示词。
• 汇报增强:provide_run_summary=True,manager 能看到搜索过程纪要。

run_gaia.py 是工程化外挂:--concurrency 32 用 ThreadPoolExecutor 并发跑 GAIA 全部题目,reformulator.prepare_response 在最后把 manager 的完整对话历史交给一个改写模型,产出干净的最终答案再交 gaia_scorer 评分——多 agent 跑任务,单模型写报告,各司其职。

💡 阶梯要点:GAIA 是"通用 AI 助手"基准,考的是多跳检索+文件理解+精确计算。open_deep_research 的答卷结构恰好是全书八章的映射:CodeAgent 管计算(第 3 章)、层级派发管检索(第 8 章)、浏览器与文件工具管感知(第 5 章)、规划步管长线任务(第 4 章)、LiteLLM 管模型接入(第 7 章)。读示例工程=把八章知识重新走一遍。

二、text_to_sql:错误观察驱动的自我纠错

examples/text_to_sql.py(80 行)用 SQLAlchemy 建内存 SQLite 表,然后只给 agent 一个工具:

50   @tool
51   def sql_engine(query: str) -> str:
52       """
53       Allows you to perform SQL queries on the table. Returns a string representation of the result.
54       The table is named 'receipts'. Its description is as follows:
55           Columns:
56           - receipt_id: INTEGER
57           - customer_name: VARCHAR(16)
58           - price: FLOAT
59           - tip: FLOAT
...
64       Args:
65           query: The query to perform. This should be correct SQL.
66       """
67       output = ""
68       with engine.connect() as con:
69           rows = con.execute(text(query))
70           for row in rows:
71               output += "\n" + str(row)
72       return output
75   agent = CodeAgent(
76       tools=[sql_engine],
77       model=InferenceClientModel(model_id="meta-llama/Meta-Llama-3.1-8B-Instruct"),
78   )
79   agent.run("Can you give me the name of the client who got the most expensive receipt?")
​

两个设计点:schema 写进 docstring(L54-59)——@tool 装饰器把它原样带进提示词,模型无需另查表结构;自我纠错不需要额外代码——SQL 写错时 con.execute 抛异常,错误与 traceback 作为观察写回 ActionStep(第 2 章的 ReAct 反馈环),下一步模型看到 OperationalError: near "FORM": syntax error 就自己改写 SQL 重试。8B 小模型也能在两三步内收敛到正确查询——纠错能力来自循环结构,而非模型规模。

三、rag.py 与 structured_output_tool.py:检索与结构化

rag.py 用 LangChain 的 BM25Retriever 包一个 RetrieverTool(继承 Tool 基类,inputs 只有一个 query),把 HuggingFace 文档切块检索交给 CodeAgent:max_steps=4 控制成本、stream_outputs=True 实时看思考。它示范了把任意检索器变成 agent 工具的标准三步:构造时注入 docs → forward 里调 retriever → 拼回文档文本。rag_using_chromadb.py 是同一思路的向量库版(ChromaDB 嵌入检索)。

structured_output_tool.py 演示 MCP+结构化输出:内联一段 FastMCP 服务脚本,MCPClient(serverparams, structured_output=True) 桥接进 smolagents,工具返回 pydantic 模型(带 Field 描述的 WeatherInfo),提示词自动渲染成"用 result['field_name'] 直接取字段,无需 print"——第 5 章 output_schema 与第 7 章 schema 生成的接力应用。

四、async_agent:同步 agent 进异步 Web 服务

examples/async_agent/main.py(49 行)全解在此:

26   async def run_agent_in_thread(task: str):
27       agent = get_agent()
28       # The agent's run method is synchronous
29       result = await anyio.to_thread.run_sync(agent.run, task)
30       return result
33   async def run_agent_endpoint(request: Request):
34       data = await request.json()
35       task = data.get("task")
36       if not task:
37           return JSONResponse({"error": 'Missing "task" in request body.'}, status_code=400)
38       try:
39           result = await run_agent_in_thread(task)
40           return JSONResponse({"result": result})
...
49   app = Starlette(debug=True, routes=routes)
​

smolagents 的 run() 是同步阻塞(可能几十秒),直接放进 async 端点会卡死事件循环。标准姿势是 anyio.to_thread.run_sync(agent.run, task)——把阻塞调用扔进线程池,await 其完成。Starlette 只是路由壳,换 FastQuest/FastAPI 同理。每次请求新建 agent(27 行)避免共享状态;若要复用,注意 agent 的 memory 是有状态的,需自己加锁或池化。

五、plan_customization:human-in-the-loop 人工改计划

plan_customization/plan_customization.py(167 行)把第 4 章的 step_callbacks 用成交互闸门:

62   def interrupt_after_plan(memory_step, agent):
67       if isinstance(memory_step, PlanningStep):
68           print("\n🛑 Agent interrupted after plan creation...")
71           display_plan(memory_step.plan)
74           choice = get_user_choice()
76           if choice == 1:  # Approve plan
78               return
81           elif choice == 2:  # Modify plan
83               modified_plan = get_modified_plan(memory_step.plan)
86               memory_step.plan = modified_plan     # 直接改写记忆中的计划
91               return
94           elif choice == 3:  # Cancel
96           agent.interrupt()
...
106      agent = CodeAgent(
107          model=InferenceClientModel(),
108          tools=[DuckDuckGoSearchTool()],
109          planning_interval=5,
110          step_callbacks={PlanningStep: interrupt_after_plan},
111          max_steps=10,
112          verbosity_level=1,
113      )
​

机制链:step_callbacks={PlanningStep: interrupt_after_plan} 只对规划步注册回调(L110 的字典形态是第 4 章 CallbackRegistry 的按类型注册);回调里直接改写 memory_step.plan(L86)——因为后续步骤的提示词从 memory 渲染,改了计划就改了 agent 的后续行为;agent.interrupt()(L96)置 interrupt_switch,主循环下一步抛 AgentError 停机。示例后半段还演示 agent.run(task, reset=False) 保留记忆续跑——人工干预后从断点继续,而非从头再来。这就是 human-in-the-loop 的最小实现:三个 input() 加一个回调。

multiple_tools.py(256 行)则是工具侧的综合练习:天气 API 请求、json 解析、多个 @tool 并存,agent 按任务自行编排调用顺序,可作为第 5 章的收官自测材料。

本节要点回顾

1. open_deep_research = 两级多智能体:manager CodeAgent(visualizer+TextInspector,imports=["*"],max_steps=12)+ search_agent ToolCallingAgent(8 个浏览工具,max_steps=20),双双 planning_interval=4,search_agent 开 provide_run_summary;GAIA 55% pass@1。
2. scripts/ 八脚本五类:浏览(text_web_browser)、感知(text_inspector/visual_qa)、转换(mdconvert)、评测(reformulator/gaia_scorer/run_agents)、基建(cookies)。
3. 运行时提示词定制:agent.prompt_templates["managed_agent"]["task"] += ... 不改库源码即可教子 agent 新规矩。
4. text_to_sql 自我纠错:schema 写进 docstring;SQL 报错的 traceback 经 ReAct 观察回流,模型下步自改——纠错来自循环结构而非模型规模。
5. rag:BM25Retriever 包成 Tool 只需构造注入+forward 调用;structured_output_tool 展示 MCP+pydantic 结构化返回。
6. async_agent:anyio.to_thread.run_sync(agent.run, task) 是同步 agent 进异步服务的标准姿势,每请求新建实例避免状态污染。
7. plan_customization:step_callbacks 按 PlanningStep 注册回调,回调里改 memory_step.plan 即改后续行为;interrupt() 停机、reset=False 断点续跑,构成 human-in-the-loop 三件套。

下一节:全书最后一节——CLI 交互式向导、GradioUI 流式界面、多模态 AgentImage/AgentAudio、视觉网页 agent,然后沿能力阶梯把八章串成一条线,做全书回顾。