夜雨聆风核心 Agent 循环与 Turn Runner
源码规模：
engine/agent.py单文件 304KB，engine/runtime.py单文件 301KB。
但核心循环不到 500 行。这就是微内核的力量。
一、源码概览：文件地图
在开始读源码之前，先看清整体布局。OpenSquilla 的核心代码在 src/opensquilla/ 下：
src/opensquilla/
├── engine/ ← 本期主角
│ ├── agent.py ← 核心 Agent 循环 (304KB)
│ ├── runtime.py ← TurnRunner 编排器 (301KB)
│ ├── pipeline.py ← TurnContext 管道上下文
│ ├── turn_runner/ ← 九阶段分解
│ │ ├── harness.py ← 阶段适配器 (43KB)
│ │ ├── prompt_assembler_stage.py
│ │ ├── stream_consumer_stage.py
│ │ └── ...
│ ├── steps/ ← 每个 turn 的执行步骤
│ │ ├── squilla_router.py ← SquillaRouter (53KB)
│ │ ├── skills_filter.py ← Skill 过滤 (14KB)
│ │ ├── meta_resolution.py ← Meta Skill 解析 (58KB)
│ │ └── coding_mode.py ← 编码模式 (17KB)
│ └── routing/ ← 路由决策引擎
├── squilla_router/ ← 本地 ONNX 推理器
├── memory/ ← 记忆系统
├── skills/ ← Skill 加载和管理
├── provider/ ← 30+ LLM 供应商统一接口
├── channels/ ← 飞书/微信/Telegram/Discord...
├── sandbox/ ← 分层沙箱
└── tools/ ← 工具系统二、第一层：Agent 状态机
OpenSquilla 的 Agent 不是一个简单的 while True 循环，而是一个显式状态机。
打开 engine/agent.py，文件开头明确定义了状态：
class AgentState(StrEnum):
"""Agent 的每一个可能状态"""
IDLE = "idle" # 等待输入
BOOTSTRAPPING = "bootstrapping" # 正在加载上下文
PREPARING = "preparing" # 准备 turn（路由、Skill、Prompt 组装）
STREAMING = "streaming" # 正在从 LLM 接收响应
TOOL_EXECUTING = "tool_executing" # 正在执行工具
COMPACTING = "compacting" # 正在压缩上下文
FINALIZING = "finalizing" # 收尾（保存记忆、更新统计）
DONE = "done" # 完成
ERROR = "error" # 出错状态流转：
IDLE → BOOTSTRAPPING → PREPARING → STREAMING
↓
有工具调用？
↙ ↘
YES NO
↓ ↓
TOOL_EXECUTING FINALIZING → DONE
↓
回到 PREPARING
↑
需要压缩？→ COMPACTING为什么用显式状态机？
相比简单的 while True，状态机的优势：
| 简单循环 | 显式状态机 |
|---|---|
| 所有逻辑混在一起 | 每个状态职责清晰 |
| 错误处理散落各处 | 状态间转换可审计 |
| 难以测试中间状态 | 每个状态独立测试 |
| 中途状态不透明 | UI 可展示"正在压缩上下文..." |
三、第二层：Agent 核心循环
Agent 的核心循环在 agent.py 中，每个 turn 的核心逻辑：
# 简化后的核心循环
class Agent:
async def run_turn(self, input: str) -> AsyncIterator[AgentEvent]:
"""执行一个完整的 turn"""
# 1. 身份注入
yield self._inject_identity(input)
# 2. 模型路由
routing_decision = await self._route_model(input)
yield routing_decision
# 3. Skill 加载
skills = await self._load_skills(input, routing_decision)
# 4. Prompt 组装
messages = await self._assemble_prompt(input, skills)
# 5. LLM 调用 + 工具执行（核心循环）
while True:
response = await self._call_llm(messages)
if response.is_text_only:
# 模型直接回答，没有工具调用
yield response
break
if response.has_tool_calls:
# 执行工具
tool_results = await self._execute_tools(response.tool_calls)
messages.extend(tool_results)
continue # 回到第 5 步
# 6. 记忆更新
await self._update_memory()
# 7. 成本记录
yield self._rollup_cost()四、第三层：TurnRunner 九阶段流水线
真正的工程实现远比上面复杂。OpenSquilla 将每个 turn 分解为九个独立阶段，每个阶段由专门的 stage 类处理。
这是 turn_runner/ 目录的九个文件：
阶段 1: agent_bootstrap_stage — Agent 配置引导
阶段 2: attachment_stage — 附件处理
阶段 3: compaction_and_history_stage — 上下文压缩 + 历史加载
阶段 4: input_stage — 输入预处理
阶段 5: prompt_assembler_stage — Prompt 组装（路由+Skill+记忆）
阶段 6: provider_and_tools_stage — 供应商解析 + 工具构建
阶段 7: stream_consumer_stage — LLM 流式消费 + Agent 运行
阶段 8: turn_finalizer_stage — Turn 收尾（记忆+成本+日志）4.1 各阶段详解
阶段 1：Agent Bootstrap — 引导阶段
# agent_bootstrap_stage.py 的输出
class _MemorySnapshotResult:
memory_snapshot: str # 当前记忆快照
agent_config: AgentConfig # Agent 配置
resolved_model: str # 解析后的模型
class _ResolvedBudgets:
tool_budget: int # Tool 结果的 token 预算
context_budget: int # 上下文 token 预算
iteration_budget: int # 迭代次数预算阶段 2：Attachment Stage — 附件处理
# 处理用户上传的文件/图片/音频
# 将附件转换为 LLM 可理解的 ContentBlock
# 支持拖拽上传、剪贴板粘贴阶段 3：Compaction & History Stage
# 核心逻辑：
# 1. 检查上下文是否超过阈值
# 2. 如果超过 → 触发压缩
# 3. 加载历史消息
# 4. 应用 CompactionConfig阶段 5：Prompt Assembler — 最关键的一步
# prompt_assembler_stage.py 的核心
async def assemble_prompt(self, context: TurnContext):
# Step 1: 路由决策
routing = self.router.decide(context.input)
# Step 2: Skill 过滤
skills = self.skills_filter.filter(context.input, routing)
# Step 3: 记忆注入
memory = self.memory.snapshot()
# Step 4: 时间前缀
time_prefix = f"Current time: {datetime.now()}"
# Step 5: 组装系统提示词
system_prompt = "\n\n".join([
self.identity.prompt,
time_prefix,
memory,
skills.prompt_injection,
])
# Step 6: 构建消息列表
messages = [
{"role": "system", "content": system_prompt},
*context.history,
{"role": "user", "content": context.input},
]
return messages阶段 7：Stream Consumer — Agent 运行
这是实际调用 LLM 并处理工具调用的阶段。它调用了 agent.py 中的核心循环。
# stream_consumer_stage.py
class AgentRunPort(Protocol):
async def run_agent_turn(
self,
messages: list[Message],
tools: list[ToolDefinition],
config: AgentConfig,
) -> AsyncIterator[AgentEvent]:
"""运行一个完整的 agent turn，流式产出事件"""
...五、第四层：Pipeline 管道上下文
连接各个阶段的是一套 Pipeline Context 机制：
# pipeline.py
@dataclass
class TurnContext:
"""一个 turn 的全量上下文"""
# 输入
user_input: str
session_id: str
attachments: list[Attachment]
# 路由
routing_decision: RoutingDecision | None = None
selected_model: str = ""
tier_config: TierConfig | None = None
# Skill
active_skills: list[Skill] = field(default_factory=list)
skill_prompts: str = ""
# 记忆
memory_snapshot: str = ""
memory_fingerprint: str = ""
# 成本
token_usage: TokenUsage = field(default_factory=TokenUsage)
cost_usd: float = 0.0
# 诊断
trace_id: str = field(default_factory=lambda: str(uuid.uuid4()))
turn_call_log: dict = field(default_factory=dict)设计理念： 每个阶段只读写 TurnContext 中自己负责的字段。阶段之间通过 TurnContext 传递数据，解耦得干干净净。
六、源码中值得注意的设计决策
6.1 没有递归调用
# engine/agent.py 文件头注释
"""Agent core — explicit state machine + tool loop.
Core loop is under 500 lines. No recursive calls.
"""所有 Agent 行为都是迭代的。工具调用不递归嵌套，而是线性展开。这保证了：
- • 堆栈不会无限增长
- • 每步都可审计
- • 错误状态不会"藏"在递归中
6.2 适配器模式隔离阶段
# turn_runner/harness.py
# 每个 stage 通过 Protocol 定义接口，不依赖具体实现
# 文件头明确写明：
"""Hosts the small adapter classes that bind TurnRunner
instance methods to the Protocol-shaped ports the stages consume."""这意味着：
- • 每个 stage 可以独立测试
- • 替换一个 stage 不影响其他 stage
- • 可以按需添加新 stage
6.3 延迟导入避免循环依赖
# harness.py 明确指出：
"""Each adapter lazy-imports TurnRunner inside its method
bodies to avoid the runtime → turn_runner → runtime import cycle."""这是 runtime.py（301KB）和 agent.py（304KB）没有形成循环依赖的关键。
七、Turn Runner 的完整执行流程
用一张图总结整个 Turn Runner 的协作方式：
┌──────────────────────────────────────────────────────────────┐
│ TurnRunner (runtime.py) │
│ 总编排器，驱动九个阶段 │
├──────────────────────────────────────────────────────────────┤
│ │
│ Stage 1: Agent Bootstrap │
│ ├─ 加载 AGENTS.md / SOUL.md / IDENTITY.md │
│ ├─ 加载记忆快照 │
│ └─ 解析 Agent 配置 + 模型选择 │
│ ↓ │
│ Stage 2: Attachment │
│ ├─ 处理文件上传 │
│ └─ 转换为 ContentBlock │
│ ↓ │
│ Stage 3: Compaction & History │
│ ├─ 检查上下文 token 数 │
│ ├─ 超阈值 → 触发压缩 │
│ └─ 加载历史消息 │
│ ↓ │
│ Stage 4: Input │
│ ├─ 预处理用户输入 │
│ └─ 注入额外上下文 │
│ ↓ │
│ Stage 5: Prompt Assembler │
│ ├─ SquillaRouter 路由决策 │
│ ├─ Skill 过滤 + 注入 │
│ ├─ 记忆注入 │
│ └─ 组装完整 system prompt + messages │
│ ↓ │
│ Stage 6: Provider & Tools │
│ ├─ 解析 LLM Provider │
│ └─ 构建 Tool 定义列表 │
│ ↓ │
│ Stage 7: Stream Consumer ← 核心循环 │
│ ├─ Agent 运行（agent.py 核心循环） │
│ ├─ 流式消费 LLM 输出 │
│ ├─ 工具执行 │
│ └─ 迭代直到 done │
│ ↓ │
│ Stage 8: Turn Finalizer │
│ ├─ 记忆捕获 + 更新 │
│ ├─ Token 成本汇总 │
│ └─ 日志记录 │
│ │
└──────────────────────────────────────────────────────────────┘八、总结：为什么这是优秀的 Harness 工程
回顾这期拆解的四个层次：
| 层次 | 组件 | 行数 | 职责 |
|---|---|---|---|
| L1 | Agent 状态机 | ~300 行 | 定义 Agent 的所有可能状态 |
| L2 | Agent 核心循环 | ~500 行 | while True 工具循环 |
| L3 | TurnRunner 九阶段 | ~3000 行 | 每个 turn 的流水线 |
| L4 | Pipeline Context | ~200 行 | 阶段间数据传递 |
核心洞察：
- 1. 微内核 — 核心循环 500 行，其余都是可插拔的扩展
- 2. 显式状态 — 不是隐藏的递归，是可观测的状态流转
- 3. 阶段解耦 — Protocol + Adapter，替换一个阶段不影响其他
- 4. 无循环依赖 —
runtime.py和agent.py通过延迟导入解耦
下期预告
OpenSquilla 源码解析 ②：SquillaRouter 模型路由引擎
- • ONNX + LightGBM 本地推理的源码实现
- • 路由分级策略：Tier 1-5 的决策树
- • 每次路由决策如何成为训练数据
- • 与 provider 层的交互机制
科技洞察助手 | 源码深度解析系列