开场白

终于开始源码了，看完了示例还是不知道 Agent 内部怎么执行任务的，DeepSeek 速速给我解释本项目怎么实现 Agent Loop。

flowchart TD
      START(["agent.run()"]) --> while
      subgraph loop
                    while -->summary
                    summary("summary()") --> llm.generate("llm.generate()")
                    llm.generate --> tool_call
                    tool_call("tool_call()") --> while("while(step &lt max_step)")
            end
            llm.generate --> END(["return response.content"])

这是循环的步骤的草图，省略了大量细节。

1. 总结
2. 思考
3. 执行

循环的终止条件是此次大模型回复不设计工具调用。这其实是 Agent 自主完成任务很关键的奥秘，在学这个项目之前，我就好奇怎么让模型判断任务完成，终止循环。

我们先补充一下 Agent运行时流程，再拆解一下这些关键节点。

主流程细节补充

主流程中有一些功能性步骤，为了确保主要步骤完整，所以草图把它们省略掉了，了解主流程后外面可以开始补充。

消息管理

消息管理涉及这些场景。

• 取消任务后清理不完整消息

• llm响应后，日志记录后，加入消息列表

• 工具调用完，异常处理完，日志记录完，加入消息列表

任务中断

一共三个地方有中断。

1. 在 while() 入口之后，取消任务之后清理不完整信息，返回字符串”用户中断任务”
2. 每次工具执行前
3. 每次工具执行后

最大步骤限制

• 进入循环前检查步骤数

• 循环完成时，增加步骤数

日志与打印

日志用的内部封装的函数，打印用的print()，以下行为顺序发生。

1. 日志，agent 进入循环前初始化，初始化日志
2. 打印，打印当前步骤，step/max_step
3. 日志，llm 调用前答应请求和工具列表
4. 先日志再打印，llm 调用后答应回复和 tool_call，没有工具调用就返回完成
5. 先日志再打印，工具执行后打印执行结果
6. 打印，工具消息添加完，打印步骤完成

异常处理

模型调用后异常捕获

• 重试异常
• 非重试异常

每次工具调用后异常捕获，用 ToolResult.success 判断调用成功与否

• 工具不在列表，这里用的判断，不是 try
• 工具执行出错

完整流程

完整版太长了，只能将就看，而且日志和打印还写不下。

flowchart TD
    %% ========== 节点定义 ==========
    agent_run(["agent.run()"])
    initial(["self.initial()"])
    step_check_and_loop("while(step &lt max_step)")

    check_cancel_1{"check_cancel()"}
    summary(["summary()"])
    llm_generate(["llm.generate()"])
    check_tool_call{"check_tool_call"}

    check_cancel_2{"check_cancel()"}
    tool_call(["tool_call()"])
    check_cancel_3{"check_cancel()"}
    step_add(["step+=1"])

    %% ========== 结果节点 ==========
    user_cancel_1(["return user cancel"])
    user_cancel_2(["return user cancel"])
    user_cancel_3(["return user cancel"])
    execute_error(["return execute error"])
    step_over(["return step over error"])
    finish(["return response.content"])

    %% ========== 边定义 ==========
    agent_run --> initial
    initial --> step_check_and_loop

    step_check_and_loop --> check_cancel_1
    check_cancel_1 --> summary
    summary --> llm_generate
    llm_generate --> check_tool_call

    check_cancel_1 --> user_cancel_1

    check_tool_call --> check_cancel_2
    check_cancel_2 --> tool_call
    tool_call --> check_cancel_3
    check_cancel_3 --> step_add
    step_add --> step_check_and_loop

    tool_call --> execute_error
    check_cancel_2 --> user_cancel_2
    check_cancel_3 --> user_cancel_3
    step_check_and_loop --> step_over
    check_tool_call --> finish

    %% ========== 子图 ==========
    subgraph loop
        check_cancel_1
        summary
        llm_generate
        check_tool_call
        check_cancel_2
        tool_call
        check_cancel_3
        step_add
        step_check_and_loop
    end

总结：

取消任务共三次，进入循环后，工具前，工具后。

日志与打印，所有操作需要可追溯性，日志记录优先于打印显示。

异常处理，模型与工具不可靠，必须设置异常处理；细粒度异常区分，不同处理策略。

关键节点

上下文压缩（ summary ）

从主流程就可以看出，上下文压缩在循环中，是同步执行的。可以从下面四个角度思考实现，怎么数，何时出发，怎么拆，怎么压缩。整体的分支与主流程如下。

async def _summarize_messages(self):

    # 触发压缩后把这个标志量设为 True ，下次省略检查
  if self._skip_next_token_check:
    self._skip_next_token_check = False
    return

    #判断是否要压缩
    estimated_tokens = self._estimate_tokens()
    should_summarize = estimated_tokens > self.token_limit or self.api_total_tokens > self.token_limit

    #按照 用户消息id 压缩
    new_messages = [self.messages[0]]  # Keep system prompt
  summary_count = 0
    for i, user_idx in enumerate(user_indices):
        #...

    # 替换消息列表
  self.messages = new_messages
  # 下次省略检查
  self._skip_next_token_check = True

怎么数

本地上下文总结

_estimate_tokens() 这个函数内部实现统计 total_token，用了 tiktoken 包下的 cl100k_base 规则，遍历 messages，对每条消息的 thinking，content，tool_call 全部做累加。

API 返回 token 统计

api_total_tokens 初始化为0，记录上次 api 调用后，模型供应商返回的 usage.total_tokens。这是模型供应商对上次响应提供的 token 统计，是你发的消息 token 以及 llm 响应的 token 的总和。

何时触发

只有当 _estimate_tokens() ， api_total_tokens 两者有一个超出限制时，才触发压缩。

另外这次触发压缩后，维护一个标志量使得下次循环一定不压缩。

为什么一定要两个条件来维护？

不同的 token 分割标准不统一，自己维护的和运营商提供的可能有出入，如果只用 tiktoken 的超出窗口限制会报错。

怎么拆

原本的消息结构如下：

 system ─→ user1 ─→ assistant ─→ tool ─→ assistant ─→ tool ─→ assistant

    ─→ user2 ─→ assistant ─→ tool ─→ assistant

    ─→ user3 ─→ assistant (进行中)

策略：

找到所有 user message，把两个 user message 之间的消息全作为执行过程，只压缩执行过程，保留 user message。

压缩后：

system → user1 → [摘要: 第1轮执行了write/read工具, 创建了hello.py]
         → user2 → [摘要: 第2轮用bash执行并验证了输出]
         → user3 → assistant (当前进行中，不动)

怎么压缩

这是最核心的一步，流程如下。

1. 找出所有 user message，记录 idx
2. 复制 user message，对 两条 user message 之间的消息放到一个列表里，执行压缩
3. 调用 self._create_summary(execution_messages, i + 1)，使用 llm 和特定的系统提示词执行压缩
4. 封装压缩后的消息，添加到 new_messages

摘要的要求,放在做摘要的 llm 的系统提示词里的：

要求	目的
聚焦完成任务和调用的工具	保留做了什么的信息
保留关键结果和重要发现	不丢重要上下文
1000 字以内	压缩比
英文	节省 token
不要包含 user 内容，只总结 Agent 执行	避免信息重复

个人优化理解：

此压缩架构感觉可以异步并行压缩。

LLM 调用（ llm_generate ）

直接调用框架层 LLMClient.generate()。

此处用到了策略模式+依赖注入

agent 无需关心 llm provider，LLMClient 会根据 config.yaml 判断调用 OpenAIClient 或者 AnthropicClient。

classDiagram

%% Context
class LLMClient {
    &lt&ltContext>>
    - _client
}

%% Strategy Interface
class BaseLLMClient {
    &lt&ltStrategy>>
}

%% Concrete Strategies
class AnthropicClient {
    &lt&ltConcreteStrategy>>
}

class OpenAIClient {
   &lt&ltConcreteStrategy>>
}

%% Relationships
LLMClient --> BaseLLMClient : uses >
BaseLLMClient --> AnthropicClient : inherits
BaseLLMClient --> OpenAIClient : inherits

%% Agent injection
Agent --> LLMClient : inject

这里的策略工厂有什么问题？

工厂模式是根据参数返回不同产品。策略模式是提供统一接口，调用方无感知，运行时切换。

Java 的策略模式一般设计策略接口，然后实现接口，利用多态，此处是组合，还是不太一样。Java 的编译器就能检查失败，Python 运行期才可以。 Java 的一般搭配注册表使用，扩展性更好，新增实现不用改工厂，这里用 if/else 硬编码，违反开闭原则。

LLMClientBase.generate() 具体的调用过程如下：

1. 转换格式（messages，tools）
2. 发起调用
3. 解析响应

OpenAIClient

系统与 OpenAI 格式映射如下。

内部 role	OpenAI role	特殊处理
system	system	放在 messages 数组里（不像 Anthropic 单独抽出来）
user	user	直接映射
assistant	assistant	转 tool_calls 的 arguments 为 JSON 字符串，保留 reasoning_details
tool	tool	带 tool_call_id 关联回调

转换消息

_convert_messages()

list[Message]->list[dict]

  # 输入: list[Message] (内部统一格式)
  Message(role="system",  content="You are a helpful...")
  Message(role="user",    content="帮我创建文件")
  Message(role="assistant", content=None, tool_calls=[...], thinking="我需要用...")
  Message(role="tool",    content="写入成功", tool_call_id="call_xxx")

  # 输出: list[dict] (OpenAI API 格式)
  [
    {"role": "system", "content": "You are a helpful..."},
    {"role": "user", "content": "帮我创建文件"},
    {
      "role": "assistant",
      "content": null,
      "tool_calls": [
        {
          "id": "call_xxx",
          "type": "function",
          "function": {
            "name": "write_file",
            "arguments": '{"path": "hello.py", "content": "..."}'  // JSON 字符串
          }
        }
      ],
      "reasoning_details": [{"text": "我需要用..."}]    // ← 保留思考链
    },
    {
      "role": "tool",
      "tool_call_id": "call_xxx",
      "content": "写入成功"
    }
  ]

转换工具列表

_convert_tools()

list[Tool]->list[dict]

  # 输入: Tool 对象列表
  tools = [ReadTool(), WriteTool(), BashTool()]

  # 每个 tool 调用 to_openai_schema()
  # Tool(基类).to_openai_schema():
  #   return {
  #       "type": "function",
  #       "function": {
  #           "name": self.name,
  #           "description": self.description,
  #           "parameters": self.parameters,
  #       },
  #   }

发起请求

_make_api_request()

请求头如下

  params = {
      "model": self.model,                            # "MiniMax-M2.5"
      "messages": api_messages,                       # 已转换的 OpenAI 格式消息
      "extra_body": {"reasoning_split": True},        # 分离思考过程
  }
  if tools:
      params["tools"] = self._convert_tools(tools)    # 已转换的 OpenAI 格式工具

  response = await self.client.chat.completions.create(**params)
  #                     ↑ AsyncOpenAI 实例 (openai SDK)

reasoning_split 变量用来指定分离 thinking 过程，方便保证思考链。

解析响应

_parse_response()

openai response-> system LLMResponse

  # 输入: OpenAI ChatCompletion 原始响应
  response.choices[0].message
      .content              → "好的，文件已创建..."      → LLMResponse.content
      .reasoning_details    → [{"text": "我需要..."}]    → LLMResponse.thinking
      .tool_calls           → [{id, function, args}]     → LLMResponse.tool_calls
      .usage                → {prompt, completion, total} → LLMResponse.usage

OpenAI 返回的 tool_call.function.arguments 是 JSON 字符串，需要 json.loads() 解析成 dict，这样才能交给 tool.execute(**dict)。

AnthropicClient

内部 role	Anthropic role	特殊处理
system	（无）	不作为 messages 元素，单独抽出来作为请求级 system 参数
user	user	直接映射
assistant	assistant	thinking / tool_calls 拆成 content block 数组： thinking block + text block + tool_use block， tool_use 的 input 保持 dict 原样，不转 JSON 字符串
tool	user	伪装成 user 角色，用 tool_result block 包装，通过 tool_use_id 关联回调

转换消息

_convert_messages()

Anthropic 格式的 system prompt 要从消息列表里拿出来。

    if msg.role == "system":
      system_message = msg.content    # 单独提出来，不放进 messages 数组

assistant message（thinking+tool call）

  # 转换前（内部）
  Message(role="assistant", content="好的",
          thinking="我需要创建一个文件",
          tool_calls=[ToolCall(id="call_001", function=FunctionCall(
              name="write_file", arguments={"path": "hello.py", "content": "print('hello')"}
          ))])

  # 转换后（Anthropic）
  {
      "role": "assistant",
      "content": [
          {"type": "thinking",  "thinking": "我需要创建一个文件"},          # thinking block
          {"type": "text",      "text": "好的"},                           # 文本 block
          {"type": "tool_use",  "id": "call_001",                         # tool_use block
                                "name": "write_file",
                                "input": {"path": "hello.py", "content": "print('hello')"}},
      ]
  }

三个 block 的顺序是 thinking → text → tool_use，对应模型的思维过程：先想、再表达、再行动。

如果 assistant 没有 thinking 也没有 tool_calls（第 160 行）——纯文本回复，直接映射：

{“role”: “assistant”, “content”: “文件已创建完毕”}

tool message

  # 转换前（内部）
  Message(role="tool", content="文件写入成功", tool_call_id="call_001")

  # 转换后（Anthropic）
  {
      "role": "user",                     # ← role 变成 user !
      "content": [
          {
              "type": "tool_result",
              "tool_use_id": "call_001",
              "content": "文件写入成功"
          }
      ]
  }

Anthropic 协议里没有 tool 这个角色。工具执行结果必须以 user 角色 + tool_result block 的形式提交。设计意图是：工具执行结果是外部世界给出的事实，和用户说的话属于同一性质。

转换工具列表

_convert_tools()

Tool 对象的转换相对简单，调用 tool.to_schema()：

  # Tool.to_schema() 产出
  {
      "name": "write_file",
      "description": "Write content to a file",
      "input_schema": {
          "type": "object",
          "properties": {
              "path": {"type": "string", "description": "File path"},
              "content": {"type": "string", "description": "Content to write"}
          },
          "required": ["path", "content"]
      }
  }

发起请求

_make_api_request()

这是请求头

  params = {
      "model": self.model,                    # "MiniMax-M2.5"
      "max_tokens": 16384,                    # Anthropic 协议强制要求
      "messages": api_messages,               # 阶段一的产物
  }

  if system_message:
      params["system"] = system_message       # system 作为独立参数

  if tools:
      params["tools"] = self._convert_tools(tools)

  response = await self.client.messages.create(**params)
  #                ↑ AsyncAnthropic() 实例

解析响应

_parse_response()

三种 block 的处理

  text_content = ""
  thinking_content = ""
  tool_calls = []

  for block in response.content:
      if block.type == "text":
          text_content += block.text                 # → 拼接文本

      elif block.type == "thinking":
          thinking_content += block.thinking          # → 拼接思考

      elif block.type == "tool_use":
          tool_calls.append(ToolCall(                 # → 构造 ToolCall
              id=block.id,
              type="function",
              function=FunctionCall(
                  name=block.name,                    # "bash"
                  arguments=block.input,               # {"command": "python hello.py"}
                  # ↑ 注意：Anthropic 的 input 已经是 dict，无需 json.loads
              ),
          ))

token 加总

Anthropic 把 input tokens 拆成三部分，需要手动合计。

  input_tokens       = 1520   # 消息本身的 token
  cache_read_tokens  = 0      # 命中 prompt cache 读到的 token
  cache_creation_tokens = 0   # 写入 prompt cache 的 token

  # 合计为 prompt_tokens
  prompt_tokens = 1520 + 0 + 0 = 1520

  # 构造统一格式
  TokenUsage(prompt_tokens=1520, completion_tokens=85, total_tokens=1605)

对比

维度	AnthropicClient	OpenAIClient
SDK	AsyncAnthropic	AsyncOpenAI
API 方法	messages.create()	chat.completions.create()
system prompt	独立参数	放 messages 数组
assistant thinking	{“type”:”thinking”} block	reasoning_details 字段
tool 定义	input_schema	parameters + 外层 type:function
tool args	dict 原生	JSON 字符串
tool result	role=”user” + tool_result block	role=”tool” 独立角色
max_tokens	必须指定	不需要
thinking 分离	自动（block type 区分）	需要 reasoning_split: True

两个 Client 的 generate() 方法结构一模一样：

_prepare_request → _make_api_request (带 retry) → _parse_response。所有差异被封装在三个 hook 方法里，对外暴露统一的 LLMResponse。

工具调用（ tool_call ）

内部格式已经统一，所以可以统一调用工具。

Agent运行时中的调用工具

  # agent.py 第 430-501 行
  for tool_call in response.tool_calls:
      function_name = tool_call.function.name     # "write_file"
      arguments = tool_call.function.arguments     # {"path": "hello.py", "content": "print('hello')"}
                                                   #   ↑ 已经是 dict，不管哪个 provider

      tool = self.tools[function_name]             # 从 {name: Tool} 字典查找
      result = await tool.execute(**arguments)     # 直接展开调用
                                                   #   ↑ **dict 拆成关键字参数

这里的 result 是 ToolResult，后续把 result 转换成 ToolMessage 并追加到消息列表也是 agent 层做的。

Tool 类属性介绍

Tool 类下有 name / description / parameters / execute(**kwargs)

parameters 和 args 有什么区别？

parameters 是对参数的定义与解释，具体实例如下。kwargs是实际执行时传入的所需数据。可以说，parameters 是静态的，kwargs 时动态的。

      @property
      def parameters(self) -> dict:
          return {
              "type": "object",
              "properties": {
                  "location": {"type": "string", "description": "City name"},
                  "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
              },
              "required": ["location"],
          }

tool.execute() 执行过程

Tool 是一个基类，其子类要去自己实现 execute() 方法，当然其提供了 to_schema()/to_openai_schema() 这两个可以复用的方法。当然实现上没有用抽象类的框架，比较简单但不安全。

Tool.execute() 会报错，强制实现。

  # tools/base.py
  class Tool:
      async def execute(self, *args, **kwargs) -> ToolResult:
          raise NotImplementedError