AssistantAgent:单个 agent 的心脏

上一章是"消息怎么流";这一章是"一个 agent 收到消息后,在自己肚子里怎么想、怎么用工具"。AssistantAgent 是 AgentChat 的主力 agent,绝大多数用法都用它。

1. 它要解决的小问题

给定一段对话,一个 agent 要能:看上下文 → 决定是直接回话还是调工具 → 调完工具拿到结果后,要么把结果当答案、要么再想一轮给个自然语言总结。这就是"工具调用循环"(tool-use loop),也是当代 agent 的标准心脏。

2. 直觉:一条带分支的流水线

AssistantAgent 的核心方法是 on_messages_stream——一个异步生成器:边处理边吐出中间事件(工具请求、工具结果、思考),最后吐一个 Response 作为收尾。源码里把它清晰地标成了 STEP 1~5(_assistant_agent.py:932-1011):

STEP 1  新消息进 model_context        (_add_messages_to_context)
STEP 2  查记忆、把命中的塞进上下文      (_update_model_context_with_memory)
STEP 3  生成 message_id (用于关联流式分片)
STEP 4  第一次调 LLM                    (_call_llm)
STEP 5  处理 LLM 输出                    (_process_model_result)
          ├─ 纯文本？ → 直接当最终答案返回，结束
          └─ 有工具调用？ → 进入"工具循环"

关键:model_context 是 agent 的"短期记忆",跨多次调用持续存在。所以类文档反复警告:每次只把新消息传进来,别把整段历史重复喂进去(_assistant_agent.py:109-115 的 .. attention::)。

3. 图示:工具调用循环

          ┌──────────── _process_model_result 的循环 (最多 max_tool_iterations 轮) ─────────────┐
          │                                                                                    │
  LLM 结果 ─┤                                                                                    │
          │  content 是 str ? ──是──▶ 返回 TextMessage / StructuredMessage  ── 结束(不管还剩几轮) │
          │      │否                                                                            │
          │      ▼                                                                              │
          │  content 是 [FunctionCall]                                                          │
          │  ① 吐 ToolCallRequestEvent                                                          │
          │  ② 并发执行所有工具 (asyncio.gather) → 吐 ToolCallExecutionEvent                     │
          │  ③ 是 handoff 调用? ──是──▶ 返回 HandoffMessage ── 结束                              │
          │  ④ 还没到最后一轮? ──是──▶ 再调一次 LLM,带着工具结果,回到循环顶                       │
          └────────────────────────────────────────────────────────────────────────────────────┘
                                       │ 循环结束后
                                       ▼
          reflect_on_tool_use ? ──是──▶ 再调一次 LLM 生成自然语言总结 (reflect)
                              └──否──▶ 直接把工具结果格式化成 ToolCallSummaryMessage 返回

怎么读: 模型一旦回纯文本就立刻收尾;只要还在回工具调用且没到 max_tool_iterations 上限,就继续"调工具→再问模型"。

4. 原理演示:循环的本质

# 示意,非源码:工具循环的骨架
async def process(model_result, max_iters, reflect):
    current = model_result
    executed = []
    for i in range(max_iters):
        if isinstance(current.content, str):       # 模型回了文本 → 完成
            yield Response(TextMessage(current.content)); return
        # 否则 current.content 是一串 FunctionCall
        yield ToolCallRequestEvent(current.content)
        executed = await run_all_tools(current.content)   # 并发执行
        yield ToolCallExecutionEvent([r for _, r in executed])
        if is_handoff(current):                    # 交接给别的 agent
            yield Response(HandoffMessage(...)); return
        if i == max_iters - 1:                     # 最后一轮,跳出去做总结
            break
        current = await call_llm_again(executed)   # 带着结果再问模型
    if reflect:
        yield await reflect_with_llm(executed)     # 再问一次,要自然语言
    else:
        yield summarize(executed)                  # 直接拼工具结果当答案

重点看 if isinstance(current.content, str):这是"模型不再要工具、给文本了"的判定,也是循环唯一的"正常出口"。

5. 真实实现

入口:on_messages 只是把流跑完、捞出那个 Response(_assistant_agent.py:882-899)。真正的逻辑在 on_messages_stream(:901),它把状态局部化后依次跑 STEP 1~5。

调 LLM:_call_llm(_assistant_agent.py:1055)。它做三件事——把 system message + 上下文拼成 llm_messages、收集所有 workbench 的工具 + handoff 工具、按 model_client_stream 走流式或一次性 create:

# _assistant_agent.py:1085-1088 (_call_llm 节选)
all_messages = await model_context.get_messages()
llm_messages = cls._get_compatible_context(model_client=model_client, messages=system_messages + all_messages)
tools = [tool for wb in workbench for tool in await wb.list_tools()] + handoff_tools

工具循环:_process_model_result(_assistant_agent.py:1117)。循环体就是 for loop_iteration in range(max_tool_iterations)(:1149)。文本出口在 :1151,工具并发执行用 asyncio.gather(:1200),并通过一个 asyncio.Queue 把工具执行期间的流式事件实时吐出来(:1194-1228)。

"模型停止要工具就提前结束" 这条规则在类文档里说得很清楚:"As soon as the model stops making tool calls, the agent will stop executing tool calls and return the result as the final response"(_assistant_agent.py:218-223)。代码层面就是 §4 那个 isinstance(content, str) 分支会 return,无视还剩几轮。

6. 三个值得记的设计点

(1) max_tool_iterations —— 多步工具工作流。 默认 1:执行一次工具就收尾。设大于 1,agent 能"调工具→看结果→再调工具"地连做多步,直到模型给文本或到上限(_assistant_agent.py:146 的类文档 + :1149 的循环)。构造时校验 >= 1(:851-855)。

(2) reflect_on_tool_use —— 工具结果要不要再过一遍模型。

取值	行为	默认
False	工具结果直接格式化成 ToolCallSummaryMessage 当答案	未设结构化输出时默认
True	带着工具结果再调一次 LLM,产出自然语言/结构化最终答案	设了 output_content_type 时默认

这个默认逻辑在 :842-848:有 output_content_type 就默认反思,否则默认不反思。两条路分别走 _reflect_on_tool_use_flow(:1408)和 _summarize_tool_use(:1317)。

(3) handoff —— 把控制权交给别的 agent。 handoff 本质是"特殊的工具":构造时把每个 handoff 变成一个 handoff_tool(:794-808)。模型"调用"这个工具就表示要交接。_check_and_handle_handoff(:1327)在工具执行后检查:若有 handoff 调用,产出一个 HandoffMessage(带 target 和上下文)并结束。注意它只执行第一个 handoff,多个会告警(:1354-1362)。这个 HandoffMessage 正是第 3 章 Swarm 团队用来路由"下一个谁发言"的信号。

7. 关键细节 / 坑

• 不是线程安全的: 类文档明确"not thread-safe or coroutine-safe",不能并发调它的方法(_assistant_agent.py:117-120)。
• 思考(thought)单独成事件: 推理模型的隐藏思考会被抽出来当 ThoughtEvent 吐出,并存进上下文(:973-988)。
• 工具 vs workbench 二选一: 给了 workbench 就不能再给 tools(:827-829)。没给 workbench 时,框架自动把 tools 包进一个 StaticStreamWorkbench(:835)。所以工具的统一抽象其实是 workbench,tools 只是糖。
• 工具结果默认就是答案: reflect_on_tool_use=False(默认)时,工具的返回值会直接成为 agent 给别人的回复,所以类文档提醒"pay close attention to how the tools' return values are formatted"(:148-156)。

8. 小结

AssistantAgent = 一条"上下文→记忆→LLM→工具循环→反思/总结"的异步流水线。它产出三类最终消息:TextMessage(或 StructuredMessage)、ToolCallSummaryMessage、HandoffMessage(produced_message_types,:861-871)。这三类消息正是下一章团队编排的输入。