Agent信息流概述

我们需要了解，当我们给Agent输入一个query时，它会经过哪些处理，最终得到输出呢？通过了解这整个过程，可以使我们对Agent有一个更深入的理解。

下面以my_agent1.py为例详细解释，测试代码如下：

@tool
def send_email(to: str, subject: str, body: str):
    """发送邮件 - 该工具可以发送电子邮件给指定收件人
    
    Args:
        to: 收件人邮箱地址或姓名
        subject: 邮件主题
        body: 邮件正文内容
    """
    email = {
        "to": to,
        "subject": subject,
        "body": body
    }
    # ...邮件发送逻辑
    print(f"📧 工具执行: send_email(to='{to}', subject='{subject}', body='{body}')")

    return f"邮件已发送至 {to}"


# 创建 React Agent
agent_executor = create_agent(
    model=llm,
    tools=[send_email],
    system_prompt="你是一个邮件助手。"
)

# 添加测试输入
if __name__ == "__main__":
    import asyncio
    async def main():
        # 测试输入
        inputs = {
            "messages": [
                {"role": "user", "content": "请帮我给张三发一封邮件，告诉他会议时间改到明天下午3点了，主题是项目进度同步。"}
            ]
        }
        print("用户请求:", inputs["messages"][0]["content"])
        print("\nAgent执行过程:")
        # 异步流式执行
        async for chunk in agent_executor.astream(inputs, stream_mode="updates"):
            print(chunk)
        print("\n执行完成")
    # 运行异步主函数
    asyncio.run(main())

Graph初始化过程

首先经过下面的代码会创建一个包含模型节点和工具节点的graph图：

agent_executor = create_agent(
    model=llm,
    tools=[send_email],
    system_prompt="你是一个邮件助手。"
)

在这个过程中会添加节点：

1. 添加模型节点：

graph.add_node("model", RunnableCallable(model_node, amodel_node, trace=False))

2. 添加工具节点：

graph.add_node("tools", tool_node)

3. 确定入口节点，代码如下，这里就是中间件发挥作用的地方，可以在query输入到模型之前进行一些预处理：

# 确定入口节点（在开始时运行一次）：before_agent -> before_model -> model
if middleware_w_before_agent:
    entry_node = f"{middleware_w_before_agent[0].name}.before_agent"
elif middleware_w_before_model:
    entry_node = f"{middleware_w_before_model[0].name}.before_model"
else:
    entry_node = "model"
print(f"🏁 确定入口节点: {entry_node}")

输出如下：

🏁 确定入口节点: model

4. 确定循环节点的入口：

if middleware_w_before_model:
    loop_entry_node = f"{middleware_w_before_model[0].name}.before_model"
else:
    loop_entry_node = "model"
print(f"🔄 确定循环入口节点: {loop_entry_node}")

输出如下：

🔄 确定循环入口节点: model

5. 确定循环节点的出口：

# 确定循环出口节点（每次迭代结束，可以运行多次）
# 这是after_model或model，但不是after_agent
if middleware_w_after_model:
    loop_exit_node = f"{middleware_w_after_model[0].name}.after_model"
else:
    loop_exit_node = "model"
print(f"🚪 确定循环出口节点: {loop_exit_node}")

输出如下：

🚪 确定循环出口节点: model

6. 确定出口节点：

# 确定出口节点（最后运行一次）：after_agent或END
if middleware_w_after_agent:
    exit_node = f"{middleware_w_after_agent[-1].name}.after_agent"
else:
    exit_node = END
print(f"🔚 确定最终出口节点: {exit_node}")

输出如下：

🔚 确定最终出口节点: __end__

7. 添加起始边：

# 添加起始边
print(f"🔗 添加起始边: START -> {entry_node}")
graph.add_edge(START, entry_node)

输出如下：

🔗 添加起始边: START -> model

规定了图的起始边是model，所以query首先会传递给amodel_node函数。

起始边 vs 入口节点 vs 循环入口节点

概念	作用域	执行次数	主要用途
起始边	整个图的物理起点	仅1次	定义图的执行起点
入口节点	工作流的逻辑起点	仅1次	决定从哪个业务节点开始
循环入口节点	循环迭代的起点	可能多次	Agent思考-行动循环的起点

在我的例子中，流程如下：

# 只有一个工作流，包含循环逻辑
START → 入口节点 → 循环入口节点 → 模型节点
                    ↑          ↓
                    └── 工具节点 ←─┘

8. 添加工具相关条件边：

graph.add_conditional_edges(
    "tools",  # 起始节点：名为 "tools" 的节点（负责调用工具）
    RunnableCallable(
        _make_tools_to_model_edge(...),  # 条件路由函数：决定下一步去哪
        trace=False,
    ),
    tools_to_model_destinations,  # 允许的目标节点列表（合法跳转范围）
)
print(f"🔗 工具节点条件边目标: {tools_to_model_destinations}")

代码解释： 这段代码定义了当"工具节点"（tools）执行完毕后，下一步应该跳转到哪个节点——是回到大模型继续推理，还是直接结束流程。

输出如下：

🔗 工具节点条件边目标: ['model']

结果解释： 输出显示工具节点执行后只能跳转到model节点，这意味着工具执行完成后必须返回模型进行下一步决策。

流程图表示：

[model] 
   │
   ├─(需调用工具)──→ [tools] ──→ [model] ──→ ...
   │
   └─(无需工具/完成)──→ [__end__]

9. 添加模型节点到工具节点的条件边：

# 添加从模型节点到工具节点的条件边（核心代理循环）
graph.add_conditional_edges(
    loop_exit_node,
    RunnableCallable(
        _make_model_to_tools_edge(
            model_destination=loop_entry_node,
            structured_output_tools=structured_output_tools,
            end_destination=exit_node,
        ),
        trace=False,
    ),
    model_to_tools_destinations,
)
print(f"🔗 模型节点条件边目标: {model_to_tools_destinations}")

输出如下：

🔗 模型节点条件边目标: ['tools', '__end__']

结果解释： 模型节点执行完毕后可以选择调用工具（tools）或者直接结束（end），这是Agent决策的核心逻辑。

10. 添加其他边：

# 添加before_agent中间件边
if middleware_w_before_agent:
    print("🔗 添加before_agent中间件边")

# 添加before_model中间件边
if middleware_w_before_model:
    print("🔗 添加before_model中间件边")

# 添加after_model中间件边
if middleware_w_after_model:
    print("🔗 添加after_model中间件边")

# 添加after_agent中间件边
if middleware_w_after_agent:
    print("🔗 添加after_agent中间件边")

11. 最后编译，返回图：

# 编译并返回图
print("✅ 图构建完成，准备编译")
return graph.compile(
    checkpointer=checkpointer,
    store=store,
    interrupt_before=interrupt_before,
    interrupt_after=interrupt_after,
    debug=debug,
    name=name,
    cache=cache,
).with_config({"recursion_limit": 10_000})

Agent运行过程

第一阶段：初始模型调用

当我们将input传入给graph时（这里我们以异步为例），会传给amodel_node方法，在这里，你的query和system prompt被包装成一个ModelRequest对象。

ModelRequest对象结构（简化）：

ModelRequest(
    model=ChatOpenAI(
        model_name='Qwen3-32B',
        openai_api_base='https://llmapi.paratera.com/v1/',
        openai_api_key='**********',  # 已脱敏
        default_headers={'Accept': 'application/json'},
        extra_body={'enable_thinking': False}
    ),
    messages=[
        HumanMessage(
            content='请帮我给张三发一封邮件，告诉他会议时间改到明天下午3点了，主题是项目进度同步。'
        )
    ],
    system_message=SystemMessage(
        content='你是一个邮件助手。'
    ),
    tools=[
        StructuredTool(
            name='send_email',
            description='发送邮件 - 该工具可以发送电子邮件给指定收件人\n\nArgs:\n    to: 收件人邮箱地址或姓名\n    subject: 邮件主题\n    body: 邮件正文内容',
            args_schema=send_email (Pydantic model)
        )
    ],
    tool_choice=None,
    response_format=None,
    state={
        'messages': [ ... ]  # 与 messages 相同，略
    }
)

代码解释： ModelRequest包含了用户query、系统提示词和结构化工具描述等信息，为模型调用做好准备。

模型绑定过程输出：

🔧 开始模型绑定过程...
📋 准备绑定工具到模型，工具数量: 1
   1. 工具名: send_email
      描述: 发送邮件 - 该工具可以发送电子邮件给指定收件人

    Args:
        to: 收件人邮箱地址或姓名
        subject: 邮件主题
        body: 邮件正文内容
      参数: ['to', 'subject', 'body']
🛠️  工具定义详情（将通过API参数传递给模型）:
   工具 1: {'type': 'function', 'function': {'name': 'send_email', 'description': '发送邮件 - 该工具可以发送电子邮件给指定收件人\n\n    Args:\n        to: 收件人邮箱地址或姓名\n        subject: 邮件主题\n        body: 邮件正文内容', 'parameters': {'description': '发送邮件 - 该工具可 以发送电子邮件给指定收件人\n\nArgs:\n    to: 收件人邮箱地址或姓名\n    subject: 邮件主题\n    body: 邮件正文内容', 'properties': {'to': {'title': 'To', 'type': 'string'}, 'subject': {'title': 'Subject', 'type': 'string'}, 'body': {'title': 'Body', 'type': 'string'}}, 'required': ['to', 'subject', 'body'], 'title': 'send_email', 'type': 'object'}}}
📦 使用标准方式绑定模型和工具, tool_choice: None
   注意: 工具信息不会注入到prompt中，而是通过API参数传递
   模型绑定参数:
     tool_choice: None
     model_settings: {}
✅ 模型和工具绑定完成 (标准方式)

结果解释： 这里展示了工具如何被绑定到模型，工具的描述信息会通过API参数传递给大模型，而不是直接注入到prompt中。

模型调用输出：

💬 准备发送2条消息给模型
📄 发送给模型的消息内容:
   [1] 你是一个邮件助手。
   [2] 请帮我给张三发一封邮件，告诉他会议时间改到明天下午3点了，主题是项目进度同步。
📥 模型返回结果，类型: AIMessage
📋 模型返回详细内容:
   content: ''
   additional_kwargs: {'tool_calls': [{'id': 'call_79b217f7070943b3bd01bf', 'function': {'arguments': '{"to": "zhangsan@example.com", "subject": "项目进度同步", "body": "张三，你好！会议时间已经调整到明天下午3点，请准时参加。谢谢！"}', 'name': 'send_email'}, 'type': 'function', 'index': 0}], 'refusal': None}
   检测到 1 个工具调用:
     [1] 工具名: send_email
         ID: call_79b217f7070943b3bd01bf
         参数: {"to": "zhangsan@example.com", "subject": "项目进度同步", "body": "张三，你好！会议时间已经调整到明天下午3点，请准时参加。谢谢 ！"}
   response_metadata: {'token_usage': {'completion_tokens': 56, 'prompt_tokens': 248, 'total_tokens': 304, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'Qwen3-32B', 'system_fingerprint': None, 'id': 'chatcmpl-5efb3cb7-7127-499e-a1b1-84692858dbca', 'service_tier': None, 'finish_reason': 'tool_calls', 'logprobs': None}
🛠️  模型返回1个工具调用
   工具调用 1: send_email({'to': 'zhangsan@example.com', 'subject': '项目进度同步', 'body': '张三，你好！会议时间已经调整到明天下午3点，请准时参加。谢谢！'})
🔄 开始处理模型输出...
   输出类型: AIMessage
   effective_response_format: None
📦 返回标准消息格式
📦 模型输出处理完成，消息数: 1

结果解释： 模型识别到需要调用工具，返回了一个包含工具调用的AIMessage，其中content为空，工具调用信息存储在additional_kwargs中。

amodel_node返回：

📤 amodel_node返回更新: ['messages']

重要说明： 此时模型并没有输出任何文本内容，而是通过additional_kwargs附带了一个工具调用请求。

第二阶段：工具调用决策

然后进入条件边决策函数model_to_tools，决定是否调用工具：

决策过程输出：

🧭 进入条件边决策函数model_to_tools
📨 最后AI消息的工具调用数: 1
🔧 已处理的工具消息数: 0
⏳ 待处理工具调用数: 1

决策逻辑代码：

if pending_tool_calls:
   print(f"🔧 存在待处理工具调用，转向工具节点")
   result = [
       Send(
           "tools",
           ToolCallWithContext(
               __type="tool_call_with_context",
               tool_call=tool_call,
               state=state,
           ),
       )
       for tool_call in pending_tool_calls
   ]
   print(f"📍 决策结果: 发送{len(result)}个工具调用到tools节点")
   return result

代码解释：

• 对每个待处理的tool_call，创建一个ToolCallWithContext对象，包含工具调用本身和当前状态上下文
• 使用Send("tools", ...)将对象发送到名为"tools"的节点
• 最终返回一个Send对象列表

决策结果：

📍 决策结果: 发送1个工具调用到tools节点

第三阶段：工具执行

调用我们创建的send_email方法：

工具执行输出：

📧 工具执行: send_email(to='zhangsan@example.com', subject='项目进度同步', body='张三，你好！会议时间已经调整到明天下午3点，请准时参加。谢谢！')

工具返回结果：

{'tools': {'messages': [ToolMessage(content='邮件已发送至 zhangsan@example.com', name='send_email', id='2c9e0cf3-d138-4e0b-ae48-d7fc063425d0', tool_call_id='call_79b217f7070943b3bd01bf')]}}

第四阶段：第二次模型调用

此时又要进入amodel_node节点，此时该节点获得了3条消息：

消息历史：

{
  "messages": [
    {
      "type": "HumanMessage",
      "content": "请帮我给张三发一封邮件，告诉他会议时间改到明天下午3点了，主题是项目进度同步。",
      "id": "7b1ccde5-6917-4324-b345-3273221fa874"
    },
    {
      "type": "AIMessage",
      "content": "",
      "additional_kwargs": {
        "tool_calls": [
          {
            "id": "call_79b217f7070943b3bd01bf",
            "function": {
              "name": "send_email",
              "arguments": "{\"to\": \"zhangsan@example.com\", \"subject\": \"项目进度同步\", \"body\": \"张三，你好！会议时间已经调整到明天下午3点，请准时参加。谢谢！\"}"
            },
            "type": "function",
            "index": 0
          }
        ],
        "refusal": null
      },
      "response_metadata": {
        "token_usage": {
          "completion_tokens": 56,
          "prompt_tokens": 248,
          "total_tokens": 304
        },
        "model_name": "Qwen3-32B",
        "finish_reason": "tool_calls"
      },
      "id": "lc_run--4dbc579d-ae7d-4d90-8786-9eee8ae767b2-0",
      "tool_calls": [
        {
          "name": "send_email",
          "args": {
            "to": "zhangsan@example.com",
            "subject": "项目进度同步",
            "body": "张三，你好！会议时间已经调整到明天下午3点，请准时参加。谢谢！"
          },
          "id": "call_79b217f7070943b3bd01bf",
          "type": "tool_call"
        }
      ]
    },
    {
      "type": "ToolMessage",
      "content": "邮件已发送至 zhangsan@example.com",
      "name": "send_email",
      "id": "2c9e0cf3-d138-4e0b-ae48-d7fc063425d0",
      "tool_call_id": "call_79b217f7070943b3bd01bf"
    }
  ]
}

代码解释： 现在消息历史包含了完整的对话上下文：用户请求、AI的工具调用决策、工具执行结果。

模型调用输出：

💬 准备发送4条消息给模型
📄 发送给模型的消息内容:
   [1] 你是一个邮件助手。
   [2] 请帮我给张三发一封邮件，告诉他会议时间改到明天下午3点了，主题是项目进度同步。
   [3] 包含1个工具调用
   [4] 邮件已发送至 zhangsan@example.com
📥 模型返回结果，类型: AIMessage
📋 模型返回详细内容:
   content: '邮件已经成功发送给张三，告诉他会议时间调整到了明天下午3点。如果有其他需要，请随时告诉我！'
   additional_kwargs: {'refusal': None}
   response_metadata: {'token_usage': {'completion_tokens': 25, 'prompt_tokens': 327, 'total_tokens': 352, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'Qwen3-32B', 'system_fingerprint': None, 'id': 'chatcmpl-0a365738-5955-4439-bbec-a002c2983368', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}
📝 模型返回内容: 邮件已经成功发送给张三，告诉他会议时间调整到了明天下午3点。如果有其他需要，请随时告诉我！
🔄 开始处理模型输出...
   输出类型: AIMessage
   effective_response_format: None
📦 返回标准消息格式
📦 模型输出处理完成，消息数: 1

结果解释： 这次模型基于完整的对话历史，生成了最终的用户响应，确认任务已完成。

第五阶段：流程结束

再次进入条件边决策函数model_to_tools：

决策输出：

📨 最后AI消息的工具调用数: 0
🔧 已处理的工具消息数: 0
🔚 无工具调用，流程结束，目标: __end__

结果解释： 由于没有待处理的工具调用，决策函数决定结束流程，Agent执行完成。

总结

整个Agent信息流可以概括为以下步骤：

1. 初始化：构建包含模型节点和工具节点的有向图
2. 模型推理：LLM分析用户请求并决定是否需要调用工具
3. 工具执行：执行具体的工具函数
4. 结果整合：将工具执行结果返回给模型进行最终响应
5. 流程结束：当没有更多工具需要调用时结束流程

这种设计使得Agent能够灵活地在模型推理和工具执行之间循环，完成复杂的多步任务。