数据截至 2026-08-16(上游 commit 3309bf4e416f)

01 — BaseAgent:主循环、状态机、记忆与卡死检测

这章讲什么: 把 app/agent/base.py(196 行)拆开,看一个 agent 最底层需要哪几件东西, 以及 OpenManus 各自用什么办法解决。读完你应该能自己手写一个同构的循环。

---

1. 要解决的小问题

LLM 一次调用只会「说一段话」。要让它连续做事,你至少得回答四个问题:

问题	OpenManus 的答案
什么时候停?	步数上限 max_steps,或状态变成 FINISHED
上一步做了啥,下一步怎么知道?	一条不断追加的消息列表 Memory
中途出错了怎么办?	用异步上下文管理器守住状态字段,异常时置 ERROR
模型开始鬼打墙怎么办?	数最近几条 assistant 消息是否一字不差重复

---

2. 主循环:一个带上限的 while

2.1 图示

  run(request)
      │
      ├─ 状态必须是 IDLE,否则直接抛 RuntimeError
      ├─ 把 request 作为 user 消息写进 Memory
      │
      ▼
  ┌──────────────── while ────────────────┐
  │ 条件: step < max_steps 且 state≠FINISHED│
  │   step += 1                            │
  │   result = await self.step()  ← 抽象   │
  │   if is_stuck(): 往提示里塞一句警告    │
  └───────────────┬────────────────────────┘
                  ▼
      步数用完 → 追加 "Terminated: Reached max steps"
                  ▼
      SANDBOX_CLIENT.cleanup()  → 返回逐步结果的拼接

2.2 原理演示

# 示意,非源码 —— agent 主循环的最小形态
async def run(self, request: str) -> str:
    self.memory.add_message(user_message(request))   # 把任务写进记忆
    results = []
    while self.current_step < self.max_steps and self.state != "FINISHED":
        self.current_step += 1
        results.append(await self.step())            # step 由子类实现
    return "\n".join(results)

重点看:循环本身完全不知道「工具」「模型」是什么——step() 是抽象方法。 这就是为什么换一套 step() 实现就能换一种智能体范式。

2.3 真实实现

BaseAgent.run 在 app/agent/base.py:116-154。循环体在 136-147 行:

while (
    self.current_step < self.max_steps and self.state != AgentState.FINISHED
):
    self.current_step += 1
    step_result = await self.step()
    if self.is_stuck():
        self.handle_stuck_state()

收尾有两处值得记住:

• 步数耗尽时把 current_step 归零、状态设回 IDLE,并往结果里追加一句 Terminated: Reached max steps(app/agent/base.py:149-152)。
• 无论怎么结束,最后都会 await SANDBOX_CLIENT.cleanup()(app/agent/base.py:153)—— 也就是说每跑完一次 run(),共享的 Docker 沙箱容器就会被销毁。

默认步数上限是分层设定的:BaseAgent 给 10(app/agent/base.py:40), ToolCallAgent 提到 30(app/agent/toolcall.py:36),Manus 又压回 20 (app/agent/manus.py:51)。

---

3. 状态机:四个状态 + 一个上下文管理器

3.1 四个状态

AgentState 是个字符串枚举(app/schema.py:32-38):

状态	含义	谁会设它
IDLE	空闲,可以开跑	初始值;run() 结束后被还原
RUNNING	正在循环里	state_context 进入时
FINISHED	任务完成,该退出循环	terminate 工具触发,或 token 超限
ERROR	出错了	state_context 捕获异常时

3.2 为什么用上下文管理器

直觉:状态字段最怕「改了没改回来」。state_context 是个 @asynccontextmanager(app/agent/base.py:58-82),写法是:

previous_state = self.state
self.state = new_state
try:
    yield
except Exception as e:
    self.state = AgentState.ERROR   # 出错先标 ERROR
    raise e
finally:
    self.state = previous_state     # 无论如何还原

这里有一个必须知道的后果。 finally 会无条件把状态还原成进入前的值(通常是 IDLE), 所以循环里被置成 FINISHED 或 ERROR 的状态,在 run() 返回之后就不存在了—— 调用方拿到的 agent.state 恒为 IDLE。

这不是纸上谈兵:PlanningFlow 在 app/flow/planning.py:128 用 executor.state == AgentState.FINISHED 来判断「智能体想提前收工」, 按上面的代码路径这个条件不会成立(见 04 章)。

3.3 另一个跨调用的副作用

current_step 只在步数耗尽时才被清零(app/agent/base.py:150)。 如果智能体是靠 terminate 正常结束的,current_step 保持在结束时的值。 于是对同一个实例第二次调用 run(),循环是从上次的步数接着数的—— 可用步数会越来越少,直到某次撞上 max_steps 才归零。

对 main.py 这种「一个实例只跑一次」的用法无害;对 PlanningFlow 那种 「同一个 Manus 实例被 run() 很多次」的用法就有实际影响。

---

4. 记忆:一条会滑窗的消息列表

4.1 数据结构

Memory 简单到只有一个列表加一个上限(app/schema.py:159-175):

class Memory(BaseModel):
    messages: List[Message] = Field(default_factory=list)
    max_messages: int = Field(default=100)

add_message 追加后,若超过 100 条就切掉最前面的:self.messages[-self.max_messages:] (app/schema.py:167-168)。没有摘要、没有向量检索、没有分层记忆—— 这是 OpenManus 刻意选的极简路线。

4.2 消息的四种角色

Message 用工厂方法建,四个角色对应四个类方法(app/schema.py:99-129):

工厂方法	角色	典型用途
Message.user_message	user	你的任务;每步注入的 next_step_prompt
Message.system_message	system	系统提示;MCP 服务器下发的 instructions
Message.assistant_message	assistant	模型的自然语言输出
Message.tool_message	tool	工具执行结果,必须带 tool_call_id

还有一个专门的 Message.from_tool_calls(app/schema.py:131-156), 把模型返回的 tool_calls 对象转成可存储的 dict 结构。

BaseAgent.update_memory 用一张字典把角色映射到工厂方法 (app/agent/base.py:102-114),角色不认识就抛 ValueError——避免手滑写错角色名。

4.3 一个要留心的地方

滑窗是按条数从头砍的。OpenAI 兼容接口要求:一条带 tool_calls 的 assistant 消息, 必须紧跟着对应 tool_call_id 的 tool 消息。如果窗口边界正好落在这一对中间, 砍掉 assistant 而留下 tool,请求就会被服务端拒绝。代码里没有针对这一点的保护 (app/schema.py:163-175)(inferred:这是从截断逻辑推出的风险,仓库里没有对应测试)。

---

5. 卡死检测:数重复

5.1 思路

模型偶尔会陷进「同一句话说三遍」的循环。OpenManus 的对策朴素得可爱: 看最后一条消息的内容,在之前的 assistant 消息里出现过几次。

5.2 图示

   记忆(从新到旧)
   [-1] assistant: "我需要先查看文件"   ← 拿它当基准
   [-2] tool:      "..."
   [-3] assistant: "我需要先查看文件"   ← 命中 +1
   [-4] tool:      "..."
   [-5] assistant: "我需要先查看文件"   ← 命中 +2  → ≥ 阈值 2,判定卡死

5.3 真实实现

is_stuck 在 app/agent/base.py:170-186,核心一句:

duplicate_count = sum(
    1
    for msg in reversed(self.memory.messages[:-1])
    if msg.role == "assistant" and msg.content == last_message.content
)
return duplicate_count >= self.duplicate_threshold

阈值 duplicate_threshold 默认 2(app/agent/base.py:43)。

命中后走 handle_stuck_state(app/agent/base.py:163-168):把一句 「Observed duplicate responses. Consider new strategies…」前缀拼进 next_step_prompt。

5.4 两个细节

• 比对的是 content,不是 tool_calls。 模型如果每轮都返回空 content 只带工具调用 (这在 function calling 下很常见),这套检测不会触发——因为 if not last_message.content: return False(app/agent/base.py:176-177)。
• 警告会累积。 handle_stuck_state 是 f"{stuck_prompt}\n{self.next_step_prompt}", 每次命中都往前面再叠一层,提示词会越来越长。

---

6. 一个 Pydantic 小技巧

BaseAgent 继承的是 BaseModel,并开了 extra = "allow"(app/agent/base.py:45-47), 所以子类可以随手加字段而不用改基类。

另外 initialize_agent 这个 @model_validator(mode="after") (app/agent/base.py:49-56)会在实例化后检查:如果 llm 不是 LLM 实例, 就按 agent 名字的小写 建一个 LLM(config_name=self.name.lower())。 配合配置文件里的 [llm.xxx] 分节,意味着你可以给不同智能体配不同模型 (配置合并逻辑见 app/config.py:313-327)。

---

7. 代码地图

主题	文件路径	符号名
智能体基类	app/agent/base.py	BaseAgent
主循环	app/agent/base.py	BaseAgent.run
状态安全切换	app/agent/base.py	BaseAgent.state_context
写记忆(按角色分派)	app/agent/base.py	BaseAgent.update_memory
卡死检测 / 处理	app/agent/base.py	BaseAgent.is_stuck、BaseAgent.handle_stuck_state
抽象单步	app/agent/base.py	BaseAgent.step
状态枚举	app/schema.py	AgentState
消息模型与工厂	app/schema.py	Message、Message.from_tool_calls
记忆与滑窗	app/schema.py	Memory、Memory.add_message
沙箱全局单例(循环结束时清理)	app/sandbox/client.py	SANDBOX_CLIENT