在上一个系列《从零实现一个 mini vLLM》的 21 篇文章里，我们从最朴素的手写 generate 循环出发，一步步搭出了一个具备 Sequence 状态抽象、Scheduler 连续批处理、Prefill 与 Decode 分离、BlockManager、PagedAttention、Prefix Caching、Sampling 系统，并且兼容 OpenAI /v1/chat/completions 接口的推理引擎。
写完最后一篇《生产部署》的时候，我本来以为关于大模型系统的核心故事已经讲完了。
毕竟此时此刻，只要启动服务，任何客户端都可以发来一段 messages，经过 Tokenizer、ModelRunner 和 Sampler，流式地拿到模型生成的文本。
但当我真正试着基于这个推理引擎去写一个能够自己查资料、读代码库、跑终端命令并修复 Bug 的 Agent 时，我很快意识到一个问题：一个能高效吐出 Token 的推理引擎，距离一个真正能干活的 Agent，中间还隔着一整层系统软件。
如果用传统计算机系统来做类比：LLM 推理引擎（如 vLLM、SGLang）扮演的角色更像是CPU 与内存控制器（MMU）——它负责把算力压榨到极致，负责管好 KV Cache 这块宝贵的显存页面，负责在每一个时钟周期（Step）里精准地算出下一个 Token 的概率分布。
但光有 CPU 和 MMU，你还没法直接运行复杂的现代应用程序。你还需要一层操作系统内核与运行时（OS Kernel & Runtime）：它要负责管理进程的生命周期与状态切换，要负责系统调用（Tool Execution）的权限隔离与异步 I/O，要负责虚拟内存的换入换出与垃圾回收（Context Compaction & Memory Tiering），还要负责多进程之间的通信与同步（Multi-Agent Orchestration）。
现在市面上关于 Agent 教程非常多。打开大多数文章，前十分钟通常是教你装一个庞大的第三方框架，定义几个装饰器，再拼一段 Prompt，跑通一个"查天气"或者"算加减法"的 Demo。一旦你想追问得深一点：
如果模型生成的 JSON 少了一个右括号，整个调用链路怎么自愈？
如果一个任务要连续跑 30 轮工具调用，Context Window 爆了或者 Prefix Cache 失效了怎么办？
如果模型调了一个 bash 工具写出了死循环或者危险指令，怎么在进程级做沙箱隔离与文件快照回滚？
如果客户端在中途断开了连接，或者某个操作需要挂起等待人类审批（Human-in-the-loop），当前执行到一半的状态怎么序列化落盘，并在三天后原样恢复？
你会发现，调包教程往往在这些真正决定系统生死的地方戛然而止。而当你试图点进某些重型框架的源码去排查问题时，迎接你的往往是十几层抽象基类、隐式的回调链条和改一行就崩全局的内置 Prompt。
这正是我决定开启《从零实现一个 mini Agent》这个新系列的原因。
和《mini vLLM》一样，我们不调现成的 Agent 框架，不堆砌玄学概念，而是完全从系统工程的视角出发，一行一行代码手搓出一个结构清晰、模块解耦、可观测、可评测的mini Agent Runtime。更重要的是，因为我们自己手握底层的 mini vLLM，我们可以把很多别人只能当黑盒看的问题——比如采样阶段的语法约束解码（Constrained Decoding）、面向 Agent 多轮负载的 KV Cache 前缀对齐——从上层 Runtime 一直打通到底层推理内核。
作为整个系列的开篇，这一篇我们先不急着堆砌复杂的外围功能。我们先做三件事：
从数学形式化与计算开销的角度，算清楚单次推理（Single-Pass Inference）和多步智能体执行（Multi-Step Agentic Execution）在系统负载上到底有什么本质区别。
拆解一个请求在 Agent Runtime 里的完整生命周期，划清楚各个核心模块的职责边界。
定义好整个 Runtime 最地基的数据契约（Data Contracts）与最小执行引擎骨架，为后续 15 篇的迭代打下基础。
一、为什么有了 /v1/chat/completions 还远远不够
在 mini vLLM 的视角里，一个请求的生命周期是非常纯粹的。
外部通过 HTTP 发来一段对话历史 x = (x₁, x₂, …, xₘ)，引擎将其封装为一个 Sequence 对象。随后在每一轮 Engine.step() 中，模型做一次前向传播，计算条件概率并采样出下一个 Token：
yₜ ~ P_θ(yₜ | x₁, …, xₘ, y₁, …, yₜ₋₁)
一旦采样出的 yₜ 命中了停止条件（比如遇到 EOS token、命中 stop_words 或者生成长度达到 max_tokens），这个 Sequence 的状态就会变成 FINISHED。Block Manager 立即回收它占用的所有物理 KV Cache 块，API 层把拼好的完整字符串返回给客户端。到此为止，引擎与这个请求彻底两清，不留下一丝痕迹。
这种**无状态、单次往返（Stateless Single-Turn）**的假设，在普通聊天场景下运转得非常好。但在 Agent 场景下，情况发生了根本性的变化。
一个 Agent 面对的不是"回答一个问题"，而是"在外部环境 E 中完成一项任务"。它不能一口气把所有 Token 吐完就结束，而是必须走入一个思考 → 行动 → 观测 → 再思考的闭环控制流。
二、从自回归生成到 POMDP：Agent 运行时的形式化建模
为了不在后续的代码设计中迷失方向，我们先用形式化的语言把 Agent Runtime 正在求解的问题写清楚。
在单次 LLM 推理中，生成过程是一个纯粹的开环自回归过程（Open-Loop Autoregressive Process）。模型生成的每一个 Token yₜ，都会直接作为下一步的输入追加到序列末尾。整条序列完全由模型内部的参数 θ 和采样策略决定，与外部真实世界没有任何交互。
而在 Agent Runtime 中，系统的运行轨迹（Trajectory）实际上是一个部分可观测马尔可夫决策过程（POMDP）。我们设第 k 步的完整历史轨迹为 τₖ：
τₖ = (u, s₁, a₁, o₁, s₂, a₂, o₂, …, sₖ, aₖ, oₖ)
其中每一项都有明确的工程实体对应：
u 是初始输入，包含系统级指令（System Prompt）、可用工具的描述集合（Tool Schemas）以及用户的初始任务目标（User Query）。
sₖ 是模型在第 k 步产生的内部推理过程（Thought / Reasoning tokens）。
aₖ 是模型在第 k 步决定的结构化行动（Action），通常表现为一个或多个工具调用请求 {(name_i, args_i)}，或者是向用户输出最终答案的终止动作 Finish。
oₖ 是外部环境 E（如本地 Bash 沙箱、代码解释器、搜索引擎或外部数据库）在执行动作 aₖ 后返回给 Runtime 的观测结果（Observation）。
在这个模型下，一整轮 Agent 任务由 K 个离散的**执行步（Execution Steps）**组成。在第 k 步中，系统的状态转移由两个截然不同的阶段交替驱动：
第一阶段：策略生成阶段（Policy Generation，运行在 GPU 推理引擎上）
Runtime 将截至上一步的历史轨迹 τₖ₋₁ 编码为模型的上下文序列，交由 LLM 采样生成当前步的思考 sₖ 与动作 aₖ：
(sₖ, aₖ) ~ π_θ(· | Context(τₖ₋₁))
这里之所以写成 Context(τₖ₋₁) 而不是直接传入 τₖ₋₁，是因为当步数 k 增长或某次观测 oⱼ 极其庞大（例如读取了一个几千行的源文件）时，原始轨迹的长度 |τₖ₋₁| 会超过模型的物理上下文窗口上限 L_max。Runtime 必须引入一个显式的上下文变换算子 C : τ ↦ τ̃，保证在任何时刻都有：
Len(C(τₖ₋₁)) + T_reserve ≤ L_max
第二阶段：环境转移阶段（Environment Transition，运行在 CPU / 沙箱环境中）
当推理引擎因为输出了完整的工具调用动作 aₖ 而停止当前轮生成时，控制权从推理引擎交还给 Agent Runtime。Runtime 解析并校验 aₖ，在受控环境 E 中执行该动作，使外部环境的状态从 Eₖ₋₁ 转移到 Eₖ，并捕获观测输出 oₖ：
(Eₖ, oₖ) ~ T(Eₖ₋₁, aₖ)
随后，Runtime 将 (sₖ, aₖ, oₖ) 追加到轨迹中得到 τₖ，再次发起第 k+1 轮推理请求，直到动作 a_K 命中终止条件（或触发最大步数、预算耗尽等熔断机制）。
三、算一笔系统账：Agent 负载下的延迟与 KV Cache 成本分析
把公式写出来不是为了好看，而是为了算清楚系统开销。
很多刚接触 Agent 开发的同学会觉得："Agent 不就是用一个 while 循环调了 K 次 LLM 接口吗？能有什么性能问题？"
我们不妨结合之前在 mini vLLM 里学到的 Prefill 与 Decode 性能特征，严谨地推导一下一次 K 步 Agent 任务的总计算量与延迟构成。
设初始系统提示词、工具定义与用户问题的 Token 长度为 L₀。在第 i 步（1 ≤ i ≤ K）中：
模型生成的思考与工具调用 (sᵢ, aᵢ) 的 Token 数量为 gᵢ；
外部工具执行耗时为 t_tool,ᵢ，工具返回的观测结果 oᵢ 被编码为 Token 后的长度为 rᵢ。
那么到了第 k 步发起模型推理时，假设不做任何上下文裁剪，当前序列的总输入前缀长度 Lₖ₋₁ 为：
Lₖ₋₁ = L₀ + Σ(i = 1 … k-1) (gᵢ + rᵢ)
3.1 无 Prefix Caching 情况下的二次方复杂度陷阱
如果底层的推理服务没有开启 Prefix Caching，或者上层 Agent 框架由于糟糕的 Prompt 拼接方式破坏了前缀一致性（比如在最前面的 System Prompt 里塞入了一个每轮都在变化的当前时间戳 current_time），那么在第 k 步，推理引擎必须对全部 Lₖ₋₁ 个 Token 重新做一次 Prefill！
在一个总共 K 步的任务中，推理引擎累计需要处理的 Prefill Token 总数 N_prefill^naive 为：
N_prefill^naive = Σ(k = 1 … K) Lₖ₋₁ = K·L₀ + Σ(k = 1 … K) Σ(i = 1 … k-1) (gᵢ + rᵢ) = K·L₀ + Σ(i = 1 … K-1) (K − i)·(gᵢ + rᵢ)
为了直观感受这个数字，我们代入一组典型的 Coding Agent 参数：
初始提示词与工具描述 L₀ = 2000 tokens；
每一步模型平均生成 gᵢ = 200 tokens；
每次工具执行（比如读取代码片段或跑测试输出）平均返回 rᵢ = 800 tokens；
任务一共执行了 K = 20 步。
此时每一步新增的上下文长度 gᵢ + rᵢ = 1000 tokens。代入上面的公式：
N_prefill^naive = 20 × 2000 + Σ(i = 1 … 19) (20 − i) × 1000 = 40000 + (19 × 20) / 2 × 1000 = 230,000 tokens
而在这 20 步里，模型真正新生成的 Decode Token 总数其实只有：
N_decode = Σ(k = 1 … 20) gₖ = 20 × 200 = 4,000 tokens
看到问题所在了吗？为了生成区区 4,000 个有效 Token，系统居然被迫计算了 230,000 个 Token 的 Prefill，计算放大倍率高达57.5 倍，且随着步数 K 呈严格的 O(K²) 二次方爆炸增长！
3.2 开启 Prefix Caching 且上下文严格对齐时的线性开销
反过来，如果我们在设计 Agent Runtime 时，严格遵守**只追加（Append-Only）、前缀不可变（Immutable Prefix）**的上下文构造原则，让底层 mini vLLM 的 BlockManager 能够完美命中历史前缀块，情况会变成什么样？
在第 1 步，系统需要计算 L₀ 的 Prefill，随后在 Decode 阶段生成 g₁ 个 Token（其对应的 KV Cache 已经在 Decode 过程中写入物理块）。
到了第 k 步（k ≥ 2），前面 Lₖ₋₂ + gₖ₋₁ 个 Token 的 KV Cache 全部可以直接从 BlockManager 的哈希缓存树中复用！引擎唯一需要新做 Prefill 的，仅仅是上一步工具刚刚返回的观测结果 rₖ₋₁。
此时，K 步任务实际需要计算的 Prefill Token 总数 N_prefill^cached 骤降为：
N_prefill^cached = L₀ + Σ(i = 1 … K-1) rᵢ
同样代入刚才的参数（L₀ = 2000, rᵢ = 800, K = 20）：
N_prefill^cached = 2000 + 19 × 800 = 17,200 tokens
从 230,000 直接降到了 17,200，Prefill 计算量减少了92.5%，复杂度从 O(K²) 降回了线性的 O(K)！
但这里还隐藏着另一个系统级的博弈：在第 k 步模型吐完 Action aₖ、等待外部工具执行的这段时间 t_tool,ₖ 里，底层推理引擎里的 KV Cache 发生了什么？
对于推理引擎来说，第 k 步的 HTTP 请求在吐完 aₖ 的那一刻就已经 FINISHED 了。在 mini vLLM 的实现里，一旦 Sequence 结束，它的引用计数归零，那些存着前 Lₖ₋₁ + gₖ 个 Token 的物理块就会退回给 Free List，随时可能被其他并发请求覆盖（Evict）。如果 t_tool,ₖ 很长（比如跑一次编译测试要 15 秒），等第 k+1 步请求发过来时，缓存块可能早就被冲掉了。
3.3 端到端延迟分解公式
把上面所有的环节串起来，一次 K 步 Agent 任务的端到端总耗时 T_E2E 可以精确分解为四项之和：
T_E2E = Σ(k = 1 … K) [ T_prefill(ΔLₖ) ｜增量预填充耗时 ＋ gₖ · T_TPOT(Lₖ₋₁) ｜自回归解码耗时 ＋ T_runtime,ₖ ｜解析校验与调度开销 ＋ T_env(aₖ) ｜沙箱/工具执行耗时 ]
其中 ΔLₖ = Lₖ₋₁ − L_cached,ₖ₋₁ 是未命中缓存的 Token 长度，T_TPOT(Lₖ₋₁) 是单步生成的每 Token 延迟（受限于随着 Lₖ₋₁ 线性增长的 PagedAttention 显存带宽读取开销）。
这个公式实际上就是我们整个《从零实现一个 mini Agent》系列的优化路线图：
为了压低 ΔLₖ 和 T_TPOT(Lₖ₋₁)，我们需要设计KV Cache 对齐的 Prompt 架构与分层记忆压缩系统（第 10–12 篇）；
为了减少总步数 K（避免因为 JSON 解析失败、参数传错而白白浪费重试轮次），我们需要设计结构化输出校验与语法约束解码（第 4 篇）；
为了压低 T_env(aₖ)（当一步中有多个工具调用或存在无依赖子任务时），我们需要设计异步并发 Tool Executor与DAG 拓扑任务调度器（第 5–6 篇）；
为了保证 T_env(aₖ) 执行期间宿主机不被破坏且错误可逆，我们需要设计安全沙箱与工作区快照回滚（第 7–8 篇）。
四、mini vLLM 与 mini Agent Runtime 的核心概念对偶
既然大家已经熟悉了 mini vLLM 的架构，我们完全可以用一组一一对应的关系，来理解 mini Agent Runtime 为什么要这样划分模块：
核心调度单元
mini vLLM：Sequence（维护 Token IDs 与生成状态）
mini Agent Runtime：AgentSession（维护 Trajectory 轨迹与会话状态机）
单步执行粒度
mini vLLM：Engine.step()（一次模型 Forward + Sample）
mini Agent Runtime：Runtime.step()（一次 LLM 决策 + 工具路由 + 环境观测）
计算/执行后端
mini vLLM：ModelRunner（调用 GPU 算子算出 Logits）
mini Agent Runtime：LLMClient + ToolExecutor（调模型接口 + 跑沙箱工具）
输出控制模块
mini vLLM：Sampler + LogitsProcessor（选出下一个 Token）
mini Agent Runtime：ActionParser + SchemaValidator（解析并校验结构化动作）
资源与容量管理
mini vLLM：BlockManager（管理物理显存 KV Cache Blocks）
mini Agent Runtime：ContextManager + MemoryStore（管理 Token 预算与长短期记忆）
并发与调度策略
mini vLLM：Scheduler（Continuous Batching / PD 分离）
mini Agent Runtime：Orchestrator / DAGPlanner（多工具并发 / 多 Agent 编排）
中断与生命周期
mini vLLM：StopChecker（EOS / Stop Words / Max Tokens / Abort）
mini Agent Runtime：TerminationPolicy（Finish 动作 / 步数上限 / 预算熔断 / 人工审批挂起）
看清了这层对偶关系，你会发现设计一个 Agent Runtime 和设计一个推理引擎在工程哲学上是高度相通的：不要把所有逻辑塞进一个巨大的 while 循环里，而是要把状态（State）、策略（Policy）、执行（Execution）和资源管理（Memory/Context）拆成职责单一、边界清晰的组件。
五、一个请求在 mini Agent Runtime 里的生命周期
在我们动手写代码之前，先把一个用户请求进入 mini Agent Runtime 后的流转路径梳理清楚：
User Query + Tool Definitions
│
▼
┌───────────────────────┐
│ AgentSession │ ◄── 维护完整轨迹 Trajectory、Token 账本与当前状态 SessionStatus
└──────────┬────────────┘
│ 1. 请求组装当前步上下文
▼
┌───────────────────────┐
│ ContextManager │ ──► 检查 Token 预算、保持固定前缀对齐、必要时触发历史折叠/压缩
└──────────┬────────────┘
│ 2. 输出模型可读的 Messages + Tool Schemas
▼
┌───────────────────────┐
│ LLMClient │ ──► 调用底层 mini vLLM (/v1/chat/completions)，获取回复与 Usage
└──────────┬────────────┘
│ 3. 返回原始响应 (Thought + Raw Tool Calls / Final Text)
▼
┌───────────────────────┐
│ ActionParser │ ──► 解析动作类型：是直接回复用户 (Finish)，还是调用工具 (ToolCall)？
└──────────┬────────────┘
│
┌─────┴───────────────┐
│ 若为 Finish │ 若为 ToolCall(s)
▼ ▼
┌──────────┐ ┌───────────────────────┐
│ 结束循环 │ │ ToolExecutor │ ──► 校验参数 Schema、权限检查、在 Sandbox 中执行
│ 返回结果 │ └──────────┬────────────┘
└──────────┘ │ 4. 捕获标准输出/异常，封装为 Observation
▼
┌───────────────────────┐
│ 追加回 AgentSession │ ──► 更新步数计数器，进入下一轮 Runtime.step()
└───────────────────────┘
注意这条链路里的几个关键边界：
AgentSession 只管状态，不管执行。它就像 mini vLLM 里的 Sequence，忠实地记录从任务开始到现在发生过的每一条消息、每一次工具调用的耗时与 Token 消耗，以及当前处于什么状态（比如 THINKING、EXECUTING_TOOLS、WAITING_APPROVAL 还是 COMPLETED）。
LLMClient 只管模型通信，不管工具逻辑。它负责把内部统一的消息结构翻译成底层推理引擎支持的格式，并把 prompt_tokens、completion_tokens 以及缓存命中指标原原本本地带回来。
ToolExecutor 永远不向主循环抛未捕获的业务异常。这一点和传统后端开发非常不同：在传统程序里，文件找不到抛出 FileNotFoundError 会直接中断程序；但在 Agent Runtime 里，工具执行失败（比如参数传错、文件不存在、命令返回非零退出码）本身是一种极其重要的环境观测（Observation）。ToolExecutor 必须把异常捕获下来，格式化成一段清晰的错误提示文本返回给模型，让模型在下一轮 step() 中看到错误并尝试自我修正。
六、核心代码实现：搭建 mini Agent Runtime 的地基
在第一篇里，我们要把整个系列最核心的数据结构（Message、ToolCall、Observation、StepRecord、AgentSession）和最小可扩展的 AgentRuntime 骨架写出来。
为了让代码既具备工业级的类型安全性，又不引入任何臃肿的第三方 Agent 框架，我们全部使用 Python 标准库（dataclasses、enum、typing、json、time、uuid）来构建核心抽象。
6.1 定义消息、动作与观测的数据契约（schema.py）
首先，我们不能在系统内部到处裸传字典 dict。在多轮执行中，字典极易出现字段拼写错误或缺失。我们把模型输入输出与工具交互定义成严格的不可变或强类型数据类：
from __future__ import annotations
import time
import uuid
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Dict, List, Optional
class Role(str, Enum):
"""对话轨迹中的角色枚举"""
SYSTEM = "system"
USER = "user"
ASSISTANT = "assistant"
TOOL = "tool"
class SessionStatus(str, Enum):
"""AgentSession 的显式生命周期状态"""
PENDING = "pending" # 已创建，尚未开始执行
THINKING = "thinking" # 正在请求底层推理引擎生成思考与动作
EXECUTING_TOOLS = "executing_tools" # 正在执行外部工具或沙箱指令
WAITING_APPROVAL = "waiting_approval"# 命中高危操作，挂起等待人工审批
COMPLETED = "completed" # 模型主动完成任务并给出最终回复
FAILED = "failed" # 遇到不可恢复的系统级错误
MAX_STEPS_EXCEEDED = "max_steps" # 达到最大步数上限被熔断
CANCELLED = "cancelled" # 被客户端或外部信号主动取消
@dataclass
class ToolCall:
"""模型发出的一次结构化工具调用请求"""
id: str
name: str
arguments: Dict[str, Any]
raw_arguments: str = "" # 保留原始字符串，方便解析失败时做自修复反馈
@dataclass
class Observation:
"""工具或沙箱环境执行后返回的观测结果"""
call_id: str
tool_name: str
output: str
is_error: bool = False
execution_time_ms: float = 0.0
metadata: Dict[str, Any] = field(default_factory=dict)
@dataclass
class TokenUsage:
"""单步或累计的 Token 消耗统计（与 mini vLLM 的 usage 字段直接对齐）"""
prompt_tokens: int = 0
completion_tokens: int = 0
cached_tokens: int = 0 # 记录 Prefix Caching 命中的 Token 数量
@property
def total_tokens(self) -> int:
return self.prompt_tokens + self.completion_tokens
def add(self, other: TokenUsage) -> None:
self.prompt_tokens += other.prompt_tokens
self.completion_tokens += other.completion_tokens
self.cached_tokens += other.cached_tokens
@dataclass
class Message:
"""轨迹中的单条统一消息结构"""
role: Role
content: Optional[str] = None
tool_calls: List[ToolCall] = field(default_factory=list)
tool_call_id: Optional[str] = None
name: Optional[str] = None
timestamp: float = field(default_factory=time.time)
def to_openai_dict(self) -> Dict[str, Any]:
"""转换为兼容 mini vLLM / OpenAI Chat Completions 的字典格式"""
payload: Dict[str, Any] = {"role": self.role.value}
if self.content is not None:
payload["content"] = self.content
if self.tool_calls:
payload["tool_calls"] = [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.name,
"arguments": tc.raw_arguments or json.dumps(tc.arguments, ensure_ascii=False),
},
}
for tc in self.tool_calls
]
if self.role == Role.TOOL and self.tool_call_id is not None:
payload["tool_call_id"] = self.tool_call_id
if self.name:
payload["name"] = self.name
return payload
注意这里的两个细节：
在 ToolCall 里，我们同时保留了解析后的 arguments: Dict[str, Any] 和原始的 raw_arguments: str。这是因为模型在输出 JSON 时非常容易犯语法错误（比如未转义引号或尾随逗号）。保留原始字符串，才能在解析失败时把原始错误输出原封不动地挂载回轨迹里，让模型在下一步看到自己到底写错了什么。
在 TokenUsage 里，我们专门放了一个 cached_tokens 字段。后面第 12 篇我们把 Agent Runtime 和 mini vLLM 的 Prefix Caching 联动起来做压测时，这个字段就是核心观测指标。
6.2 定义会话状态机与轨迹容器（session.py）
接下来是对应 mini vLLM 中 Sequence 的核心对象——AgentSession。它不仅存消息列表，还要按"步（Step）"记录完整的执行切片，方便后续做 Tracing 回放、断点恢复与评测：
@dataclass
class StepRecord:
"""记录单轮 Runtime.step() 的完整切片，用于可观测性与轨迹评测"""
step_index: int
thought: Optional[str] = None
tool_calls: List[ToolCall] = field(default_factory=list)
observations: List[Observation] = field(default_factory=list)
usage: TokenUsage = field(default_factory=TokenUsage)
llm_latency_ms: float = 0.0
step_total_ms: float = 0.0
@dataclass
class AgentConfig:
"""单个 Agent 会话的运行期控制参数"""
model: str = "mini-llama"
system_prompt: str = "You are a helpful and rigorous AI agent."
max_steps: int = 15
max_context_tokens: int = 16384
temperature: float = 0.0 # Agent 决策默认优先采用确定性较高的采样
class AgentSession:
"""管理单次 Agent 任务的完整生命周期与状态机"""
def __init__(self, user_query: str, config: Optional[AgentConfig] = None) -> None:
self.session_id: str = f"sess_{uuid.uuid4().hex[:12]}"
self.config: AgentConfig = config or AgentConfig()
self.status: SessionStatus = SessionStatus.PENDING
self.created_at: float = time.time()
# 初始化消息轨迹：严格保持 System Prompt 在最前，确保 Prefix Cache 友好
self.messages: List[Message] = [
Message(role=Role.SYSTEM, content=self.config.system_prompt),
Message(role=Role.USER, content=user_query),
]
self.steps: List[StepRecord] = []
self.total_usage: TokenUsage = TokenUsage()
self.final_answer: Optional[str] = None
self.error_reason: Optional[str] = None
@property
def current_step(self) -> int:
return len(self.steps)
def is_finished(self) -> bool:
return self.status in {
SessionStatus.COMPLETED,
SessionStatus.FAILED,
SessionStatus.MAX_STEPS_EXCEEDED,
SessionStatus.CANCELLED,
}
def append_assistant_step(
self,
thought: Optional[str],
tool_calls: List[ToolCall],
usage: TokenUsage,
llm_latency_ms: float,
) -> StepRecord:
"""记录模型的思考与动作输出，并将 Assistant Message 追加到轨迹"""
self.messages.append(
Message(role=Role.ASSISTANT, content=thought, tool_calls=tool_calls)
)
self.total_usage.add(usage)
record = StepRecord(
step_index=self.current_step + 1,
thought=thought,
tool_calls=tool_calls,
usage=usage,
llm_latency_ms=llm_latency_ms,
)
self.steps.append(record)
return record
def append_observation(self, step_record: StepRecord, obs: Observation) -> None:
"""将工具执行的观测结果写回轨迹，形成闭环"""
step_record.observations.append(obs)
self.messages.append(
Message(
role=Role.TOOL,
content=obs.output,
tool_call_id=obs.call_id,
name=obs.tool_name,
)
)
大家可以看到，AgentSession 的设计非常克制。它提供了两个原语：append_assistant_step（对应 POMDP 中的策略生成 (sₖ, aₖ)）和 append_observation（对应环境转移产生的 oₖ）。所有消息只追加不原地篡改前缀，这为后续接入底层 KV Cache 复用打下了天然的基础。
6.3 实现最小可扩展的 Runtime 引擎骨架（runtime.py）
有了 AgentSession，我们就可以写出整个 Runtime 最核心的驱动引擎 AgentRuntime。
注意我们是怎么组织的：就像 mini vLLM 的 Engine 提供单步 step() 和完整运行 run() 一样，我们的 AgentRuntime 也把"推进单步状态转移"抽离成独立的 step(session) 方法。这样做的好处极大——如果你以后想做单步调试器、想在网页端一步步点击"Next Step"观察 Agent 的决策，或者想在多 Agent 调度器里交替推进多个 Session，你只需要在外层调用 runtime.step(session) 即可。
import json
@dataclass
class LLMResponse:
"""LLMClient 返回给 Runtime 的标准化响应结构"""
content: Optional[str]
tool_calls: List[ToolCall]
usage: TokenUsage
class LLMClientProtocol:
"""模型调用层抽象接口：底层可对接 mini vLLM、OpenAI 或本地 Mock 模型"""
def generate(
self,
messages: List[Message],
tool_schemas: List[Dict[str, Any]],
config: AgentConfig,
) -> LLMResponse:
raise NotImplementedError
class ToolExecutorProtocol:
"""工具执行层抽象接口"""
def get_schemas(self) -> List[Dict[str, Any]]:
raise NotImplementedError
def execute(self, tool_call: ToolCall) -> Observation:
raise NotImplementedError
class AgentRuntime:
"""mini Agent 的核心运行时引擎"""
def __init__(
self,
llm_client: LLMClientProtocol,
tool_executor: ToolExecutorProtocol,
) -> None:
self.llm_client = llm_client
self.tool_executor = tool_executor
def step(self, session: AgentSession) -> StepRecord:
"""
推进 AgentSession 走完一个完整的 (Thought, Action, Observation) 周期。
对标 mini vLLM 中的 Engine.step()。
"""
if session.is_finished():
raise RuntimeError(f"Session {session.session_id} has already finished with status {session.status}.")
step_start = time.perf_counter()
# 1. 检查是否达到最大步数限制（防止模型陷入死循环消耗无限 Token）
if session.current_step >= session.config.max_steps:
session.status = SessionStatus.MAX_STEPS_EXCEEDED
session.error_reason = f"Exceeded maximum allowed steps ({session.config.max_steps})."
raise RuntimeError(session.error_reason)
# 2. 策略生成阶段：请求 LLM 生成下一步思考与动作
session.status = SessionStatus.THINKING
llm_start = time.perf_counter()
llm_resp = self.llm_client.generate(
messages=session.messages,
tool_schemas=self.tool_executor.get_schemas(),
config=session.config,
)
llm_latency_ms = (time.perf_counter() - llm_start) * 1000.0
step_record = session.append_assistant_step(
thought=llm_resp.content,
tool_calls=llm_resp.tool_calls,
usage=llm_resp.usage,
llm_latency_ms=llm_latency_ms,
)
# 3. 动作判定：如果模型没有发出任何 tool_calls，说明它认为任务已完成，直接给出最终答复
if not llm_resp.tool_calls:
session.status = SessionStatus.COMPLETED
session.final_answer = llm_resp.content or ""
step_record.step_total_ms = (time.perf_counter() - step_start) * 1000.0
return step_record
# 4. 环境转移阶段：依次执行模型请求的工具调用，并收集 Observation
session.status = SessionStatus.EXECUTING_TOOLS
for tool_call in llm_resp.tool_calls:
obs = self.tool_executor.execute(tool_call)
session.append_observation(step_record, obs)
step_record.step_total_ms = (time.perf_counter() - step_start) * 1000.0
return step_record
def run(self, user_query: str, config: Optional[AgentConfig] = None) -> AgentSession:
"""完整运行一个用户任务直到终止状态"""
session = AgentSession(user_query=user_query, config=config)
while not session.is_finished():
try:
self.step(session)
except Exception as exc:
if not session.is_finished():
session.status = SessionStatus.FAILED
session.error_reason = str(exc)
break
return session
仔细看上面这段 AgentRuntime.step() 的代码，不到 50 行，却已经把整个 Agent 运行时的主干立住了：
它不耦合任何具体的模型厂商或提示词模板（那是 LLMClient 和下一篇 ReAct 解析器的事）；
它不耦合任何具体的工具函数（那是 ToolExecutor 的事）；
每一轮的耗时（llm_latency_ms 与 step_total_ms）、Token 账本（usage）、思考内容与观测结果全部干净地记录在 StepRecord 里。
七、整个系列的演进蓝图
到这里，我们已经把 mini Agent Runtime 的地基搭好了。但现在的它还只是一个最朴素的骨架，面对真实世界复杂场景时还有一大堆硬骨头要啃。
在接下来的 15 篇文章里，我们会沿着下面这张演进图，一步步把这个精简的内核打磨成工业级的 Agent 运行时系统：
最小闭环与状态机（第 1–3 篇）
第 1 篇（本篇）：厘清推理引擎与 Agent Runtime 的边界，建立 POMDP 轨迹模型、开销分析与基础数据契约。
第 2 篇（最小执行闭环）：基于 Python 反射自动从函数签名生成 JSON Schema，手写 ReAct 循环与工具路由，跑通第一个真正能干活的 Agent。
第 3 篇（Session 状态机与持久化）：解决长任务中断、进程崩溃与人工审批（Human-in-the-loop）问题，实现状态的序列化落盘、暂停挂起与断点恢复。
可靠输出与工具调度引擎（第 4–6 篇）
第 4 篇（结构化输出与语法约束解码）：解决大模型 JSON 格式崩溃顽疾——上层实现错误回传自修复，底层联动 mini vLLM 的 Sampler 实现基于有限状态机（FSM）的 Constrained Decoding。
第 5 篇（异步 Tool Executor）：把第 1 篇里的串行 for tool_call in ... 升级为基于 asyncio 的并发执行引擎，加入超时熔断、幂等重试与异常转文本机制。
第 6 篇（Plan-and-Solve 与 DAG 调度）：超越单步走一步看一步的 ReAct，引入任务有向无环图（DAG）规划器与动态重规划机制。
执行环境、安全沙箱与生态协议（第 7–9 篇）
第 7 篇（安全沙箱 Sandbox）：让 Agent 敢于执行 Bash 与 Python 代码——实现进程组管理、资源限制、目录隔离与输出流实时捕获。
第 8 篇（状态快照与事务回滚）：当 Agent 改坏了代码库怎么办？基于影子 Git 仓库实现每次 step 的工作区快照与一键 Time-travel 回滚。
第 9 篇（从零实现 MCP 协议）：告别硬编码工具胶水，基于 JSON-RPC 2.0 手写 Model Context Protocol 的 Client 与 Server。
上下文工程与底层推理协同（第 10–12 篇）
第 10 篇（上下文窗口管理）：实现 Token 预算记账、超大观测结果自动落盘引用（File Reference）与历史轮次滚动折叠压缩。
第 11 篇（分层记忆系统 Memory Tiering）：构建 Working Memory、Episodic 向量检索与 Agent 自主维护的长期知识库。
第 12 篇（KV Cache 对齐的 Prompt 架构）：直接把 mini Agent 接到我们的 mini vLLM 上，解剖什么操作会毁掉 Prefix Caching，并实测缓存对齐前后的 TTFT 与吞吐差距。
多智能体编排与生产级工程化（第 13–16 篇）
第 13 篇（Multi-Agent 编排）：实现父子 Sub-agent 派生（上下文隔离）与同级专家 Agent 的状态机移交（Handoff）。
第 14 篇（流式事件总线与异步 Engine）：对标 mini vLLM 的异步输出队列，通过 SSE 实时向外推送思考 Token 流、工具调用状态与沙箱终端流。
第 15 篇（可观测性与 Tracing）：实现 Span 树轨迹追踪器与本地可视化回放面板，一眼看清 Agent 到底在哪一步犯蠢。
第 16 篇（Benchmark 与自动化评测）：构建确定性 Mock 环境、轨迹匹配度指标与 Pass@k 评测体系，用数据证明每一次架构迭代的真实收益。
八、小结
这一篇作为《从零实现一个 mini Agent》系列的开篇，我们完成了一次从"推理引擎视角"向"智能体运行时视角"的切换。
核心结论可以浓缩为四句话：
推理引擎是 CPU 与 MMU，Agent Runtime 是操作系统内核。推理引擎只负责单轮 Sequence 的高效 Token 生成，而 Agent Runtime 负责管理跨越多个推理轮次与外部环境交互的完整轨迹（Trajectory）。
Agent 任务的本质是 POMDP 状态转移。每一轮 step() 都由 GPU 上的策略生成 (sₖ, aₖ) ~ π_θ 和沙箱环境中的状态转移 (Eₖ, oₖ) ~ T 交替组成。
复杂度控制是系统设计的生命线。如果不做 Prefix Cache 对齐与上下文管理，K 步 Agent 任务的 Prefill 计算量会呈现 O(K²) 的二次方爆炸；只有严格保证前缀不可变与只追加语义，才能把复杂度压回 O(K)。
状态、模型通信与工具执行必须解耦。通过 AgentSession、LLMClientProtocol 和 ToolExecutorProtocol 三个核心抽象，我们已经搭起了清晰的第一版运行时骨架。
下一篇文章，我们会真正动手实现 ToolRegistry 和 ReAct 协议适配器，教你如何利用 Python 的类型注解与 AST/反射机制自动生成工具 Schema，并让我们的 mini Agent 跑通第一个完整的"思考—调工具—看结果—给答案"闭环。
