Soup's Blog

Back
Agent实战（四）深入理解Agent从输入到输出的信息流Blur image
19 min
中文
agent

Agent实战（四）深入理解Agent从输入到输出的信息流

Agent实战

Agent信息流概述#

代码开源Github ↗

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

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

Graph初始化过程#

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

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

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

1. 添加模型节点：

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

2. 添加工具节点：

graph.add_node("tools", tool_node)
bash

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}")
bash

输出如下：

🏁 确定入口节点: model
bash

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}")
bash

输出如下：

🔄 确定循环入口节点: model
bash

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}")
bash

输出如下：

🚪 确定循环出口节点: model
bash

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}")
bash

输出如下：

🔚 确定最终出口节点: __end__
bash

7. 添加起始边：

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

输出如下：

🔗 添加起始边: START -> model
bash

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

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

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

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

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

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}")
bash

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

输出如下：

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

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

流程图表示：

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

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}")
bash

输出如下：

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

结果解释： 模型节点执行完毕后可以选择调用工具（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中间件边")
bash

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})
bash

Agent运行过程#

第一阶段：初始模型调用#

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

ModelRequest对象结构（简化）：

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

模型绑定过程输出：

结果解释： 这里展示了工具如何被绑定到模型，工具的描述信息会通过API参数传递给大模型，而不是直接注入到prompt中。这里需要注意，为什么在下面的模型调用输出的content为空，而将工具信息都存储在additional_kwargs中，这是因为通过使用 bind_tools 方法时：

bound_model = request.model.bind_tools(
                final_tools, tool_choice=tool_choice, **request.model_settings
            )
bash

这个过程告诉模型：

  • 有哪些工具可以使用（final_tools就是工具列表）
  • 每个工具的名称、描述和参数格式
  • 工具调用的规范格式

现代大语言模型（如Qwen3-32B）被训练成能够生成特定结构的输出。当您绑定工具后，模型知道需要按照特定格式返回工具调用，结构如下：

{
  "tool_calls": [
    {
      "id": "call_xxx",
      "function": {
        "name": "工具名称",
        "arguments": "{\"参数1\": \"值1\", \"参数2\": \"值2\"}"
      },
      "type": "function"
    }
  ]
}
bash

模型调用输出：

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

amodel_node返回：

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

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

第二阶段：工具调用决策#

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

决策过程输出：

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

决策逻辑代码：

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
python

代码解释：

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

决策结果：

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

第三阶段：工具执行#

调用我们创建的send_email方法：

工具执行输出：

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

工具返回结果：

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

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

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

消息历史：

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

模型调用输出：

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

第五阶段：流程结束#

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

决策输出：

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

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

总结#

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

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

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

Agent实战（四）深入理解Agent从输入到输出的信息流
http://www.soupcola.top/blog/agent_blogs/agent_blogs-4
Author Soup Cola
Published at 2026年2月12日
Copyright CC BY-NC-SA 4.0

Agent实战（五）MCP的demo

Agent实战（三）Agent的短期记忆