2-DeepAgents快速入门与流式解析

2 - 深度研搜：DeepAgents 快速入门与流式解析

---

本章课程目标：

• 跑通第一个 DeepAgent：给智能体配置联网搜索工具，并用 invoke() 获取最终结果。
• 理解 result["messages"] 的结果结构，知道一次 Agent 执行过程中模型和工具分别做了什么。
• 掌握 stream() 的基本使用方式，能解析 DeepAgents 的流式输出。
• 能区分“模型决策”“工具执行”和“最终回复”几类关键事件，并知道子智能体调度也可以作为后续扩展事件识别。
• 为下一章学习 subagents 配置和主智能体调度做准备。

学习建议： 这一章先跑，再拆。先把最小示例执行成功，确认普通返回和 stream() 都能看到结果；再回头看 messages、chunk、tool_calls、content 这些字段分别从哪里来。DeepAgents 内部节点第一遍不用追到底，先能读懂流式输出里每一段代表什么。

对应代码分支： 02-quickstart-streaming

---

上一章我们已经理解了 DeepAgents 的定位：它适合处理长链路、复杂、多步骤、需要自主规划和分工的智能体任务。

从这一章开始，我们进入代码。先不急着做完整的「深度研搜」项目，而是先用一个最小示例理解 DeepAgent 的创建、执行和流式输出。

本章要完成的案例很简单：

用户提问：请查询人工智能和机器人领域的热门新闻信息，并整理为一份简要报告。
智能体动作：判断需要搜索 -> 调用 internet_search -> 整理搜索结果 -> 返回报告

通过这个案例，你会看到一次 Agent 执行过程并不是“模型直接返回答案”，而是下面这条链路：

用户问题
  -> 模型判断是否需要工具
  -> Agent 执行搜索工具
  -> 工具返回搜索结果
  -> 模型整理搜索结果
  -> 返回最终报告

这段流程图可以按“谁在做事”来读：用户只负责提出问题，模型负责判断下一步，工具负责拿外部资料，最后还是由模型把工具结果整理成人能读懂的报告。这条链路理解清楚后，后面加入子智能体、知识库、数据库、文件生成和 WebSocket 实时推送时，本质上都是在扩展它。

---

1、快速入门：创建第一个 DeepAgent

1.1 安装依赖

在运行代码之前，先准备 Python 环境和依赖。当前教学代码仓库 deepsearch-agents，已经使用 uv 管理 Python 环境和依赖。

如果你还不了解 uv，可以先参考「3 - 电商问数：开发环境与基础服务准备」中关于 uv 的入门说明。本章不再展开讲 uv 的基础概念，只需要知道：在这个项目里，依赖已经写在 pyproject.toml 中，第一次运行时直接同步即可。

当前 deepsearch-agents/pyproject.toml 中指定的 Python 版本是：

requires-python = ">=3.12,<3.13"

也就是说，本项目建议使用 Python 3.12。如果你已经安装好 uv，进入 deepsearch-agents 项目根目录后，执行：

uv sync

uv sync 会根据项目中的 pyproject.toml 和 uv.lock 自动创建 .venv，并安装锁定好的依赖版本。当前项目核心依赖如下：

依赖	当前版本	作用
deepagents	0.5.7	创建 DeepAgent 的核心框架
langchain	1.2.17	提供工具封装、消息结构和模型初始化入口
langchain-openai	1.2.1	通过 OpenAI 兼容协议连接模型服务
langgraph	1.1.10	DeepAgents 底层图执行与状态流转能力
python-dotenv	1.2.2	从 .env 文件读取环境变量
tavily-python	0.7.24	调用 Tavily 联网搜索

如果你后面需要新增依赖，再使用 uv add 包名；如果只是运行当前仓库中的教学示例，通常执行 uv sync 就可以了。

1.2 代码位置与执行步骤

项目对应文件路径：deepsearch-agents/examples/1-deep-agent-quickstart-search.py、deepsearch-agents/examples/2-deep-agent-streaming-chunks.py

其中，1-deep-agent-quickstart-search.py 用来演示非流式调用，2-deep-agent-streaming-chunks.py 用来演示流式解析。

代码执行顺序如下：

1. 安装本章依赖。
2. 配置 .env。
3. 创建 Tavily 搜索客户端。
4. 使用 @tool 把普通 Python 函数封装成 Agent 可调用工具。
5. 初始化大模型。
6. 使用 create_deep_agent() 创建 DeepAgent。
7. 使用 invoke() 非流式执行。
8. 从 result["messages"][-1].content 中取最终结果。

运行非流式示例：

uv run examples/1-deep-agent-quickstart-search.py

运行流式解析示例：

uv run examples/2-deep-agent-streaming-chunks.py

1.3 配置环境变量

代码会从 .env 中读取模型和 Tavily 搜索配置。实际开发时不要把真实 Key 写进文档，也不要提交到仓库中，建议使用下面这种占位方式：

OPENAI_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
OPENAI_API_KEY=你的大模型_API_KEY
LLM_QWEN_MAX=qwen-max

# 在 https://app.tavily.com/ 注册账号，然后创建API KEY
TAVILY_API_KEY=你的_TAVILY_API_KEY

1.4 封装联网搜索工具

DeepAgents 需要通过工具与外部世界交互。本章使用 TavilyClient 封装一个简单的互联网搜索工具。

import os
from typing import Literal

from dotenv import load_dotenv, find_dotenv
from langchain.tools import tool
from tavily import TavilyClient

load_dotenv(find_dotenv())

tavily_key = os.getenv("TAVILY_API_KEY")

# Tavily 客户端负责真正的联网搜索，工具函数中会复用这个客户端
tavily_client = TavilyClient(api_key=tavily_key)


@tool
def internet_search(
    query: str,
    max_results: int = 5,
    topic: Literal["news", "finance", "general"] = "general",
    include_raw_content: bool = False,
):
    """
    互联网搜索工具

    DeepAgent 会根据工具描述和参数签名，自动决定是否调用该工具
    :param query: 搜索关键词
    :param max_results: 返回结果数量
    :param topic: 查询主题，可选 news、finance、general
    :param include_raw_content: 是否返回更详细的原文内容，include_raw_content=False 时返回摘要内容；True 时会尝试返回更完整的网页原文
    :return: Tavily 搜索结果
    """
    print(
        f"开始调用网络搜索工具，核心参数为：{query},{max_results},{topic},{include_raw_content}"
    )
    return tavily_client.search(
        query=query,
        max_results=max_results,
        topic=topic,
        include_raw_content=include_raw_content,
    )

这里最关键的是 @tool 装饰器。它会把一个普通 Python 函数包装成 LangChain 工具，让 Agent 能够在推理过程中调用它。

工具的函数签名和注释也很重要。模型会根据工具名称、参数类型和描述判断什么时候该调用这个工具、应该传什么参数。如果工具描述太含糊，模型就更容易误用。

1.5 初始化模型

接下来初始化大模型对象：

from langchain.chat_models import init_chat_model

llm_name = os.getenv("LLM_QWEN_MAX")

llm = init_chat_model(
    model=llm_name,
    model_provider="openai",
)

这里的 llm 是 DeepAgent 做判断和生成回复时使用的模型对象。无论是普通 Agent 还是 DeepAgent，都必须有模型参与决策。

1.6 创建 DeepAgent

创建 DeepAgent 的核心函数是 create_deep_agent()。

from deepagents import create_deep_agent

# 当前示例不配置子智能体，重点观察“主智能体 + 搜索工具”的基本流程
deep_agent = create_deep_agent(
    model=llm,
    tools=[internet_search],
    subagents=[],
    system_prompt="""
    你是一名严谨的研究员，可以使用 internet_search 工具检索网络信息。
    请根据检索结果进行归纳、分析和交叉验证，生成一份结构清晰、信息可靠的中文报告。
    """,
)

先看四个核心参数：

参数	含义
model	智能体使用的大模型
tools	主智能体可以直接调用的工具列表
subagents	子智能体列表，本节先留空
system_prompt	主智能体的系统提示词，用来定义角色、目标和工具使用边界

这一版代码中虽然 subagents=[]，但它依然是 DeepAgent。只是当前示例先用一个搜索工具帮助我们理解 Agent 的执行过程，子智能体会在后面单独讲。

1.7 先用 invoke 跑通结果

最小示例先使用 invoke() 执行：

# 非流式执行，invoke 会等整条 agent 链路完成后，一次性返回最终状态
result = deep_agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "请查询人工智能和机器人领域的热门新闻信息，并整理为一份简要报告。",
            }
        ]
    }
)

# result 中会保留完整消息轨迹，便于观察模型决策、工具返回和最终回答
print(result)

# messages 的最后一条通常就是 DeepAgent 整理后的最终回答
print(result["messages"][-1].content)

注意入参不是简单字符串，而是一个包含 messages 的字典：

{
    "messages": [
        {"role": "user", "content": "用户问题"}
    ]
}

这是因为 DeepAgents 底层沿用了 LangGraph / LangChain 的消息模型，整个执行过程都会围绕 messages 传递状态。

执行文件验证，成功：

uv run examples/1-deep-agent-quickstart-search.py

LangChainPendingDeprecationWarning: The default value of `allowed_objects` will change in a future version.
开始调用网络搜索工具，核心参数为：人工智能 机器人 热门新闻,5,news,False

{
  "messages": [
    HumanMessage(
      content="请查询人工智能和机器人领域的热门新闻信息，并整理为一份简要报告。"
    ),
    AIMessage(
      content="",
      tool_calls=[
        {
          "name": "internet_search",
          "args": {
            "query": "人工智能 机器人 热门新闻",
            "topic": "news",
            "max_results": 5,
            "include_raw_content": False
          }
        }
      ]
    ),
    ToolMessage(
      name="internet_search",
      content="Tavily 返回了 5 条新闻结果，包含 url、title、score、published_date、content 等字段"
    ),
    AIMessage(
      content="## 人工智能与机器人领域热门新闻简报\n\n1. Genesis AI 推出新型机器人大脑...\n2. Meta 收购 ARI...\n3. 无人驾驶出租车将重塑出行市场..."
    )
  ]
}

## 人工智能与机器人领域热门新闻简报

1. Genesis AI 推出新型机器人“大脑”
2. Meta 收购 Assured Robot Intelligence（ARI）
3. 无人驾驶出租车将重塑出行市场
4. 四月机器人行业十大新闻回顾
5. 值得关注的欧洲初创企业

其中，LangChainPendingDeprecationWarning 是 LangGraph / LangChain 依赖内部的版本提示，不影响本章示例运行。真正要观察的是后面的工具调用日志、messages 轨迹和最终报告。

这段输出可以按三层来看：

输出位置	说明
开始调用网络搜索工具...	工具函数内部的 print()，说明 internet_search 被真正执行
result["messages"]	完整执行轨迹，包含用户消息、模型工具调用、工具返回和最终回答
最后的 Markdown 报告	result["messages"][-1].content，也就是最终展示给用户的内容

这里先建立一个直觉：invoke() 虽然最后拿到的是一份报告，但它内部并不是“一次模型回答”，而是经历了“模型决定调工具 -> 工具返回结果 -> 模型整理答案”的完整链路。

---

2、非流式调用：invoke 与 messages

2.1 实际调用解析

invoke() 返回的不是一段纯文本，而是一个包含完整消息过程的字典。简化后可以理解成这样：

result = {
    "messages": [
        # 用户问题
        HumanMessage(
            content="请查询人工智能和机器人领域的热门新闻信息，并整理为一份简要报告。"
        ),
        # 模型决定调用 internet_search 工具
        AIMessage(
            content="",
            tool_calls=[
                {
                    "name": "internet_search",
                    "args": {
                        "query": "人工智能 机器人 热门新闻",
                        "topic": "news",
                        "max_results": 5,
                        "include_raw_content": False
                    }
                }
            ]
        ),
        # internet_search 工具返回搜索结果
        ToolMessage(
            name="internet_search",
            content='{"query": "人工智能 机器人 热门新闻", "results": [...]}'
        ),
        # 模型基于工具结果生成最终回答
        AIMessage(
            content="## 人工智能与机器人领域热门新闻简报..."
        ),
    ]
}

也就是说，一次 Agent 执行大致经历下面几步：

1. 用户把问题发给 Agent。
2. Agent 把用户问题、系统提示词、工具描述一起交给模型。
3. 模型判断需要调用 internet_search。
4. Agent 真正执行搜索工具。
5. 工具返回搜索结果。
6. Agent 再把工具结果交给模型。
7. 模型整理搜索结果，生成最终报告。

这 7 步背后，其实有四个角色在配合：

用户 -> Agent -> 大模型 -> 工具 -> Agent -> 大模型 -> 用户

第一步，用户并不是直接调用大模型，而是把问题交给 deep_agent：

result = deep_agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "请查询人工智能和机器人领域的热门新闻信息，并整理为一份简要报告。",
            }
        ]
    }
)

这一步传入的是 messages，DeepAgents 会先把这条用户消息包装成 HumanMessage，放进本次执行状态中。

第二步，Agent 会把下面几类信息一起交给大模型：

• 用户问题：请查询人工智能和机器人领域的热门新闻信息，并整理为一份简要报告。
• 系统提示词：你是一名严谨的研究员，可以使用搜索工具生成结构清晰、信息可靠的中文报告
• 工具说明：当前可用工具是 internet_search
• 工具参数结构：query、max_results、topic、include_raw_content

也就是说，大模型并不是“天然知道”Agent 有哪些工具，而是 Agent 在调用模型时，把工具名称、参数和描述一起放进上下文里。模型读完这些信息后，才会判断：这个问题需要联网搜索，于是生成一次工具调用。

这一步对应 result["messages"] 里的第二条消息：

AIMessage(
    content="",
    tool_calls=[
        {
            "name": "internet_search",
            "args": {
                "query": "人工智能 机器人 热门新闻",
                "topic": "news",
                "max_results": 5,
                "include_raw_content": False
            }
        }
    ]
)

这里要注意：这条 AIMessage 还不是最终回答。它的 content 为空，但 tool_calls 有值，意思是模型在告诉 Agent：“下一步请调用 internet_search 工具，并传入这些参数。”

第三步，Agent 根据 tool_calls 真正去执行工具，也就是调用我们前面定义的 Python 函数：

@tool
def internet_search(query: str, max_results: int = 5, ...):
    return tavily_client.search(...)

工具执行完成后，会生成一条 ToolMessage。这条消息里保存的是 Tavily 返回的原始搜索结果，通常是结构化数据，里面可能包含标题、链接、摘要、发布时间等信息。

结合实际输出，ToolMessage(content=...) 里的 JSON 主要包含这些字段：

字段	含义
query	本次传给 Tavily 的搜索关键词
results	搜索结果列表
url	单条搜索结果的网页链接
title	单条搜索结果的标题
content	单条搜索结果的摘要内容
published_date	新闻发布时间
response_time	Tavily 本次请求耗时
request_id	Tavily 本次请求 ID

第四步，工具结果不会直接返回给用户。Agent 会把 ToolMessage 再交给大模型，让模型根据搜索结果进行整理、翻译、筛选和润色。比如 Tavily 返回的结果可能包含英文网页摘要，但用户提问是中文，所以最终结果通常会被模型整理成中文报告。

这一步对应最后一条 AIMessage：

AIMessage(
    content="## 人工智能与机器人领域热门新闻简报..."
)

到了这里，tool_calls 已经没有值，content 有了完整内容，说明模型不再继续调用工具，而是已经生成最终回复。

所以我们取最终回复时，通常直接取最后一条消息：

final_answer = result["messages"][-1].content

如果只关心最终报告，读取最后一条消息即可；如果要调试 Agent 的执行链路，再查看完整的 result。

2.2 为什么模型知道该调用哪个工具

它是 Agent 工具调用的核心。

模型本身并不会主动扫描你的 Python 代码，也不会自己知道项目里有哪些函数。真正发生的是：当我们通过 create_deep_agent() 创建智能体时，已经把工具列表传给了 Agent。

deep_agent = create_deep_agent(
    model=llm,
    tools=[internet_search],
    subagents=[],
    system_prompt="..."
)

执行 invoke() 时，Agent 会把 internet_search 的名称、描述和参数结构整理给模型。模型读完以后，才会决定是否生成 tool_calls。

因此，工具调用的判断链路可以这样记：

工具注册到 Agent
  -> Agent 把工具描述交给模型
  -> 模型判断是否需要调用工具
  -> 模型生成 tool_calls
  -> Agent 根据 tool_calls 执行真实工具

这也解释了为什么前面一直强调工具函数的名称、参数类型和注释要写清楚。它们不是只给开发者看的，也会影响模型能不能正确调用工具。

2.3 读懂四类消息

为了读懂 result["messages"]，可以先记住下面这张表：

消息类型	在执行链路中的作用	常见特征
HumanMessage	用户输入的问题	content 是用户原始问题
AIMessage	模型的思考结果，可能是调用工具，也可能是回复	有 tool_calls 时表示准备调工具
ToolMessage	工具真实执行后的返回结果	name 通常是工具名
AIMessage	模型基于工具结果整理出的最终回答	content 有最终自然语言结果

其中最容易混淆的是 AIMessage。它有两种情况：

• AIMessage(content="", tool_calls=[...])：模型还没有回答，它只是决定下一步调用什么工具。
• AIMessage(content="...")：模型已经生成最终回答。

所以判断 Agent 到底在干什么，不能只看消息类型，还要继续看 tool_calls 和 content。

---

3、流式调用：stream 与 chunk

3.1 为什么要改用 stream

invoke() 属于非流式执行。它的特点是调用后一直等待，直到整个智能体任务执行完，再一次性返回完整结果。

用户点击发送 -> 后端执行 5 秒或 10 秒 -> 一次性返回最终结果

这种方式写起来简单，但用户体验不够好。尤其是 DeepAgents 这类长链路任务，智能体可能会搜索、阅读、调用多个工具、分派子任务。如果前端一直没有反馈，用户很难判断系统是在处理，还是已经卡住。

流式执行的目标就是让用户看到过程：

开始处理
模型决定调用搜索工具
搜索工具返回结果
模型整理搜索结果
最终回复

后面做 Web 项目时，这些过程通常会通过 SSE 或 WebSocket 推送给前端。本项目完整版本采用的是 WebSocket，因为它更适合持续推送 Agent 的执行状态和工具日志。

3.2 用 stream 获取执行过程

项目对应文件路径：deepsearch-agents/examples/2-deep-agent-streaming-chunks.py

把 invoke() 换成 stream() 后，返回值就不再是完整结果，而是一个可以不断迭代的流：

# 流式执行，stream 会在每个图节点完成后产出一个 chunk
# 常见节点包括 model（模型决策或最终回答）和 tools（工具执行结果）
stream = deep_agent.stream(
    {
        "messages": [
            {
                "role": "user",
                "content": "请查询人工智能和机器人领域的热门新闻信息，并整理为一份简要报告。",
            }
        ]
    }
)

# 每个 chunk 是一个按节点名组织的字典
for chunk in stream:
    print(chunk)

每个 chunk 都表示某个节点刚刚产生了一次状态更新。由于 DeepAgents 底层基于 LangGraph，流式输出会带有节点信息。

简化来看，常见的 chunk 可能长这样：

{
    "model": {
        "messages": [AIMessage(content="", tool_calls=[...])]
    }
}

或者：

{
    "tools": {
        "messages": [ToolMessage(content="工具返回结果...")]
    }
}

解析流式输出时，重点看两层信息：

1. 外层节点名：当前是 model 节点，还是 tools 节点。
2. 内层最后一条消息：模型是决定调用工具，还是已经给出最终回答。

3.3 chunk 的四种关键状态

状态	如何识别	含义
模型决定调用工具	node_name == "model" 且 last_msg.tool_calls 中工具名不是 task	模型判断下一步要调用某个普通工具
模型决定调用子智能体	node_name == "model" 且 tool_call["name"] == "task"	模型判断下一步要把任务分派给某个子智能体
工具执行完成	node_name == "tools"	Agent 真正调用工具，并拿到了工具返回结果
模型返回最终结果	node_name == "model" 且没有 tool_calls，但 last_msg.content 有内容	模型基于前面结果生成最终回复

本章的示例只配置了 internet_search，所以运行时主要会看到“普通工具调用”。表格中的子智能体状态是先给后面留一个入口：等第 3 章真正配置 subagents 后，流式输出里就可能出现这类调度事件。

3.4 chunk 四种状态的对象视图示例

只看上面的表格还比较抽象。下面参考项目文档中的写法，把 stream() 迭代过程中可能出现的几类 chunk 展开成 Python 对象视图。

注意，这里不是要求你手写这些对象，而是帮助你在 print(chunk) 时能快速识别当前 Agent 运行到哪一步。

场景 A：模型决定调用普通工具。

本章代码里最常见的就是这一类。模型读完用户问题后，发现需要联网搜索，于是生成一个 tool_calls，工具名是 internet_search。

{
    "model": {
        "messages": [
            AIMessage(
                content="",
                tool_calls=[
                    {
                        "name": "internet_search",
                        "args": {
                            "query": "人工智能和机器人 热门新闻",
                            "topic": "news",
                            "max_results": 5,
                        },
                        "id": "call_xxx",
                    }
                ],
            )
        ]
    }
}

这个状态的重点是：content 为空，tool_calls 有值。说明模型还没有回答用户，而是在告诉 Agent：“下一步请调用这个工具。”

场景 B：工具执行完成并返回结果。

当 Agent 真正执行完 internet_search 后，会从 tools 节点产出 ToolMessage：

{
    "tools": {
        "messages": [
            ToolMessage(
                content='{"query": "人工智能和机器人 热门新闻", "results": [...]}',
                name="internet_search",
                tool_call_id="call_xxx",
            )
        ]
    }
}

这个状态的重点是：外层节点名变成了 tools。此时不是模型在思考，而是工具已经执行完，并把原始结果交回给 Agent。

场景 C：模型决定调用子智能体。

本章还没有真正配置子智能体，所以这类状态通常不会在当前示例里出现。但后面第 3 章加入 subagents 后，如果模型决定把任务委派给子智能体，就会看到特殊工具 task：

{
    "model": {
        "messages": [
            AIMessage(
                content="",
                tool_calls=[
                    {
                        "name": "task",
                        "args": {
                            "subagent_type": "network_search_agent",
                            "description": "查询人工智能和机器人领域的热门新闻，并返回来源、摘要和关键信息。",
                        },
                        "id": "call_yyy",
                    }
                ],
            )
        ]
    }
}

task 本质上也是一次工具调用，只是它调用的不是普通 Python 工具，而是 DeepAgents 的子智能体调度入口。解析时只要看到 tool_call["name"] == "task"，就可以把它识别成“正在分派给子智能体”。

场景 D：模型生成最终回复。

当工具结果已经返回，模型完成整理后，会再次从 model 节点产出消息。这一次没有 tool_calls，而是有真正的 content：

{
    "model": {
        "messages": [
            AIMessage(
                content="以下是关于人工智能和机器人领域的热门新闻概要：...",
                tool_calls=[],
            )
        ]
    }
}

这个状态说明 Agent 链路已经收束，模型不再继续调用工具或子智能体，而是把最终结果返回给用户。

除了上面四类业务事件，实际输出中还可能看到一些中间件节点，例如 PatchToolCallsMiddleware.before_agent、TodoListMiddleware.after_model。这些节点可能不包含 messages，或者不是我们要展示给用户的业务进度，所以后面的解析代码会用下面这句过滤掉：

if not state or "messages" not in state:
    continue

3.5 解析 chunk 的完整代码

项目对应文件路径：deepsearch-agents/examples/2-deep-agent-streaming-chunks.py

# 流式执行，stream 会在每个图节点完成后产出一个 chunk
stream = deep_agent.stream(
    {
        "messages": [
            {
                "role": "user",
                "content": "请查询人工智能和机器人领域的热门新闻信息，并整理为一份简要报告。",
            }
        ]
    }
)

for chunk in stream:
    # chunk 是一个按节点名组织的字典，例如
    # {"model": {"messages": [...]}} 或 {"tools": {"messages": [...]}}
    for node_name, state in chunk.items():
        # DeepAgents 内部中间件也可能产出空状态或非消息状态，这里只解析消息类状态
        if not state or "messages" not in state:
            continue

        messages = state["messages"]

        if not messages or not isinstance(messages, list):
            continue

        # 每个 chunk 的最后一条消息，通常就是这个节点本次产出的核心信息
        last_msg = messages[-1]

        if node_name == "model":
            # model 节点有两类重点事件：
            # 1. tool_calls 非空，模型决定下一步调用工具或子智能体
            # 2. content 非空，模型已经生成最终回答
            if last_msg.tool_calls:
                for tool_call in last_msg.tool_calls:
                    if tool_call["name"] == "task":
                        print(
                            f"【大模型】决定调用子智能体：{tool_call['args']['subagent_type']}"
                        )
                    else:
                        print(
                            f"【大模型】决定调用工具：{tool_call['name']} 传入的参数：{tool_call['args']}"
                        )

            elif last_msg.content:
                print(f"【大模型】最终执行的结果：{last_msg.content}")

        elif node_name == "tools":
            # tools 节点返回的是具体工具的执行结果，通常可以推送给前端展示执行进度
            tool_return_result = last_msg.content[:100] + "..."
            tool_name = last_msg.name
            print(f"【agent】调用了{tool_name}工具，返回的结果为：{tool_return_result}")

这段代码里有几个细节要特别注意。

**第一，**外层循环取的是 node_name 和 state：

for node_name, state in chunk.items():

node_name 用来判断当前是哪类节点产生了输出，例如 model 或 tools。state 里才是真正的状态数据。

**第二，**有些中间件节点不包含 messages，要先跳过：

if not state or "messages" not in state:
    continue

DeepAgents 内部可能会出现一些中间件节点，例如 TodoListMiddleware.after_model。它们对框架执行有意义，但不一定是我们要展示给用户的业务事件。

**第三，**只取 messages[-1]：

last_msg = messages[-1]

因为每个状态更新里可能包含多条消息，但当前这次更新最关键的一条通常在最后。

第四，model 节点要继续区分两类情况：

• last_msg.tool_calls 有值：模型不是最终回答，而是在决定下一步调用什么。
• last_msg.content 有值：模型已经生成最终回复。

第五，tools 节点代表工具已经真的执行完了。这里可以把工具名和返回结果截断后推送给前端，避免一大段搜索结果把界面刷满。

执行文件验证，成功：

uv run examples/2-deep-agent-streaming-chunks.py

LangChainPendingDeprecationWarning: The default value of `allowed_objects` will change in a future version.
【大模型】决定调用工具：internet_search 传入的参数：{'query': '人工智能和机器人 热门新闻', 'topic': 'news', 'max_results': 5}
开始调用网络搜索工具，核心参数为：人工智能和机器人 热门新闻,5,news,False
【agent】调用了internet_search工具，返回的结果为：{"query": "人工智能和机器人 热门新闻", "follow_up_questions": null, "answer": null, "images": [], "results": [{"....
【大模型】最终执行的结果：以下是关于人工智能和机器人的几条热门新闻概要：

1. **新AI大脑让机器人像人类一样移动 - Fox News**
   - Genesis AI推出了一款名为GENE-26.5的机器人“大脑”，它能够帮助通用型机器人执行复杂的物理任务，就像人类一样操作。该系统通过大量数据训练，可以让机器人手部模仿人类手部的动作，完成诸如制作煎蛋等多步骤过程的任务。

2. **如何重塑乘车市场 - Financial Times 视频报道**
   - 本视频探讨了人工智能如何推动下一代自动驾驶出租车的发展，并分析了这一技术对当前乘车市场的潜在影响。

3. **为什么实体AI才是真正的制造业革命 - The Robot Report**
   - 文章讨论了实体AI在制造业中的应用以及其对于硬件创新的重要性。指出实体AI正在使机器人学习与适应环境的能力变得更快更强，并且强调了区分人形机器人炒作与实际硬件扩展之间的差异。

4. **不仅仅是汽车：关注2026年中国汽车展上的机器人转变 - CleanTechnica**
   - 在北京国际汽车展览会上，中国汽车产业开始从电动汽车和软件定义车辆转向所谓的“实体AI”——即超越车辆本身、进入现实世界的机器智能。多家参展商展示了基于共享智能平台的各种机器人和服务机器人。

如果您需要更多详细信息或想深入了解某个特定主题，请告诉我！

这段输出比 invoke() 更适合观察执行过程，可以按顺序读：

输出片段	对应的执行阶段
【大模型】决定调用工具...	model 节点输出，模型决定下一步调用 internet_search
开始调用网络搜索工具...	Python 工具函数真正开始执行
【agent】调用了internet_search工具...	tools 节点输出，工具结果已经返回
【大模型】最终执行的结果...	model 节点再次输出，模型基于工具结果生成最终回答

这就是流式解析最核心的价值：不用等到最终报告生成，开发者和前端都能看到 Agent 当前走到了哪一步。

3.6 对接前端时怎么展示

虽然本章还没有写 FastAPI 接口，但可以提前建立后端事件和前端展示之间的对应关系。

后端解析出的事件	前端可以展示成什么
模型决定调用工具 internet_search	正在搜索网络资料
工具返回搜索结果	已获取搜索结果，正在整理
模型决定调用子智能体	正在分派给某个专业助手
模型返回最终结果	展示最终答案或报告

这也是为什么流式解析很重要。企业级智能体项目不只看最终答案，还要让用户能感知执行过程，开发者也能定位每一步做了什么。

一句话总结： invoke() 看最终结果，stream() 看执行过程；做真实产品时，流式处理几乎是必须掌握的能力。

---

本章小结：

这一章我们完成了 DeepAgents 的快速入门。先定义了 internet_search 搜索工具，再初始化模型，最后通过 create_deep_agent() 组装主智能体，并用 invoke() 获取最终结果。

接着我们重点学习了 stream() 的流式解析方式。只要抓住 model 和 tools 两类节点，再结合 tool_calls 和 content，就能判断当前到底是模型在做决策、工具在执行，还是模型已经生成最终结果；后面加入子智能体后，也是在这套解析逻辑上继续扩展。

完整的子智能体配置、调度信号和异步执行，会在下一章展开。你现在只需要先掌握这一层：invoke() 适合看最终结果，stream() 适合看清 Agent 每一步正在做什么。