跳到主要内容

执行与上下文:无状态 Agent 怎么跑起来

这章端到端追一次 get_response,重点理解一个反直觉但很关键的设计:Agent 是无状态的——它不持有线程、不持有「我能找谁」,这些全部在运行时由 Agency 注入。这让同一个 agent 实例能被多个 agency 复用。

3.5 核心设计:无状态 Agent + 运行期注入​

它要解决的小问题。 如果把「对话历史、子 agent 列表、send_message 工具」直接挂在 Agent 实例上,那这个 agent 就和某一个 agency、某一次会话绑死了,没法复用、没法并发跑多个会话。

思路。 把所有「随会话/随 agency 变化的东西」抽出来,放进运行期对象,执行时作为参数传进去。源码 docstring 写得很明确:Agents are stateless. Agency-specific resources like thread managers, subagent mappings and shared instructions are provided at runtime via AgencyContext(src/agency_swarm/agent/core.py:74-76)。

这里有三个上下文对象,各管一摊,别混:

对象谁创建装什么文件
AgencyContextAgency.get_agent_context一次执行的「环境」:thread_manager、该 agent 的 runtime_state、持久化回调、shared_instructionsagent/context_types.py:36-63
AgentRuntimeStateAgency 构造期,每 agent 一个可变的运行期状态:subagents、send_message_tools、handoffs、待处理集合、并发管理器agent/context_types.py:14-33
MasterContext每次 run 由 prepare_master_context 造传给 SDK Runner 的共享上下文:所有 agent、user_context、thread_manager、运行期状态表、流式标志context.py:24-63

一句话记忆:AgencyContext 是「外层封装,交给 Agent.get_response」;MasterContext 是「内层,塞进 SDK Runner、工具能在 wrapper.context 上拿到」。

 Agency.get_agent_context(name)
   └── AgencyContext { thread_manager, runtime_state, shared_instructions, callbacks }
          │ 传给 agent.get_response(..., agency_context=ctx)
          ▼
   Execution.get_response()
          │ prepare_master_context(agent, override, agency_context)
          ▼
   MasterContext { agents{全部}, user_context, thread_manager,
                   agent_runtime_state{全部}, _is_streaming, ... }
          │ 包进 RunContextWrapper,交给 agents.Runner
          ▼
   工具 / send_message 在 wrapper.context 上读到 MasterContext

3.6 一次 get_response 端到端​

下面按 Execution.get_response(src/agency_swarm/agent/execution.py:73-402)的真实顺序走一遍。

① 入口与上下文。 agency.get_response 先选 recipient(默认第一个 entry point),取 AgencyContext,再委派给 agent.get_response,见 agency/responses.py:179-215。Agent.get_response 若没拿到 context 就造一个最小 context(standalone 用法),见 agent/core.py:413-427。

② setup 与历史。 setup_execution 校验「agent↔agent 委派必须有 agency」并暂存原始 instructions 以便事后还原,见 agent/execution.py:107-109 + execution_helpers.py:304-317。然后处理消息和文件附件,生成一个唯一 current_agent_run_id(execution.py:121)。

③ 准备投喂 Runner 的历史。 MessageFormatter.prepare_history_for_runner(messages/message_formatter.py:256-320)做四件事:取出「这一对 agent 的历史」、给新消息打元数据并存进表、校验历史协议、出库前剥掉 agency 私有元数据(因为 OpenAI 不认 agent/callerAgent 这些字段),见 message_formatter.py:315-316。

④ 跑推理循环。 run_with_guardrails(包在守卫逻辑里)最终调用 SDK 的 agents.Runner 执行标准「模型↔工具」循环,见 execution.py:212-228。Agency Swarm 不重写这个循环——它把推理交给 SDK,自己只管循环外的编排和记录。这是它「在 SDK 之上而非替代 SDK」定位的核心证据。

还有一条快路:如果这是首条消息且命中对话启动器缓存(conversation starter cache),框架会跳过真正调模型,直接用缓存的 run items 拼出 RunResult,见 execution.py:212-250。这是延迟优化,细节见第 4 章。

⑤ 写回历史。 拿到 run_result 后,遍历 new_items,逐条转成可存格式、补 citations、打 agency 元数据(agent、callerAgent、agent_run_id、parent_run_id、run_trace_id、history_protocol),过滤掉不需要持久化的类型,最后 thread_manager.add_messages 落表,见 execution.py:284-347。

⑥ cleanup。 finally 里还原 instructions、清理附件,见 execution.py:391-402。

# 示意,非源码:get_response 的骨架
async def get_response(self, message, sender_name, agency_context, ...):
    original = setup_execution(self.agent, sender_name, agency_context, ...)   # 暂存 instructions
    try:
        items = await attachment_manager.process_message_and_files(message, ...) # 处理输入
        run_id = f"agent_run_{uuid4().hex}"
        history = MessageFormatter.prepare_history_for_runner(items, ...)        # 取历史+打标+剥元数据
        ctx = prepare_master_context(self.agent, override, agency_context)       # 造 MasterContext
        run_result = await run_with_guardrails(agent=self.agent, history=history, # 调 SDK Runner
                                               master_context=ctx, ...)
        save_items = [打元数据(i) for i in run_result.new_items]                 # 给产出打标
        agency_context.thread_manager.add_messages(save_items)                   # 落表(顺手持久化)
        return run_result
    finally:
        self.agent.instructions = original                                       # 还原

3.7 send_message 里那份「临时上下文」​

回到上一章的 send_message:当 sender 委派给 recipient 时,recipient 也要跑一次 get_response,它也需要一份 AgencyContext。但工具 on_invoke_tool 手上只有 SDK 给的 wrapper.context(一个 MasterContext),没有 Agency 实例。

怎么办:用 MasterContext 里的零件,临时拼一个最小 agency 上下文。 _create_recipient_agency_context(tools/send_message.py:268-312)定义了一个内联的 MinimalAgency,把 wrapper.context 里的 agents、user_context、agent_runtime_state、shared_instructions 转手包成一份 AgencyContext,复用同一个 thread_manager(send_message.py:307)——所以子对话和主对话写进的是同一张历史表。

这也解释了为什么子 agent 的消息能和主对话共存在一张表里却互不污染:靠的是 callerAgent 标签(下一章详述)。

3.8 跨 agent 的调用树与成本归集​

两个贯穿始终的 id 让多 agent 调用可追踪:

  • agent_run_id —— 本次 agent 执行的唯一 id(execution.py:121)。
  • parent_run_id —— 调用方的执行 id;send_message 把工具调用 id(tool_call_id)当作 recipient 的 parent_run_id 传下去(send_message.py:427、510),从而把「谁委派了谁」串成一棵树。

成本归集。 子 agent 跑完后,它的 raw_responses 连同子 agent 自己的模型名被收集到 wrapper.context._sub_agent_raw_responses(send_message.py:481-492)。父 Execution 在收尾时把这些 (model_name, response) 元组转存到 run_result 上,供后续按模型分别算价,见 execution.py:252-273。这让一次「manager 用 GPT-A、analyst 用 GPT-B」的混合委派能正确分模型计费。

接下来​

你已经多次看到 agent / callerAgent 这对标签和「同一张 thread 表」。下一章 03-threads-and-messages.md 专讲这张表怎么设计,以及它如何用两个字段同时表达「用户线程」和「每一对 agent 的子对话」。