从零开始的 Nonoka Agent 框架设计
这篇文章简单介绍一下前面 nonoka cli 文章中提到的自研 nonoka agent 框架的设计思路，这个框架也是经过了很多次推倒重来才初步完成的，在开发过程中也学到了很多东西，欢迎讨论。
1. 设计的历史
最初想做 nonoka agent 框架是因为四月份看到 babyagi 项目和深入学了下装饰器的使用，想做一个类型安全、事件驱动、使用装饰器配置组装的 Agent 框架，但是开发过程中遇到了些问题，包括但不限于：
- 装饰器边界没设计好边界模糊，而且一些语义不适合用装饰器表示，如
@plan对应的 Plan 是运行时动态生成的执行意图，用装饰器静态定义不符合直觉。 - 事件驱动架构下，重试、Checkpoint、错误恢复在事件驱动模型里难以表达。如果使用 Control Flow 的话，我们可以很方便地记录执行的输入和输出结果，然后利用这个进行幂等重试或者错误恢复；但是事件驱动中，我们无法保证同一个输入在 LLM 的驱动下会进入同一个 event，这让”重放恢复”失去意义。
这之后的框架设计核心就转移到了 “类型安全 + 执行编排 + 可观测性 + 可扩展性” 这四个方面，整个 Agent Runtime 流程也设计成以控制流为主、事件降级为观测用途（日志、追踪、指标）。
另外前面提到了装饰器的静态属性也对之后的设计有很大启发：最终的 tool 和 prompt 仍然采用的装饰器注册来实现，但是比如对于 tool，具体的运行上下文从哪里来？schema 怎么安全推导？这些东西都是动态的，这促成了框架中重要的组件 RunContext[Deps]、负责完成运行时的动态上下文注入。
之后就有了目前这个初步版本的 nonoka agent 框架。
2. 模块设计
总体架构
目前初步定型的 nonoka-agent 框架是四层架构：
┌────────────────────────────────────────────────────────────────────────┐
│ Paradigm Layer (User-facing API) │
│ ├─ ReActAgent — Exploratory tasks with dynamic decision-making │
│ ├─ ReflectiveAgent — Quality-driven tasks with Execute-Eval-Refine │
│ └─ PlanExecutor — Deterministic tasks with predefined Plan │
├────────────────────────────────────────────────────────────────────────┤
│ Runtime Layer │
│ ├─ Runner — Session lifecycle, checkpoint & resume │
│ ├─ ToolRegistry — Explicit tool registration (no global state) │
│ └─ LLMProvider — Pluggable LiteLLM gateway │
├────────────────────────────────────────────────────────────────────────┤
│ Core Abstractions │
│ ├─ Tool / Capability — Function signature to JSON Schema │
│ ├─ Plan / Step / Ref — Immutable DAG & topological grouping │
│ ├─ RunContext / Deps — Dependency injection & tool call helpers │
│ └─ Session / State — Mutable runtime state & serialization │
├────────────────────────────────────────────────────────────────────────┤
│ Infrastructure │
│ ├─ types, config, errors, event, logger │
│ └─ CheckpointStore, MemoryBackend │
└────────────────────────────────────────────────────────────────────────┘整个框架的核心思想如下：
- Agent 是配置——一个 frozen 不可变的数据对象，声明”用什么模型、有哪些工具、执行策略是什么”；
- Session 是状态——唯一可变的事实源，是一个可序列化、可持久化的状态机；
- Runner 是执行内核——负责把配置和状态组装起来，选择范式、驱动循环、管理生命周期。
用户创建和调用 Agent 变得非常简单明了：
# 1. 声明式配置：读取 nonoka.yaml，得到配置对象
config = Config.load("nonoka.yaml") # 或 Config.auto_find()
# 2. 构建 Agent（纯配置对象）与 Runner（执行内核）
agent = config.agents["weather_assistant"].build()
runner = config.runner.build()
# 3. 选择范式执行（内部：创建 Session → ReAct 循环 → checkpoint 持久化）
result = await runner.run_react(agent, "北京天气怎么样？", deps=None)下面简单介绍各个模块的设计。
Agent：声明式配置
Agent 是一个 frozen dataclass，没有任何运行时状态、不执行任何东西：
# nonoka/core/agent.py
@dataclass(frozen=True)
class Agent(Generic[DepsT, ResultT]):
"""
Agent is a state-less, immutable configuration object.
It holds the model, tools, and execution policy.
Runtime state (plan progress, checkpoint, memory) lives in ``Session``.
"""
model: str
tools: list[Capability] | ToolRegistry = field(default_factory=list)
system_prompt: str = ""
# 执行策略（None 显式关闭对应的累计会话预算）
max_turns: int | None = 10
max_steps: int | None = 50
max_concurrency: int = 10 # 单轮内最大并发工具调用数
temperature: float | None = None
max_tokens: int | None = None
default_retry: RetryPolicy = field(default_factory=RetryPolicy)
default_timeout: float | None = None
# 运行时契约（详见 Capacity 一节）
runtime_limits: RuntimeLimits | None = None
completion_contract: CompletionContract | None = None
# 循环扩展（loop detection / 修复计数等，状态存于 checkpoint）
extensions: list[Any] = field(default_factory=list)
# 技能包：构造时展开合并（详见 Capacity 一节）
skills: list[Skill] = field(default_factory=list)
# 元数据：路由、可观测性、平台集成
metadata: dict[str, Any] = field(default_factory=dict)
tags: list[str] = field(default_factory=list)早期的设计中 Agent 还会有 memory 配置、checkpoint 配置之类的，后面觉得这些东西全部塞给 Agent 会给用户造成很大的心智负担（有太多莫名其妙的参数了，明明我只想配一下用什么模型、挂哪些工具这样的）、而且会让 Agent 过于臃肿，就拆到 Runner 了，正好 Runner 是负责实际的执行的、而这些组件其实是执行的时候才会用到。
可以发现 tools 也在 Agent 中配置，但是 tools 可能需要热重载之类的，为此后面引入了
ToolListProxy这个东西，来获取当前时候的快照、作为ToolRegistry的代理视图。如果只是单纯挂了工具没用 Registry 就直接锁死了：
# nonoka/core/hot_reload.py
class (Sequence):
def __init__(self, static_tools: list[Capability], registries: list[ToolRegistry]):
...
def _snapshot(self) -> list[Capability]:
# 每次迭代时重新收集 registry 中的工具，热加载立即可见
...Runner：执行内核
Runner 是唯一持有”运行环境”的装配器：LLM provider 缓存、checkpoint store、memory backend、hooks、gateway、circuit breaker 这些执行中的组件都在这里被挂载，具体执行的 Agent 范式通过显式方法选择：
# nonoka/core/runner.py
class Runner:
def __init__(self, checkpoint=..., memory=..., circuit_breaker=...,
hooks=..., gateway=..., observability=...):
...
# 范式入口（每个都是"创建 Session → 范式循环 → 持久化"的完整生命周期）
async def run_react(self, agent, prompt, deps=None) -> RunResult
async def run_react_stream(self, agent, prompt, deps=None) -> AsyncIterator[StreamEvent]
async def run_plan(self, agent, plan, deps=None) -> RunResult
async def run_reflective(self, agent, prompt, deps=None) -> RunResult
# 从 checkpoint 恢复：按 session 里是否有未完成的 plan 路由范式
async def resume(self, agent, session_id, deps=None) -> RunResult之前 nonoka-agent 将所有的组件全部设计成接口了，但是这会导致代码冗余：有的组件用户压根不会想实现自己的接口而是只想用现有的、甚至自己都懒得装配，因此最后只保留了 Capability、MemoryBackend、CheckpointStore 三个用户真正可能替换的 Protocol。
不同 Agent 范式的实现
最初的设计想的是让 Runner 动态决策用 Plan-and-Execute 范式或者 Reflective 范式、设计一个 Scheduler 来统一调度，但是后面发现这样太不可控了，于是改成了显式调用、同时将 Plan 单独抽了出来作为一个组件。现有的范式如下：
| 范式 | 定位 | 决策来源 | 错误模型 |
|---|---|---|---|
| ReActAgent | 探索性任务，每步动态决策 | 每轮 LLM 推理 | LLM 重新思考策略 |
| PlanExecutor | 确定性任务，预定义 Plan 编排 | 用户预定义 DAG | 重试/跳过/中断 |
| ReflectiveAgent | 质量驱动任务 | 执行-评估-改进循环 | Evaluator 决定是否重做 |
每个范式都有一个 run 方法，如 ReActAgent.run()；然后 Runner 有独立的 Runner.run_react() 方法。
具体的 Agent Loop，以 ReAct 范式为例如下：
[1] begin_model_turn (Budget Check) ──► [2] enforce_context_budget (Compress Context)
│
┌──────────────────────────────────────────────────────┘
▼
[3] Build Messages & Tool Schemas ──► [4] LLM Inference ──► Record Usage
│
┌──────────────────────────────────────────────────┘
├─► No Tool Calls ──► [5] Final Answer Path (Check Completion Contract Evidence)
│
└─► Has Tool Calls
│
▼
[6] Persist Assistant Msg w/ tool_calls (Crash Recovery Base)
│
▼
[7] reserve_tool_calls (Pre-check Batch Budget -> Abort All on Limit)
│
▼
[8] Concurrent Tool Execution (Wave Runner / Tool Execution Coordinator)
│
▼
[9] Append Tool Results (with tool_call_id)
│
▼
[10] Inject SYSTEM Guidance ──► [11] Loop Detection ──► [12] Checkpoint之后画图都用 ASCII Graph 了，方便和快一些hhh。
可以看到 checkpoint 之类的处理维护都已经封装在范式执行方法中了，用户只需要调用，Runner 和范式的 run 方法就会自己处理好 Agent Loop、持久化、错误处理重放之类的细节。之后会单独写一篇简单讲解这个 Agent Loop 设计的文章。
Checkpoint：nonoka-agent 的状态持久化机制
nonoka-agent 默认使用 sqlite3 状态持久化机制，设计如下：
| 表 | 内容 | 写入时机 |
|---|---|---|
checkpoints | SessionState 全量快照 | 每个 checkpoint 边界 |
step_updates | 单 step 的 status/result/error 增量 | 每次工具执行前后 |
memory_entries | WorkingMemory 的持久化记忆 | 异步落盘 |
nonoka 的 checkpoint 提供 CheckpointStore Protocol，用于存储 Agent 状态快照。
Session：对话状态机
Session 就是 nonoka-agent 的对话状态管理中心，Runner 中 LLM Chat 的乱七八糟状态全部由它负责，nonoka agent 将其拆分为静态快照和可变运行时：
class SessionState(BaseModel):
"""不可变快照——checkpoint 持久化的就是它。"""
status: SessionStatus
current_plan: Plan | None
completed_steps: list[str]
step_statuses: dict[str, StepStatus]
memory_entries: list[MemoryEntry]
turn_count: int
runtime_state: dict # 预算累计（turns/steps/cost）
extension_state: dict # loop/HITL/lineage 扩展计数
class Session:
"""可变运行时——一轮执行期间由 Runner/范式驱动。"""
def begin_model_turn(self): ...
def reserve_tool_calls(self, n): ... # 整批预占
def enforce_context_budget(self): ... # 上下文压缩触发器
def cancel(self): ... # 协作取消
def to_state(self) -> SessionState: ...
@classmethod
def from_state(cls, state): ... # 恢复入口整个对话状态机流程如下：
[ CREATED ] ──► [ RUNNING ] ──► [ PAUSED ] ──( Resume )──► [ RUNNING ]
│
├──► ( Complete ) ──► [[ COMPLETED ]]
├──► ( Fail/Error ) ─► [[ FAILED ]]
└──► ( Cancel ) ────► [[ CANCELLED ]]这里面的
PAUSED状态是一个重要的设计，它是实现 Agent 外部介入的关键，nonoka agent 不会一直 await 阻塞等待，而是采用“冻结 + 持久化 + 进程退出”的设计方式。这种方式能确保 Agent 暂停以后信息肯定不会丢失，不管当前的进程是否还在（我们的 nonoka cli就是每次新 spawn 一个 server 的嘛，每次都用 sessionId 来恢复之前的记录）：
PAUSED Pattern (Stateless / Resilient):
[ Phase 1: Suspend ]
Process ──► Tool Call ──► Save Checkpoint ──► Set PAUSED ──► Exit Process
[ Phase 2: Async Approval ]
External Trigger / Human Approver completes task
[ Phase 3: Resume ]
New Process ──► Load session_id ──► Resume ExecutionCapacity：外部能力抽象
本地函数工具、MCP 工具、Skill、外部宿主工具，最终都统一到 Capability 接口。
# nonoka/core/types.py
@runtime_checkable
class Capability(Protocol):
"""Every tool/agent should implement this protocol."""
@property
def name(self) -> str: ... # 工具名
@property
def description(self) -> str: ... # 给模型看的描述
@property
def parameters(self) -> dict[str, Any]: ... # 参数 JSON Schema
@property
def execution(self) -> Any: ... # ToolExecution 元数据（并发/副作用标志）
async def invoke(self, ctx: RunContext, arguments: dict[str, Any]) -> Any: ...
def to_json_schema(self) -> dict[str, Any]: ...其中 Tool 用装饰器声明：
@nonoka.tool
async def get_weather(city: str) -> str:
"""Get the weather for a city."""
return f"Sunny in {city}!"| 能力 | 加载方式 | 执行位置 | 生命周期 |
|---|---|---|---|
| 本地工具 | @tool / "module:func" 字符串 | 框架进程内 | 无 |
| MCP | MCPManager 启动 server | 独立子进程（stdio/SSE） | 连接/健康检查/退避重启/部分失败 |
| Skill | SkillLoader 扫描目录 | 框架进程内 | 构造期展开 |
| ExternalCapability | 注册表声明 | 外部宿主（如 OpenCode） | 挂起/恢复 |
其中
ExternalCapability是 nonoka-cli 为了支持 Opencode 原生工具而引入的，这个工具的invoke调用永远抛出ExternalToolExecutionRequiredError**——框架进程内根本无法执行这个工具，只能让外部宿主执行。执行结果通过ExternalToolReceipt回传。下面是 nonoka-cli 中的运行流程：
OpenCode Exec External Tool
└─► Pack Result: ExternalToolReceipt
│
▼
ChatRequest (NDJSON)
│
▼
Bridge Handler Extracts tool_results
│
▼
_sanitize_messages_for_resume (Skip replaying old messages)
│
▼
orchestrator.resume_external_tools(session_id, results)
│
▼
Runner.resume_external_tools (Load PAUSED Session)
├── [1] Find Pending Calls: Patch missing results (No replay)
├── [2] Format Receipt: ExternalToolReceipt.from_value(raw)
└── [3] Security Attestation: Require workspace attestation if mutated
│
▼
Write TOOL Entry (w/ tool_call_id) ──► ReAct Loop ContinuesHook：用户自定义逻辑与中间件
nonoka cli 使用传统的 hook 机制让用户在关键生命周期点插入逻辑，可以直接继承一个 Hooks 类或者用 nonoka agent 提供的装饰器语法糖：
# Style 1：继承子类
class LoggingHooks(Hooks):
async def on_llm_request(self, ctx, messages, tools):
print(f"LLM call with {len(messages)} messages")
runner = Runner(hooks=LoggingHooks())
# Style 2：构造函数传函数列表
async def log_request(ctx, messages, tools):
print(f"LLM call with {len(messages)} messages")
runner = Runner(hooks=Hooks(on_llm_request=[log_request]))
# Style 3：实例装饰器注册
hooks = Hooks()
@hooks.on_llm_request
async def log_request(ctx, messages, tools):
print(f"LLM call with {len(messages)} messages")
runner = Runner(hooks=hooks)hooks 可在如下的事件中挂载：
- 生命周期：
on_session_start/on_session_end； - LLM 调用：
on_llm_request/on_llm_response/on_llm_usage； - 工具调用：
on_tool_start/on_tool_start_intercept（可改写参数，用于实现 HITL）/on_tool_end； - Plan 执行：
on_plan_start/on_plan_step_start/on_plan_step_end。
LoopExtension：对 Agent Loop 本身的拓展组件
LoopExtension 是我觉得设计的很有意思的一个组件，它可以在 Agent Loop 本身的生命周期上挂载受约束的增强点。和观测拦截事件的 hooks 不同，LoopExtension 是 Agent Loop 层级的，它有下面四个生命周期钩子：
# nonoka/core/extensions.py
class LoopExtension(Protocol):
name: str
async def before_turn(self, context: LoopExtensionContext) -> ExtensionDecision | None: ...
async def after_tool_batch(self, context: LoopExtensionContext) -> ExtensionDecision | None: ...
async def before_final_answer(self, context: LoopExtensionContext) -> ExtensionDecision | None: ...
async def after_run(self, context: LoopExtensionContext, result: Any) -> None: ...每个钩子返回一个 ExtensionDecision：
@dataclass(frozen=True)
class ExtensionDecision:
feedback: str | None = None # 注入一条 SYSTEM 消息
continue_loop: bool = False # 消耗下一轮正常 turn
replacement_content: str | None = None # 替换最终答案
failure: str | None = None # 终止运行
disable_tools: bool = False # 本轮禁用工具（tool-free finalization）
details: dict[str, Any] = field(default_factory=dict)设计这个组件的缘由是运行开源 coding benchmark 时，由于对一些特定的规则没有好的校验机制 nonoka 会将一些原本还不是正确完整的答案认为是对的直接提交、导致测评不通过，在给 benchmark 使用的 Adapter 中添加下面的插件后，这个问题得到了解决：
class VerifierRepairExtension:
"""Request a bounded repair only after a deterministic verifier fails."""
name = "verifier_repair"
async def before_final_answer(self, context: LoopExtensionContext) -> ExtensionDecision:
result = RunResult(success=True, data=context.content or "", session=context.session)
evaluation = await self.evaluator.evaluate(result) # 验证器检查最终答案
if evaluation.passed:
return ExtensionDecision(details={"passed": True}) # ✅ 通过，放行
attempt = int(attempts.get(self.name, 0)) # 修了几次（存在 session 里）
if attempt >= self.max_repairs: # 修满 2 次还不行
return ExtensionDecision(failure="Verifier rejected the final answer...") # 终止
return ExtensionDecision(
feedback=f"[Verifier feedback — repair attempt 1/2]\n{feedback}", # 注入纠正消息
continue_loop=True, # 让模型再修一轮
)它挂在
before_final_answer（模型给出答案、循环还没结束时），这会触发下面的流程：
模型说"完成了" → before_final_answer 触发 → 验证器检查
├─ 通过 → 放行（final answer）
└─ 不通过 → 注入 feedback + continue_loop → 模型再修一轮（最多 2 次）
└─ 修满 2 次还不行 → failure 终止运行