13 - 深度研搜：主智能体搭建与异步执行

本章课程目标：

  • 理解主智能体和三个子智能体的分工关系。
  • 明确为什么主智能体需要文件类工具：读取上传文件、生成 Markdown、转换 PDF。
  • 完成 main_agent.py 的核心组装：模型、提示词、工具、子智能体、检查点。
  • 读懂主智能体系统提示词中的任务边界和执行顺序要求。
  • 理解为什么主智能体执行要选择 astream() 异步流式方式。
  • 拆解 run_deep_agent()：会话目录、上传文件、ContextVar、流式执行、事件推送和上下文清理。
  • 看清前端、FastAPI、主智能体、WebSocket、monitor 之间的协作关系，为接口层开发做准备。

学习建议： 这一章分两段看。前半先看主智能体的“创建态”：它是调度中心，负责理解任务、分派任务、汇总结果和生成文件；真正查网络、查数据库、查知识库的是子智能体和工具。后半再看主智能体的“运行态”：前端请求如何触发它，它如何用 astream() 流式执行，执行过程又如何通过 monitor 和 WebSocket 回到前端。把这两段连起来，下一章看 FastAPI 接口层就不会乱。

对应代码分支： 13-deepsearch-main-agent

前面三章已经分别完成了三个专家助手：

子智能体 负责内容 底层工具
网络搜索助手 查询互联网公开信息 Tavily
数据库查询助手 查询企业内部结构化数据 MySQL
RAGFlow 助手 查询企业内部非结构化知识库 RAGFlow API

现在要把它们交给一个“总负责人”统一调度，这个总负责人就是主智能体。

可以先用一句话理解本章：主智能体 = 模型 + 主提示词 + 文件工具 + 三个子智能体 + 会话检查点 + 异步运行入口

本章按这条路线展开：

1
2
3
4
5
先区分主智能体和子智能体
  -> 再接入主智能体自己的文件工具
  -> 组装 main_agent
  -> 理解 run_deep_agent 的异步执行链路
  -> 最后做本地验证

主智能体接到用户问题后，会先判断任务需要哪些信息来源，再调用合适的子智能体。等信息拿回来以后，如果用户要求生成文件，主智能体再调用自己的文件工具，把最终结果写成 Markdown 或 PDF。

1、主智能体和子智能体有什么不同

1.1 子智能体只负责某一类专业任务

前面写子智能体时，每个子智能体的结构都比较固定：子智能体 = name + description + system_prompt + tools

比如网络搜索助手，它仅需知道：

  • 自己什么时候应该被调用；
  • 自己应该怎么搜索；
  • 自己可以使用 internet_search 工具。

数据库助手也类似，它只负责按数据库查询流程做事：先看表、再看表结构和样例数据、最后执行 SQL。

这些子智能体都是“专家”，但它们不是最终交付者。

1.2 主智能体负责统筹和交付

主智能体不只是一个字典配置，它要通过 create_deep_agent() 创建出来，并且需要额外注册子智能体。

主智能体要负责几件事：

主智能体职责 说明
理解用户任务 判断用户到底要查资料、查数据、读文档，还是生成文件
调度子智能体 根据任务边界调用网络、数据库、RAGFlow 助手
汇总信息 把不同助手返回的信息整理成最终答案
文件交付 根据用户要求生成 Markdown 或 PDF
维持会话上下文 通过 thread_id 和 checkpointer 支持同一会话持续执行

所以主智能体比子智能体多了一个关键能力：它能持有多个子智能体，并把任务分派给它们。

2、主智能体为什么要有三个文件工具

2.1 三个工具分别解决什么问题

本项目的最终目标不是只回答一句话，而是完成“深度研搜”任务。用户可能会提出这样的需求：从网络查询机器人相关信息，并生成 Markdown 和 PDF 文件。

或者：读取我上传的文档，结合数据库中的药品信息，整理成一份分析报告。

这时主智能体就需要三个自己的工具：

工具 文件 作用
read_file_content app/tools/upload_file_read_tool.py 读取用户上传的附件内容
generate_markdown app/tools/markdown_tools.py 把最终内容写入 .md 文件
convert_md_to_pdf app/tools/pdf_tools.py 把 Markdown 转成 PDF

这三个工具都不是网络搜索、数据库查询、知识库问答，所以不应该放到某个子智能体里。它们属于最终交付环节，因此由主智能体直接掌握。

2.2 为什么读取上传文件也属于主智能体

上传文件是用户本次任务的补充上下文。比如用户上传了一份错误日志、产品说明、报告草稿，主智能体必须先读懂这个文件，才能决定下一步要不要查网络、查数据库或生成文档。

它不属于 RAGFlow，因为这是用户临时上传的会话文件，不是企业长期知识库。

它也不属于数据库助手，因为上传文件不是结构化业务表。

所以读取上传文件的工具应该交给主智能体。

2.3 文件工具都依赖当前会话目录

文件工具有一个共同点：它们不能随便读写系统任意位置，而是必须读写当前会话的工作目录。

这个目录会在本章后半的 run_deep_agent() 中创建：output/session_{session_id}

工具内部通过 get_session_context() 拿到当前会话目录，再通过 resolve_path() 把模型传入的相对路径解析成真实路径。

这背后有一个重要设计：

1
2
模型只负责说“我要生成 report.md”
系统负责把它解析成当前会话目录下的真实文件路径

这样可以减少模型处理长路径时出错，也能避免多个用户的文件串到一起。

3、工具一：读取上传文件 read_file_content

项目对应文件路径：deepsearch-agents/app/tools/upload_file_read_tool.py

read_file_content 负责把用户本次上传的附件读出来，交给主智能体继续分析。它支持常见格式：

文件类型 处理方式
.md / .txt 直接按 UTF-8 文本读取
.docx 使用 python-docx 读取段落
.pdf 使用 pypdf 提取页面文本
.xlsx / .xls 使用 pandas 读取，并输出行数、列名、前 5 行和统计描述
其他文本文件 尝试按 UTF-8 文本读取

核心代码可以这样看：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
@tool
def read_file_content(
    filename: Annotated[
        str, "要读取的文件名或路径（支持 .md, .docx, .pdf, .xlsx, .xls）"
    ],
    instruction: Annotated[
        str, "对提取内容的具体指令（例如：'提取摘要', '统计数据'）"
    ] = "提取全部内容",
) -> str:
    """
    读取当前会话目录中的指定文件内容

    对于 Excel 文件，会自动提供数据统计信息（head 和 describe）。
    :param filename: 文件名或相对路径，通常由主智能体从上传文件列表中选择
    :param instruction: 模型传入的读取意图，用于监控展示，不改变底层解析逻辑
    :return: 文件文本内容、表格摘要，或中文错误提示
    """
    monitor.report_tool(
        "文件内容读取工具", {"filename": filename, "instruction": instruction}
    )

    # 解析路径时优先约束在当前 session_dir 内，避免模型传入绝对路径导致越界读取
    session_dir = get_session_context()
    file_path = Path(resolve_path(filename, session_dir))

    if not file_path.exists():
        return f"错误：文件 '{filename}' 不存在 (解析路径: {file_path})。"

    # 根据文件后缀选择解析方式；未知后缀会先按 UTF-8 文本兜底读取
    ext = file_path.suffix.lower()

读这段代码时关注三点：

  1. @tool 会把函数注册成主智能体可以调用的工具。
  2. monitor.report_tool(...) 会把“正在读取文件”这类进度推给前端。
  3. resolve_path(filename, session_dir) 会把模型给出的文件名限制到当前会话目录。

如果用户上传了 开篇.txt，提示词会告诉模型直接传入文件名：read_file_content(filename="开篇.txt")

工具会自动拼到当前会话目录下读取，不需要模型自己拼绝对路径。

在项目根目录，执行：uv run python -m app.tools.upload_file_read_tool

脚本里的测试入口会把 get_session_context() 临时固定为项目根目录下的 examples/test_docs，然后读取测试文件。

上传文件读取工具本地验证：读取测试目录中的 Markdown、Word 和 PDF 文件内容

4、工具二：生成 Markdown 文件 generate_markdown

项目对应文件路径：deepsearch-agents/app/tools/markdown_tools.py

generate_markdown 负责把模型整理好的内容写成 Markdown 文件。

核心参数有三个：

参数 说明
content 要写入 Markdown 的正文
filename 文件名，可以带 .md，也可以不带
path 保存路径，通常由提示词约束在当前工作目录下

核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
@tool
def generate_markdown(
    content: Annotated[str, "要写入Markdown文档的文本内容"],
    filename: Annotated[str, "Markdown文档的文件名（不包含扩展名或包含.md）"],
    path: Annotated[str, "文件保存的绝对路径"] = "",
):
    """
    根据提供的文本内容生成 Markdown 文件

    :param content: 要写入 Markdown 文档的完整文本
    :param filename: 输出文件名，缺少 .md 后缀时会自动补全
    :param path: 可选保存路径；通常由运行时工作目录指令约束为相对路径
    :return: 文件生成结果说明
    """
    print(f"[MarkdownTool] 输入保存路径: {path or '当前会话目录'}")
    monitor.report_tool("Markdown文档生成工具", {"写入的文本内容": content})

    if not filename.endswith(".md"):
        filename += ".md"

    # session_dir 由 run_deep_agent 写入 ContextVar，保证文件写入当前会话工作目录
    session_dir = get_session_context()

    # 先把模型传入的 path/filename 合成一个逻辑路径，再交给 resolve_path 做统一清洗
    if path and path != ".":
        full_input_path = str(Path(path) / filename)
    else:
        full_input_path = filename

    full_path_str = resolve_path(full_input_path, session_dir)
    file_path = Path(full_path_str)
    parent_dir = file_path.parent

    # 允许模型指定 session_dir 下的子目录；不存在时自动创建
    if not parent_dir.exists():
        parent_dir.mkdir(parents=True, exist_ok=True)

    file_path.write_text(content, encoding="utf-8")
    return f"Markdown文件 '{file_path}' 已成功生成并保存。"

这段代码的重点不是写文件本身，而是路径处理：

1
2
3
模型给出 path + filename
  -> 工具用 resolve_path 解析
  -> 最终写入当前会话 output/session_xxx 下

这样前端后续查询文件列表时，仅需看当前会话目录，就能看到生成的 .md 文件。

在项目根目录，执行：uv run python -m app.tools.markdown_tools

测试入口会创建类似 examples/test_docs/sub_dir/测试文件.md 这样的文件。

如果看到“已成功生成”，说明 Markdown 文件写入链路没问题。

5、工具三：Markdown 转 PDF convert_md_to_pdf

项目对应文件路径：deepsearch-agents/app/tools/pdf_tools.py

convert_md_to_pdf 负责把 Markdown 文件转换成 PDF。它不重新生成内容，只基于已经写好的 .md 文件做格式转换。

核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
@tool
def convert_md_to_pdf(
    md_filename: Annotated[str, "要转换的Markdown文档路径（包含.md后缀）"],
    pdf_filename: Annotated[
        Optional[str], "输出的PDF文件路径（可选，默认与源文件同名）"
    ] = None,
) -> str:
    """
    将当前会话目录中的 Markdown 文档转换为 PDF

    :param md_filename: Markdown 文件名或相对路径，缺少后缀时会自动补为 .md
    :param pdf_filename: 可选 PDF 输出文件名；不传时与 Markdown 同名
    :return: 转换结果说明
    """
    monitor.report_tool("Markdown转PDF工具")

    try:
        # 输入路径必须先落到当前会话目录，避免模型传入任意系统路径
        session_dir = get_session_context()
        md_path = Path(md_filename).with_suffix(".md")
        md_abs_path = Path(resolve_path(str(md_path), session_dir))

        if not md_abs_path.exists():
            return f"错误：文件不存在 {md_abs_path}"

        # 未指定 PDF 文件名时，默认与源 Markdown 同目录同名
        if pdf_filename:
            pdf_path = Path(pdf_filename).with_suffix(".pdf")
            pdf_abs_path = Path(resolve_path(str(pdf_path), session_dir))
        else:
            pdf_abs_path = md_abs_path.with_suffix(".pdf")

        # PDF 版式、中文字体和 Markdown 解析细节都封装在底层转换模块中
        return convert_md_to_pdf_via_word(md_abs_path, pdf_abs_path)

    except Exception as e:
        logging.error(f"转换失败: {e}", exc_info=True)
        return f"转换失败: {str(e)}"

它的工作流程是：

1
2
3
4
确认 Markdown 文件存在
  -> 确定 PDF 输出路径
  -> 调用 convert_md_to_pdf_via_word
  -> 返回转换结果

在 app/utils/word_converter.py 中，底层会使用 ReportLab 生成 PDF：

1
2
3
4
5
6
7
8
9
10
11
12
doc = SimpleDocTemplate(
    str(pdf_abs_path),
    pagesize=A4,
    leftMargin=2 * cm,
    rightMargin=2 * cm,
    topMargin=2 * cm,
    bottomMargin=2 * cm,
    title=md_abs_path.stem,
)
styles = _build_styles()
story = _markdown_to_story(md_content, styles)
doc.build(story)

在项目根目录，执行：uv run python -m app.tools.pdf_tools

测试入口会先创建一份示例 Markdown，再调用 convert_md_to_pdf 生成同名 PDF。这里主要验证两件事：Markdown 路径能被解析到当前会话目录下，ReportLab 转换链路也能正常生成中文 PDF。

6、创建 main_agent.py

项目对应文件路径：deepsearch-agents/app/agent/main_agent.py

主智能体文件先做导入：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import shutil
from pathlib import Path

from deepagents import create_deep_agent
from langgraph.checkpoint.memory import InMemorySaver

from app.agent.llm import model
from app.agent.prompts import main_agent_content
from app.agent.subagents.database_query_agent import database_query_agent
from app.agent.subagents.knowledge_base_agent import knowledge_base_agent
from app.agent.subagents.network_search_agent import network_search_agent
from app.api.context import (
    reset_session_context,
    set_session_context,
    set_thread_context,
)
from app.api.monitor import monitor

# 文件类工具由主智能体直接掌握，负责读取上传附件和生成最终交付文档
from app.tools.markdown_tools import generate_markdown
from app.tools.pdf_tools import convert_md_to_pdf
from app.tools.upload_file_read_tool import read_file_content

这几组导入可以分成五类：

导入内容 作用
shutil / Path 创建会话目录、复制上传文件
三个子智能体 让主智能体可以委派专业任务
三个文件工具 让主智能体可以读取上传文件、生成 Markdown、转换 PDF
create_deep_agent / InMemorySaver 创建 DeepAgents 主智能体并启用会话检查点
model / main_agent_content / 上下文 注入模型、系统提示词、运行时会话上下文和监控上报

然后创建主智能体：

1
2
3
4
5
6
7
8
9
10
11
# 主智能体是调度中心：
# 1. tools 只放最终交付相关的文件工具
# 2. subagents 放网络、数据库、RAGFlow 三类信息获取助手
# 3. checkpointer 通过 thread_id 保存同一会话中的执行上下文
main_agent = create_deep_agent(
    model=model,
    system_prompt=main_agent_content["system_prompt"],
    tools=[generate_markdown, convert_md_to_pdf, read_file_content],
    checkpointer=InMemorySaver(),
    subagents=[database_query_agent, network_search_agent, knowledge_base_agent],
)

这就是本章关键的代码。

6.1 create_deep_agent 参数说明

参数 作用 本项目配置
model 主智能体使用的大模型 第 9 章配置好的 model
system_prompt 主智能体行为约束 prompts.yml 中的 main_agent.system_prompt
tools 主智能体自己的工具 文件读取、Markdown、PDF
checkpointer 会话状态检查点 InMemorySaver()
subagents 可调度的子智能体 数据库、网络搜索、RAGFlow

这里的 checkpointer=InMemorySaver() 是为了配合后续的 thread_id。同一个前端会话会带着同一个 thread_id 来调用，主智能体就可以在这个会话维度上维持状态。

6.2 子智能体注册顺序

当前代码里的注册顺序是：

1
2
3
4
5
subagents=[
    database_query_agent,
    network_search_agent,
    knowledge_base_agent,
]

顺序不是最核心的问题，关键是每个子智能体的 description 写得足够清楚。主智能体会根据用户问题和子智能体描述来决定调用谁。

例如：

用户需求 更可能调用
查询某个商品库存 数据库查询助手
查询机器人行业公开信息 网络搜索助手
查询企业内部制度或知识库文档 RAGFlow 助手
查询资料并生成 PDF 先调用相关子智能体，再由主智能体生成 Markdown 和 PDF

7、主智能体提示词要约束什么

项目对应文件路径：deepsearch-agents/app/prompt/prompts.yml

主智能体的提示词比较长，但核心不是“写得长”，而是要约束清楚几个关键点。

7.1 角色定位

提示词开头把主智能体定位成团队负责人：

1
2
3
4
5
6
7
main_agent:
  system_prompt: |
    你是金融与电商研究团队负责人，负责协调三个专家助手完成复杂任务
    你的团队成员如下：
      1. **网络搜索助手**
      2. **数据库查询助手**
      3. **RAGFlow助手**

这里要和课程项目背景保持一致。当前主提示词把它设定成“金融与电商研究团队负责人”，文章后面的例子也沿用这个设定。不要被行业名称带偏，本章真正要抓住的是：主智能体是团队负责人，不是某一个具体工具。

同时要留意提示词和代码能力是否一致。当前 main_agent 注册的文件工具只有 generate_markdown、convert_md_to_pdf 和 read_file_content。如果提示词里还保留“Word 文件生成”或“交给文件生成助手”之类的说法，读者容易误以为项目里还有一个单独的文件生成子智能体。更稳妥的表述是：Markdown 由主智能体调用 generate_markdown 生成；PDF 先生成 Markdown，再调用 convert_md_to_pdf 转换。

7.2 信息来源分工

提示词里要告诉主智能体，三类信息分别找谁：

信息类型 调用对象
背景知识、外部知识、最新公开资料 网络搜索助手
企业内部非公开文档知识 RAGFlow 助手
企业内部商品、库存、销售等结构化数据 数据库查询助手

如果问题边界不清楚，可以让主智能体尝试多种方式获取信息。

这一点很重要。因为真实用户不会按你的系统边界来提问，用户只会说：帮我整理一下某个产品的市场信息和内部库存情况，生成 PDF。

这个任务同时涉及外部资料、内部数据库和文件生成。主智能体必须先拆任务，而不是直接生成文件。

7.3 文件生成顺序

提示词中关键的是执行顺序：

1
2
3
4
1. 必须先调用子智能体获取信息。
2. 绝不允许在获取信息之前调用 generate_markdown。
3. 严禁使用“等待子任务完成”之类的占位符内容生成文件。
4. 如果需要先搜索再生成，请分两步进行。

为什么要把顺序写清楚？

因为模型在复杂任务里可能提前调用 generate_markdown，写入“等待搜索结果补充”这类占位内容。这样生成出来的文件看起来存在，但没有价值。

所以主智能体必须遵守：

1
2
3
4
先拿信息
  -> 再组织内容
  -> 再生成 Markdown
  -> 如用户要求 PDF，再转换 PDF

到这里，main_agent 对象已经创建好了。但对象创建好并不等于用户已经能用。接下来还要解决三个运行时问题：

1
2
3
前端用户点发送后，怎么触发 main_agent？
main_agent 执行过程中，怎么把工具调用、子智能体调用、最终结果推给前端？
多个用户同时请求时，怎么保证消息不串台？

这就是 run_deep_agent() 要负责的部分。

8、运行时先看全链路

上半部分解决的是“主智能体怎么创建”。从这里开始，看它在 Web 服务里怎么跑。

这一节先把 astream()、HTTP、WebSocket、session_id 和 ContextVar 放到一张图里。第 15 章会正式写 FastAPI 接口，本章只讲主智能体运行时必须提前理解的几件事。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
flowchart TD
    Frontend[前端页面] --> WS[先建立 WebSocket /ws/thread_id]
    Frontend --> HTTP[再提交 HTTP POST /api/task]
    HTTP --> Run[run_deep_agent]
    Run --> Dir[创建 session_dir]
    Run --> Ctx[写入 ContextVar]
    Run --> Agent[main_agent.astream]
    Agent --> Subagents[调用三个子智能体]
    Agent --> FileTools[调用文件工具]
    Subagents --> Monitor[monitor 上报事件]
    FileTools --> Monitor
    Agent --> Monitor
    Monitor --> WS
    WS --> Frontend

按一条线读：HTTP 只负责启动任务，run_deep_agent() 负责准备会话环境，main_agent.astream() 负责异步执行，执行中的进度和最终结果再通过 monitor -> WebSocket 回到前端。

8.1 为什么主智能体要选 astream

DeepAgents 常见执行方式有三种：

执行方式 特点 本项目是否使用
invoke() 同步一次性返回 不适合多客户端长任务
stream() 同步流式返回 仍然会阻塞当前同步调用链
astream() 异步流式返回 本项目使用

本项目是一个后端服务，而不是一个本地脚本。前端可能有多个会话同时提问：

1
2
3
用户 A 正在查网络并生成 PDF
用户 B 正在查数据库
用户 C 正在读取上传文件

如果用同步方式，一个任务执行时间长，其他请求就容易排队等待。使用 astream() 可以把主智能体执行变成异步流式任务，让 FastAPI 服务更适合多客户端并发。

8.2 完整前后端通信流程

先看整体链路：

1
2
3
4
5
6
7
8
9
10
11
12
前端页面打开
  -> 建立 WebSocket: /ws/{thread_id}
  -> 后端缓存 thread_id -> WebSocket

用户点击发送
  -> HTTP POST /api/task
  -> FastAPI 创建后台任务
  -> run_deep_agent(query, thread_id)
  -> main_agent.astream(...)
  -> 执行过程中调用 monitor
  -> monitor 根据 thread_id 找到 WebSocket
  -> WebSocket 推送进度和结果给前端

这条链路里有两个入口：

入口 作用 调用时机
/ws/{thread_id} 建立长连接，缓存当前会话的 WebSocket 页面打开时先调用
/api/task 启动一次智能体任务 用户点击发送时调用

注意顺序：通常先建立 WebSocket，再发起任务。

如果没有先建立 WebSocket，后端执行过程中就不知道把进度推给哪个前端。

8.3 为什么 HTTP 接口不能直接返回最终答案

用户发起任务时会调用 HTTP 接口，例如：POST /api/task

直觉上，好像可以直接在 HTTP response 中返回最终答案。但本项目不这么做。

原因是主智能体任务可能很长：

  • 先调用网络搜索助手；
  • 再调用数据库助手；
  • 再调用 RAGFlow 助手；
  • 再生成 Markdown；
  • 再转换 PDF；
  • 中间还要不断把状态展示给前端。

如果 HTTP 请求一直等最终结果，前端会长时间卡住，后端连接也会变得脆弱。

所以 /api/task 只做一件事：告诉前端：任务已经开始

真正的执行进度和最终结果，通过 WebSocket 推送。

这也是为什么接口响应通常是：

1
2
3
4
{
  "status": "started",
  "thread_id": "custom-session-id-123"
}

而不是直接返回智能体最终答案。

8.4 WebSocket 在这里负责什么

WebSocket 负责实时推送主智能体执行过程中的事件。

这些事件大致有五类：

事件 触发时机 前端展示
session_created 当前会话工作目录创建完成 前端记录文件目录，后续查询文件列表
assistant_call 主智能体准备调用某个子智能体 展示“正在调用网络搜索助手”等
tool_start 某个工具开始执行 展示工具名和参数
task_result 主智能体返回最终答案 展示 AI 回复
error 执行过程中出现异常 展示错误提示

在代码里，业务层不用直接操作 WebSocket。业务层只调用：

1
2
3
4
monitor.report_tool(...)
monitor.report_assistant(...)
monitor.report_task_result(...)
monitor.report_session_dir(...)

然后 monitor 再通过 ConnectionManager 找到对应的 WebSocket，把消息推给前端。

这样业务代码就不用关心 WebSocket 连接细节：

1
2
3
4
5
工具 / 主智能体
  -> monitor
  -> ConnectionManager
  -> WebSocket
  -> 前端

8.5 session_id 如何避免多客户端串台

假设现在有三个前端会话：

1
2
3
session_A -> websocket_A
session_B -> websocket_B
session_C -> websocket_C

后端会用一个字典保存它们：

1
2
3
4
5
active_connections = {
    "session_A": websocket_A,
    "session_B": websocket_B,
    "session_C": websocket_C,
}

当 session_A 的任务执行到工具调用时，后端必须把消息发给 websocket_A，不能发给 websocket_B。

所以每次任务请求都要带上当前会话的 session_id 或 thread_id。

这个 ID 在项目里有三个作用：

作用 说明
找 WebSocket monitor 根据 thread_id 找到当前前端连接
建工作目录 每个会话创建 output/session_{thread_id}，文件互不混淆
取检查点 InMemorySaver() 按 thread_id 维持同一会话状态

这就是多客户端系统的基本设计：前端每次请求都带同一个会话标识，后端根据标识找到该会话对应的资源。

8.6 ContextVar：把会话信息传给深层工具

项目对应文件路径：deepsearch-agents/app/api/context.py

这里定义了两个上下文变量：

1
2
3
4
5
6
7
8
9
10
11
12
13
from contextvars import ContextVar, Token
from typing import Optional

# ContextVar 是协程级上下文变量，适合 FastAPI 这类异步 Web 服务
# 它可以避免多个并发请求共用全局变量时出现 thread_id 或 session_dir 串台
_session_dir_ctx: ContextVar[Optional[str]] = ContextVar(
    "session_dir",
    default=None,
)
_thread_id_ctx: ContextVar[Optional[str]] = ContextVar(
    "thread_id",
    default=None,
)

它们分别保存：

ContextVar 保存内容 谁会使用
_session_dir_ctx 当前会话的工作目录 文件读取、Markdown、PDF 工具
_thread_id_ctx 当前会话 ID monitor 推送 WebSocket 消息

设置和读取方法如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def set_session_context(path: str) -> Token[Optional[str]]:
    """设置当前请求链路的会话目录，返回 reset 时需要使用的 token"""
    return _session_dir_ctx.set(path)

def get_session_context() -> Optional[str]:
    """获取当前请求链路的会话目录"""
    return _session_dir_ctx.get()

def set_thread_context(thread_id: str) -> Token[Optional[str]]:
    """设置当前请求链路的线程 ID"""
    return _thread_id_ctx.set(thread_id)

def get_thread_context() -> Optional[str]:
    """获取当前请求链路的线程 ID"""
    return _thread_id_ctx.get()

def reset_session_context(
    session_token: Token[Optional[str]],
    thread_token: Optional[Token[Optional[str]]] = None,
) -> None:
    """恢复请求上下文，避免本次任务信息残留到后续请求"""
    _session_dir_ctx.reset(session_token)
    if thread_token is not None:
        _thread_id_ctx.reset(thread_token)

为什么不用普通全局变量？

因为 FastAPI 是异步并发服务。多个用户的请求可能在同一个线程中交替执行，如果用全局变量，用户 A 的 session_id 可能被用户 B 覆盖，导致消息或文件串台。

ContextVar 是协程级上下文变量，更适合 asyncio 场景。只要在同一个任务调用链里，深层工具也能拿到当前会话自己的值。

9、run_deep_agent 的七步拆解

项目对应文件路径：deepsearch-agents/app/agent/main_agent.py

主智能体的执行方法是：

1
2
3
4
5
6
7
8
9
async def run_deep_agent(task_query, session_id):
    """
    异步流式执行主智能体

    API 层会为每次任务传入用户问题和 session_id。本函数负责准备会话目录、
    复制上传文件、写入 ContextVar，并在流式执行过程中把关键事件上报给前端。
    :param task_query: 前端提交的原始任务问题
    :param session_id: 当前任务 ID，同时用于 thread_id、输出目录和 WebSocket 定向推送
    """

它做的事情可以拆成七步：

1
2
3
4
5
6
7
1. 创建当前会话的 output/session_{session_id} 目录
2. 检查 updated/session_{session_id} 下是否有上传文件
3. 如果有上传文件，复制到 output/session_{session_id}
4. 把 session_dir 和 session_id 写入 ContextVar
5. 拼接工作目录提示词，调用 main_agent.astream(...)
6. 解析流式 chunk，向前端推送子智能体调用和最终结果
7. finally 中重置 ContextVar

9.1 第一步：创建当前会话输出目录

代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
# 当前文件位于 app/agent/main_agent.py，parents[1] 即 app 目录
project_root_path = Path(__file__).parents[1].resolve()

# 每个会话独立使用 output/session_{session_id}，避免不同用户的产物互相覆盖
session_dir = project_root_path / "output" / f"session_{session_id}"
session_dir.mkdir(parents=True, exist_ok=True)

# 前端和工具使用绝对路径；提示词里只给模型相对路径
session_dir_str = str(session_dir).replace("\\", "/")
relative_session_dir_str = str(session_dir.relative_to(project_root_path)).replace(
    "\\", "/"
)

这里生成了两个路径：

路径 示例 给谁用
session_dir_str /.../app/output/session_xxx 给前端和工具使用
relative_session_dir_str output/session_xxx 给大模型提示词使用

为什么要有相对路径？

因为大模型不擅长处理很长的本机绝对路径。我们告诉它：工作目录: output/session_xxx

再通过工具内部的 resolve_path() 去拼真实绝对路径，这样更稳定。

9.2 第二步：处理上传文件

上传文件先进入：updated/session_{session_id}

执行任务时，代码会把它们复制到：output/session_{session_id}

核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# 上传文件先落在 updated/session_{session_id}，执行前复制到本次 output 工作目录
# 这样读文件工具和生成文件工具都只需要围绕同一个 session_dir 工作
updated_dir_path = project_root_path / "updated" / f"session_{session_id}"
updated_info_prompt = ""

if updated_dir_path.exists():
    files = [f.name for f in updated_dir_path.iterdir() if f.is_file()]

    if files:
        for filename in files:
            # copy2 会保留上传文件的修改时间、权限等元数据，便于后续排查文件来源
            shutil.copy2(updated_dir_path / filename, session_dir / filename)

        # 把上传文件列表注入用户消息，提醒模型先调用 read_file_content 获取附件内容
        updated_info_prompt = (
            "\n    [已上传文件] 已加载到工作目录:\n"
            + "\n".join([f"    - {f}" for f in files])
            + "\n    请优先使用工具（read_file_content）读取并参考这些文件。"
        )

为什么要复制一份？

因为前端后续展示文件列表时，仅需读取一个目录：output/session_{session_id}

这个目录里既有用户上传文件，也有模型生成文件。前端不用分别查 updated 和 output 两个地方。

同时，提示词会告诉模型：

1
2
3
[已上传文件] 已加载到工作目录:
- 开篇.txt
请优先使用 read_file_content 读取并参考这些文件。

这样用户上传文件就变成了本次任务的补充上下文。

9.3 第三步：绑定 ContextVar 并推送会话目录

准备好会话目录后，要把当前任务的上下文保存起来：

1
2
3
4
5
session_dir_token = set_session_context(session_dir_str)
session_id_token = set_thread_context(session_id)

# 前端拿到工作目录后，可以展示本次任务生成的 Markdown/PDF 等产物
monitor.report_session_dir(session_dir_str)

这三行分别做三件事：

代码 作用
set_session_context(session_dir_str) 让文件工具知道当前会话目录
set_thread_context(session_id) 让 monitor 知道当前要发给哪个 WebSocket
monitor.report_session_dir(session_dir_str) 把当前会话文件目录推给前端

前端拿到 session_created 事件后，就能记录这个目录。后续查询文件列表时，会把这个目录传给 /api/files。

9.4 第四步：构建工作目录提示词

主智能体真正执行前，会把工作目录约束拼到用户问题后面：

1
2
3
4
5
6
7
8
9
10
11
path_instruction = f"""
【工作环境指令】
工作目录: {relative_session_dir_str}
{updated_info_prompt}

规则：
1. 新生成文件必须保存到工作目录：'{relative_session_dir_str}/filename'
2. 读取已上传的文件时，请直接将文件名（例如：'开篇.txt'）作为 filename 参数传入（read_file_content）读取工具，不要带上任何目录前缀。
3. 使用相对路径，禁止使用绝对路径
4. 若存在上传文件，请先分析内容
"""

这段提示词解决两个问题：

  1. 生成文件要生成到哪里；
  2. 读取上传文件时应该怎么传参。

尤其是第 2 条很关键。不要让模型把上传文件路径写成 output/session_xxx/开篇.txt，而是让它只传 开篇.txt。

因为路径拼接应该由工具处理，不应该交给模型猜。

9.5 第五步：使用 main_agent.astream 执行

主智能体执行代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
# checkpointer 依赖 thread_id 区分会话记忆；同一 session_id 会复用同一条执行上下文
config = {
    "configurable": {
        "thread_id": session_id
    }
}

# 工作环境指令是运行时动态补充的，约束模型只在当前会话目录读写文件
async for chunk in main_agent.astream(
    {"messages": [{"role": "user", "content": task_query + path_instruction}]},
    config=config,
):
    ...

这里有两个重点。

第一，thread_id 要放进 configurable：

1
config = {"configurable": {"thread_id": session_id}}

它配合前面配置的 InMemorySaver()，让同一个会话拥有自己的检查点。

第二，传给模型的用户内容不是单纯的 task_query，而是：

1
task_query + path_instruction

也就是说，用户问题后面会追加工作目录和文件规则。模型看见完整指令后，才知道该把文件生成到哪个目录、该如何读取上传文件。

9.6 第六步：解析流式 chunk

main_agent.astream(...) 会不断吐出 chunk。每个 chunk 大致长这样：

1
2
3
4
5
{
    "model": {
        "messages": [...]
    }
}

代码会遍历 chunk 中每个节点的 state：

1
2
3
4
5
6
7
8
9
async for chunk in main_agent.astream(...):
    # chunk 形如 {"model": {"messages": [...]}}，这里主要关心模型最新消息
    for node_name, state in chunk.items():
        if not state or "messages" not in state:
            continue

        messages = state["messages"]
        if messages and isinstance(messages, list):
            last_msg = messages[-1]

然后根据 last_msg 判断当前发生了什么。

1
2
3
4
5
6
7
8
9
10
11
flowchart TD
    Chunk["收到一个 chunk"] --> HasMessages{state 中是否有 messages}
    HasMessages -- 否 --> Skip["跳过"]
    HasMessages -- 是 --> Last["取最后一条消息 last_msg"]
    Last --> Calls{是否有 tool_calls}
    Calls -- 是 --> Task{工具名是否为 task}
    Task -- 是 --> Assistant["monitor.report_assistant：上报子智能体调用"]
    Task -- 否 --> Tool["等待 tools 节点或工具内部上报"]
    Calls -- 否 --> Content{是否有 content}
    Content -- 是 --> Result["monitor.report_task_result：上报最终结果"]
    Content -- 否 --> Skip

这张图和第 2 章的 chunk 状态识别是一脉相承的，只是这里多了前端事件上报：识别到子智能体，就推 assistant_call；识别到最终内容，就推 task_result。

9.6.1 情况一：调用子智能体

如果最后一条消息包含 tool_calls，并且工具名是 task，说明主智能体正在调用某个子智能体：

1
2
3
4
5
6
7
8
9
10
if node_name == "model":
    if last_msg.tool_calls:
        # DeepAgents 调用子智能体时，本质上会产生名为 task 的工具调用
        for tool_call in last_msg.tool_calls:
            if tool_call["name"] == "task":
                # 子智能体调用单独上报，前端可以展示“正在调用哪个专家助手”
                monitor.report_assistant(
                    tool_call["args"]["subagent_type"],
                    {"description": tool_call["args"]["description"]},
                )

DeepAgents 内部会用一个特殊的 task 工具来委派子智能体。它的参数里通常会有：

字段 说明
subagent_type 要调用哪个子智能体
description 交给子智能体的任务描述

前端收到 assistant_call 事件后，就可以展示：正在调用助手：网络搜索助手

9.6.2 情况二：主智能体返回最终结果

如果没有工具调用，但最后一条消息有 content，说明主智能体给出了最终答案：

1
2
3
4
elif last_msg.content:
    # 模型没有继续调用工具时，最新文本内容就是本轮可反馈给前端的结果
    print(f"主智能体执行结果，最终结果：{last_msg.content[:100]}")
    monitor.report_task_result(last_msg.content)

这时通过 monitor.report_task_result(...) 推送给前端。

前端收到 task_result 后，就可以把最终回复显示在对话框中。

9.6.3 工具调用进度在哪里推送

这里还有一个容易混的点：普通工具调用并不是都在 run_deep_agent() 里推送。

比如 generate_markdown、internet_search、execute_sql_query 这些工具，工具函数内部已经写了：

1
monitor.report_tool(...)

所以工具级进度在工具内部上报。run_deep_agent() 主要补充四类事件：

事件 谁负责上报
工具开始执行 工具内部的 monitor.report_tool(...)
调用子智能体 run_deep_agent() 解析 task 工具调用后上报
最终结果 run_deep_agent() 解析最终 content 后上报
会话目录 run_deep_agent() 准备阶段上报

9.7 第七步：异常处理和上下文清理

执行过程外层有 try / except / finally：

1
2
3
4
5
6
7
8
9
try:
    async for chunk in main_agent.astream(...):
        ...
except Exception as e:
    # 异步执行异常也走 monitor，保证前端能收到明确错误事件
    monitor._emit("error", f"执行主智能体发生异常信息：{str(e)}")
finally:
    # 任务结束后恢复 ContextVar，避免后续请求复用到本次会话目录或 thread_id
    reset_session_context(session_dir_token, session_id_token)

异常时通过 monitor._emit("error", ...) 通知前端。

finally 里必须重置上下文：

1
reset_session_context(session_dir_token, session_id_token)

否则 Web 服务长期运行时，上一轮任务的 session_id 或 session_dir 可能残留，影响后续请求。

10、run_deep_agent 完整骨架

把上面的部分合起来，run_deep_agent() 的骨架如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
async def run_deep_agent(task_query, session_id):
    """
    异步流式执行主智能体

    API 层会为每次任务传入用户问题和 session_id。本函数负责准备会话目录、
    复制上传文件、写入 ContextVar，并在流式执行过程中把关键事件上报给前端。
    :param task_query: 前端提交的原始任务问题
    :param session_id: 当前任务 ID，同时用于 thread_id、输出目录和 WebSocket 定向推送
    """
    print(f"[MainAgent] 开始执行会话，session_id={session_id}")

    # 每个会话独立使用 output/session_{session_id}，避免不同用户的产物互相覆盖
    session_dir = project_root_path / "output" / f"session_{session_id}"
    session_dir.mkdir(parents=True, exist_ok=True)

    # 前端和工具使用绝对路径；提示词里只给模型相对路径，降低模型误用系统绝对路径的概率
    session_dir_str = str(session_dir).replace("\\", "/")
    relative_session_dir_str = str(session_dir.relative_to(project_root_path)).replace(
        "\\", "/"
    )

    # 上传文件先落在 updated/session_{session_id}，执行前复制到本次 output 工作目录
    # 这样读文件工具和生成文件工具都只需要围绕同一个 session_dir 工作
    updated_dir_path = project_root_path / "updated" / f"session_{session_id}"
    updated_info_prompt = ""
    if updated_dir_path.exists():
        files = [f.name for f in updated_dir_path.iterdir() if f.is_file()]
        if files:
            for filename in files:
                # copy2 会保留上传文件的修改时间、权限等元数据，便于后续排查文件来源
                shutil.copy2(updated_dir_path / filename, session_dir / filename)

            # 把上传文件列表注入用户消息，提醒模型先调用 read_file_content 获取附件内容
            updated_info_prompt = (
                "\n    [已上传文件] 已加载到工作目录:\n"
                + "\n".join([f"    - {f}" for f in files])
                + "\n    请优先使用工具（read_file_content）读取并参考这些文件。"
            )

    # ContextVar 让深层工具无需显式传参，也能拿到当前会话目录和 WebSocket thread_id
    session_dir_token = set_session_context(session_dir_str)
    session_id_token = set_thread_context(session_id)

    # 前端拿到工作目录后，可以展示本次任务生成的 Markdown/PDF 等产物
    monitor.report_session_dir(session_dir_str)

    # checkpointer 依赖 thread_id 区分会话记忆；同一 session_id 会复用同一条执行上下文
    config = {"configurable": {"thread_id": session_id}}

    # 工作环境指令是运行时动态补充的，约束模型只在当前会话目录读写文件
    path_instruction = f"""
    【工作环境指令】
    工作目录: {relative_session_dir_str}
    {updated_info_prompt}

    规则：
    1. 新生成文件必须保存到工作目录：'{relative_session_dir_str}/filename'
    2. 读取已上传的文件时，请直接将文件名作为 filename 参数传入，不要带目录前缀。
    3. 使用相对路径，禁止使用绝对路径
    4. 若存在上传文件，请先分析内容
    """

    try:
        # astream 会持续产出模型节点、工具节点和子智能体节点的状态片段
        async for chunk in main_agent.astream(
            {"messages": [{"role": "user", "content": task_query + path_instruction}]},
            config=config,
        ):
            # chunk 形如 {"model": {"messages": [...]}}，这里主要关心模型最新消息
            for node_name, state in chunk.items():
                if not state or "messages" not in state:
                    continue

                messages = state["messages"]
                if messages and isinstance(messages, list):
                    last_msg = messages[-1]
                    if node_name == "model":
                        if last_msg.tool_calls:
                            # DeepAgents 调用子智能体时，本质上会产生名为 task 的工具调用
                            for tool_call in last_msg.tool_calls:
                                if tool_call["name"] == "task":
                                    # 子智能体调用单独上报，前端可以展示“正在调用哪个专家助手”
                                    monitor.report_assistant(
                                        tool_call["args"]["subagent_type"],
                                        {
                                            "description": tool_call["args"][
                                                "description"
                                            ]
                                        },
                                    )
                        elif last_msg.content:
                            # 模型没有继续调用工具时，最新文本内容就是本轮可反馈给前端的结果
                            print(
                                f"主智能体执行结果，最终结果：{last_msg.content[:100]}"
                            )
                            monitor.report_task_result(last_msg.content)

    except Exception as e:
        # 异步执行异常也走 monitor，保证前端能收到明确错误事件
        monitor._emit("error", f"执行主智能体发生异常信息：{str(e)}")
    finally:
        # 任务结束后恢复 ContextVar，避免后续请求复用到本次会话目录或 thread_id
        reset_session_context(session_dir_token, session_id_token)

前面三个文件工具已经分别进行了本地验证。这里再进行主智能体层面验证，用来确认 run_deep_agent() 能把模型、子智能体、文件工具和会话目录串起来。

1
2
3
4
5
6
7
8
9
if __name__ == "__main__":
    import asyncio

    asyncio.run(
        run_deep_agent(
            "从网络查询机器人信息，并生成Markdown文件",
            "test_session_001"
        )
    )

但要注意：这个测试入口只适合本地脚本验证。下一章接入 FastAPI 时，接口里不要用 asyncio.run()，而是用 asyncio.create_task(...) 把任务丢到 FastAPI 已有事件循环里。

在项目根目录下执行：uv run python -m app.agent.main_agent

主智能体脚本验证结果：生成 session_test_session_001 工作目录和机器人信息 Markdown 文件

本章小结：

本章把主智能体从“创建出来”推进到“能被异步执行”。你需要记住两条主线。

第一条是主智能体组装链路：

1
2
3
模型 + 主提示词 + 文件工具 + 三个子智能体 + InMemorySaver
  -> create_deep_agent
  -> main_agent

第二条是主智能体运行链路：

1
2
3
4
5
HTTP 启动任务
  -> run_deep_agent 异步执行
  -> main_agent.astream 流式输出
  -> monitor 上报关键事件
  -> WebSocket 推送给前端

最后再记住两个隔离点：

  1. thread_id 隔离 WebSocket 消息和检查点，避免消息发错前端。
  2. session_dir 隔离文件目录，避免上传文件和生成文件写错会话。

下一章会正式编写 FastAPI 接口，把本章的 run_deep_agent() 接到前端请求上，并解释 /api/task、/api/upload、/api/files、/api/download、/ws/{thread_id} 各自负责什么。