1. 从“七个 PoC 各自为政”到一条能跑通的链路
Aegis 平台要解决的核心问题,是把 Agent Harness 从一堆散落的概念验证,收敛成一个能端到端跑真实任务的系统。它是什么?一句话:一个把 ETCLOVG 七层理论(执行沙箱、工具接口、分层记忆、生命周期编排、任务追踪、验证评估、治理安全)拼成可运行代码的 Agent 运行平台。能做什么?接收一个自然语言目标,自主规划、调用工具、在沙箱里改代码、跑测试、自我验证,最后给出带证据的结论。适合谁?正在从“调 API 写 demo”往“搭生产级 Agent 平台”走的工程师,尤其是被 Query Loop 调度、PEV 闭环、统一模型接入这几件事卡住的人。
我试过最典型的翻车场景:七个模块各自能跑,拼起来就断。模型调用散落在每个模块里,Key 到处复制,换一个模型要改五处代码;Query Loop 没有统一入口,工具调用绕过治理直接执行;验证环节靠模型自己说“我做完了”,结果测试根本没跑。这些问题的根子不在算法,在“接入层”和“编排层”没有统一。
这一篇就按从零搭建的顺序走:先定 Harness 配置和统一 Key 通道,再写 Query Loop 调度,再补 PEV 验证脚本,最后端到端跑一次并演示失败重试。模型调用统一走 TaoToken 的 API 通道,Base URL 固定为https://taotoken.net/api,这样模型切换、Key 管理、调用审计都在一个地方收口,Harness 内部只认一个 LLM 接口。
先把整体结构摆出来,后面每一节都对应其中一块:
aegis/ ├── aegis.config.yaml # 平台运行时配置(模型/沙箱/治理) ├── packages/ │ ├── harness-core/ # 接口与领域类型 │ ├── orchestrator/ # Query Loop + PEV │ ├── tools/ # 工具注册表 │ ├── sandbox/ # 执行沙箱 │ ├── memory/ # 分层记忆 │ ├── governance/ # 治理拦截 │ └── observability/ # trace 采集 └── apps/cli/ # aegis run 入口关键决策只有一个:编排器是主干,不是并列的一层。它 import 其他所有包,反过来不行。依赖方向单向,CI 里用规则文件固化,谁想反向 import 立刻红灯。这条约束决定了后面所有代码长什么样。
2. TaoToken 前置:统一 Key 与 API 通道
在写 Query Loop 之前,先把模型接入这层收干净。很多团队在这一步偷懒,把 Key 写进每个模块的环境变量,结果就是换模型要改一堆文件、Key 泄露风险高、调用量无法统一统计。正确做法是:Harness 内部只依赖一个LLM接口,具体实现走 TaoToken 的统一 API 通道。
TaoToken 在这里扮演的是“模型网关”角色。你不需要在代码里区分不同厂商的 SDK,只需要一个 Base URL 和一个 Key。Base URL 是https://taotoken.net/api,Key 在控制台创建。这样 Harness 的 LLM 实现只做一件事:把统一的请求格式发到这个地址,拿回统一的响应。
先看配置。在aegis.config.yaml里,模型段只保留必要信息:
# aegis.config.yaml(模型接入片段) model: baseUrl: https://taotoken.net/api apiKeyEnv: TAOTOKEN_API_KEY # 从环境变量读取,不写死在文件里 name: claude-3-5-sonnet # 模型 ID,按需替换 maxTokens: 4096 timeoutMs: 120000注意apiKeyEnv这个设计:配置文件里只写环境变量名,真实 Key 通过export TAOTOKEN_API_KEY=...注入。这样配置文件可以进版本库,Key 不会泄露。启动时读取:
export TAOTOKEN_API_KEY=你的Key然后是 LLM 接口的实现。Harness 内部定义的是抽象接口,TaoToken 实现是其中一个可替换的适配器:
// apps/cli/src/llm.ts import type { LLM, ToolCall } from '@aegis/harness-core'; export interface TaoTokenConfig { baseUrl: string; apiKey: string; name: string; maxTokens: number; timeoutMs: number; } export class TaoTokenLLM implements LLM { constructor(private cfg: TaoTokenConfig) {} async complete(input: { system: string; context: string; tools: unknown[]; }): Promise<{ thought: string; toolCalls: ToolCall[]; finalAnswer?: string }> { const controller = new AbortController(); const timer = setTimeout(() => controller.abort(), this.cfg.timeoutMs); try { const resp = await fetch(`${this.cfg.baseUrl}/v1/messages`, { method: 'POST', headers: { 'content-type': 'application/json', 'x-api-key': this.cfg.apiKey, 'anthropic-version': '2023-06-01', }, body: JSON.stringify({ model: this.cfg.name, max_tokens: this.cfg.maxTokens, system: input.system, messages: [{ role: 'user', content: input.context }], tools: input.tools, }), signal: controller.signal, }); if (!resp.ok) { const text = await resp.text(); throw new Error(`模型请求失败 ${resp.status}: ${text.slice(0, 300)}`); } const data = await resp.json(); return this.parseResponse(data); } finally { clearTimeout(timer); } } private parseResponse(data: any) { const blocks = data.content ?? []; const thought = blocks .filter((b: any) => b.type === 'text') .map((b: any) => b.text) .join('\n'); const toolCalls: ToolCall[] = blocks .filter((b: any) => b.type === 'tool_use') .map((b: any) => ({ id: b.id, name: b.name, args: b.input ?? {}, })); const finalAnswer = toolCalls.length === 0 && thought ? thought : undefined; return { thought, toolCalls, finalAnswer }; } }这段代码有几个工程要点。第一,超时用AbortController控制,模型请求卡住不会拖死整个 Query Loop。第二,错误信息截断到 300 字符,避免把整页 HTML 错误塞进日志。第三,响应解析把文本块和工具调用块分开,finalAnswer只在没有工具调用时才赋值——这个判断直接决定了 Query Loop 是继续还是收尾。
如果你用的是 OpenAI 兼容格式,把请求体换成messages+tools的 OpenAI 结构即可,Base URL 不变。TaoToken 的通道对两种格式都支持,Harness 侧只需要换一个适配器类,Query Loop 一行不用改。这就是“统一 Key 通道”的价值:模型接入的差异被隔离在一个文件里。
Key 创建和模型列表可以在控制台查看,接入文档里有完整的请求示例。建议先把 Key 配好、用 curl 验证一次请求能通,再往下写 Query Loop,否则后面报错你分不清是编排逻辑问题还是接入问题。
3. 可复制配置:Harness 配置片段与 Query Loop 调度
这一节给两份可直接复制的配置:一份是 Harness 运行时配置,一份是 Query Loop 的调度伪代码。先看配置,它决定了平台启动时装配哪些实现。
# aegis.config.yaml(完整运行时配置) model: baseUrl: https://taotoken.net/api apiKeyEnv: TAOTOKEN_API_KEY name: claude-3-5-sonnet maxTokens: 4096 timeoutMs: 120000 workdir: . sandbox: type: local # local | docker docker: image: aegis-runtime:latest loop: maxSteps: 30 # 单次 Query Loop 最大步数,防死循环 maxReplans: 2 # PEV 重规划次数 recentSteps: 12 # 上下文窗口保留的最近步数 governance: default: allow rules: - match: { tool: run_command, argContains: "rm -rf" } effect: deny reason: "禁止递归强制删除" - match: { tool: write_file, pathPattern: "\\.(env|pem|key)$" } effect: confirm reason: "写入敏感文件需人工确认" - match: { tool: run_command, argContains: "git push" } effect: confirm reason: "推送远端需人工确认" observability: exporter: console # console | otlp这份配置里,loop.maxSteps和loop.maxReplans是两个必须设的闸。前者防单次循环跑飞,后者防 PEV 无限重规划。governance.rules是纯确定性规则匹配,不经过模型推理,改红线只改这个文件。
接下来是 Query Loop 的调度主体。它把“感知-决策-行动-观察”四步做成一个带治理拦截、trace 采集、异常驯服的循环:
// packages/orchestrator/src/query-loop.ts import type { Task, AgentStep, RunResult, ToolCall, ToolResult, Sandbox, ToolRegistry, MemoryStore, PolicyEngine, Tracer, LLM, Verifier, } from '@aegis/harness-core'; export interface LoopDeps { llm: LLM; tools: ToolRegistry; sandbox: Sandbox; memory: MemoryStore; policy: PolicyEngine; tracer: Tracer; verifier: Verifier; } export interface LoopConfig { maxSteps: number; systemPrompt: string; } export async function runQueryLoop( task: Task, deps: LoopDeps, cfg: LoopConfig, ): Promise<RunResult> { const { llm, tools, sandbox, memory, policy, tracer, verifier } = deps; const steps: AgentStep[] = []; const rootSpan = tracer.startSpan('query_loop', { taskId: task.id }); try { for (let i = 0; i < cfg.maxSteps; i++) { const step: AgentStep = { index: i, thought: '', action: { kind: 'final', answer: '' }, startedAt: Date.now(), }; const stepSpan = tracer.startSpan('step', { index: i }); // 感知:组装上下文 const context = await memory.buildContext(task.id, task.goal); // 决策:模型给出下一步 const decision = await llm.complete({ system: cfg.systemPrompt, context, tools: tools.toSchema(), }); step.thought = decision.thought; // 模型认为完成 → 强制走验证 if (decision.finalAnswer && decision.toolCalls.length === 0) { step.action = { kind: 'final', answer: decision.finalAnswer }; step.endedAt = Date.now(); steps.push(step); await memory.appendStep(task.id, step); stepSpan.end('ok'); const report = await verifier.verify(task, sandbox); rootSpan.setAttr('verified', report.passed); rootSpan.end(report.passed ? 'ok' : 'error'); return { taskId: task.id, status: report.passed ? 'success' : 'failed', answer: decision.finalAnswer, steps, verification: report, }; } // 行动:取第一个工具调用 const call: ToolCall = decision.toolCalls[0]; step.action = { kind: 'tool_call', call }; // 治理校验 const verdict = await policy.evaluate(call); let result: ToolResult; if (verdict.effect === 'deny') { result = { callId: call.id, ok: false, content: `[治理拦截] ${verdict.reason}`, }; } else if (verdict.effect === 'confirm') { result = { callId: call.id, ok: false, content: `[需人工确认] ${verdict.reason}`, }; } else { result = await executeTool(call, tools, sandbox, tracer); } // 观察:回注结果 step.result = result; step.endedAt = Date.now(); steps.push(step); await memory.appendStep(task.id, step); stepSpan.setAttr('tool', call.name); stepSpan.setAttr('ok', result.ok); stepSpan.end(result.ok ? 'ok' : 'error'); } rootSpan.end('error'); return { taskId: task.id, status: 'aborted', answer: '达到最大步数仍未完成', steps, }; } catch (err) { rootSpan.setAttr('error', String(err)); rootSpan.end('error'); return { taskId: task.id, status: 'failed', answer: `执行异常:${String(err)}`, steps, }; } } async function executeTool( call: ToolCall, tools: ToolRegistry, sandbox: Sandbox, tracer: Tracer, ): Promise<ToolResult> { const span = tracer.startSpan('tool_exec', { name: call.name }); const tool = tools.get(call.name); if (!tool) { span.end('error'); return { callId: call.id, ok: false, content: `未知工具:${call.name}` }; } try { const r = await tool.execute(call.args, { sandbox, callId: call.id }); span.end(r.ok ? 'ok' : 'error'); return r; } catch (err) { span.setAttr('error', String(err)); span.end('error'); return { callId: call.id, ok: false, content: `工具执行异常:${String(err)}`, }; } }这段调度有三个不能省的硬约束。第一,模型说“完成”不算数,必须走verifier.verify()客观验证,验证不过照样返回 failed。第二,治理拦截不抛异常,而是把拒绝理由包装成ok: false的观察结果回注,让模型下一轮能“看见”自己被拦了,自主换路,而不是整个任务崩掉。第三,工具异常被驯服成可观察的失败数据,一个工具崩了不会拖垮整个循环。
maxSteps是兜底保险。Agent 陷入“反复重试同一个错误动作”的死循环是最常见也最烧钱的故障模式,没有这道闸,它会越陷越深还越来越自信。
4. 验证请求:端到端跑通与失败重试
配置和调度都就位后,跑一次真实的端到端任务。工作区里放一个真实的 bug:src/discount.ts里算折扣时忘了处理“折扣为 0”的边界,导致除零,测试红着。
先验证模型通道能通,用一条最小请求确认 Key 和 Base URL 正确:
curl -s https://taotoken.net/api/v1/messages \ -H "content-type: application/json" \ -H "x-api-key: $TAOTOKEN_API_KEY" \ -H "anthropic-version: 2023-06-01" \ -d '{ "model": "claude-3-5-sonnet", "max_tokens": 64, "messages": [{"role":"user","content":"回复 OK 两个字母"}] }'返回里有content数组且包含文本块,说明通道正常。然后跑平台任务:
aegis run "修复 src/discount.ts 中计算折扣价时的除零 bug,让 pnpm test 全部通过"一次成功的运行轨迹大致如下(节选):
[PLAN] 规划: 1. 读 src/discount.ts 与 discount.test.ts,理解期望行为 2. 跑一次 pnpm test,确认失败的具体断言 3. 定位除零位置,加边界处理 4. 重跑测试验证 [步骤 0] 动作: read_file({"path":"src/discount.test.ts"}) 观察: ok | expect(applyDiscount(100, 0)).toBe(100) [步骤 1] 动作: read_file({"path":"src/discount.ts"}) 观察: ok | return price / (rate === 0 ? 0 : 1/rate) // 明显的除零 [步骤 2] 动作: run_command({"cmd":"pnpm test"}) 观察: FAIL exit=1 | 折扣为0时应返回原价:得到 Infinity [步骤 3] 动作: write_file({"path":"src/discount.ts","content":"..."}) [治理] write_file 命中规则检查……effect=allow 观察: ok | 已写入 src/discount.ts(98 字符) [步骤 4] 动作: run_command({"cmd":"pnpm test"}) 观察: ok exit=0 | Test Files 1 passed, Tests 5 passed [步骤 5] 动作: 给出最终答案 [VERIFY] 执行验收命令 pnpm test …… exit=0 ✓ 状态: success 答案: 已修复 src/discount.ts 的除零 bug:当折扣率 <= 0 时直接返回原价。pnpm test 全部通过(5/5)。 步数: 6 验证报告: [✓] 命令验证: pnpm test: 通过 --- 执行轨迹 --- [ok] query_loop (8421ms) [ok] plan (1203ms) [ok] step (902ms) tool=read_file [ok] step (876ms) tool=read_file [error] step (3201ms) tool=run_command ← 第一次跑测试是红的,符合预期 [ok] step (1102ms) tool=write_file [ok] step (3340ms) tool=run_command轨迹里那条[error] ... tool=run_command很关键:它标出“第一次测试本来就是红的”,让你一眼看懂 Agent 的工作过程,而不是只看到一个成功结论。
现在演示失败重试。假设第一次修复不完整,测试仍然红,PEV 会触发重规划:
[步骤 4] 动作: run_command({"cmd":"pnpm test"}) 观察: FAIL exit=1 | 折扣为负数时也应返回原价 [VERIFY] 执行验收命令 pnpm test …… exit=1 ✗ 状态: failed(触发重规划) [PLAN] 重规划(第 2 轮): 上次失败原因:折扣为负数时也应返回原价 调整:边界条件从 rate === 0 改为 rate <= 0 [步骤 5] 动作: write_file({"path":"src/discount.ts","content":"..."}) [步骤 6] 动作: run_command({"cmd":"pnpm test"}) 观察: ok exit=0 | Tests 5 passed [VERIFY] 执行验收命令 pnpm test …… exit=0 ✓ 状态: success重规划时,summarizeFailure把上一轮的验证未通过项和最后几步喂回规划阶段,让模型“看着自己上次栽在哪”重新想办法。这就是 PEV 的价值:第一次失败不等于任务失败,而是触发一次带失败原因的重新规划。
5. 本篇常见错排查
接入和跑通过程中,报错集中在几个地方。逐个对照。
401 Unauthorized / invalid api key。最常见。先确认环境变量真的注入了:echo $TAOTOKEN_API_KEY有输出且没有多余空格。再确认请求头字段名对——Anthropic 格式用x-api-key,OpenAI 格式用Authorization: Bearer。如果 Key 是从控制台复制的,注意别把前后空白带进去。401 基本就是 Key 本身或注入方式的问题,跟 Query Loop 无关。
local proxy failed / connection refused。这个报错通常出现在你把 Base URL 配成了本地地址,或者环境里残留了某个代理配置。检查aegis.config.yaml里model.baseUrl是不是https://taotoken.net/api,检查 shell 里有没有HTTP_PROXY/HTTPS_PROXY之类的变量干扰。清掉这些变量再试。如果用的是 Docker 沙箱,容器内访问外网需要确认网络策略允许出站。
reading 'choices' of undefined。这是响应解析阶段的错,说明你按 OpenAI 格式解析,但实际返回的是 Anthropic 格式(或反过来)。parseResponse里读data.choices[0]之前,先打印一次原始响应结构确认格式。TaoToken 通道两种格式都支持,但请求体和解析逻辑必须匹配。改parseResponse时把data.content和data.choices两条路径都考虑进去。
OAuth / authentication_error。如果你在 Claude Code 或某些客户端里配置,可能遇到 OAuth 相关的报错。这类客户端有时会优先走 OAuth 流程而不是 API Key。确认你配置的是 API Key 模式,Base URL 填https://taotoken.net/api,Key 填控制台创建的 Key。如果客户端强制 OAuth,换用直接发 HTTP 请求的方式验证通道。
工具调用参数校验失败。报错形如参数校验失败:...。这是defineTool里 zod 校验拦下的,说明模型传的参数不符合 schema。检查工具定义里的schema是否和description描述一致——模型是照着 description 生成参数的,描述含糊它就会传错。把参数类型、必填项、取值范围在 description 里写清楚。
达到最大步数仍未完成。不是报错,是maxSteps兜底触发。说明 Agent 在某个环节打转。打开 trace 看最后几步是不是在重复同一个动作。常见原因是工具返回的观察结果格式让模型无法理解,或者治理规则误伤了正常操作。先调大maxSteps到 50 观察,定位到打转点后再针对性修。
验证永远不通过。检查task.acceptance.command的路径和命令是否正确。如果验收命令是pnpm test,要确认沙箱工作目录里有package.json且测试脚本存在。另外注意CommandVerifier在没有任何验收标准时直接判失败——这是刻意的防自欺设计,一个没有验收标准的任务本身就是有问题的任务。
排查顺序建议:先 curl 验证通道,再跑单步工具,再跑完整 Query Loop,最后看 PEV 重规划。每一层单独验证,出问题能快速定位在哪一层。
6. 把 Key 收口,把闭环跑通
回到最开始那个问题:七个 PoC 为什么拼不起来?因为模型接入散、编排无主干、验证靠自述。这一篇给出的解法是三步——用 TaoToken 统一 Key 和 API 通道,把模型接入的差异隔离在一个适配器里;用 Query Loop 做编排主干,治理和 trace 以横切方式接入;用 PEV 闭环把“模型说完成”换成“验证说通过”。
配置和代码都可以直接复制去用。aegis.config.yaml里的模型段换成你的 Key 环境变量名,TaoTokenLLM适配器按你的请求格式微调,Query Loop 和 PEV 逻辑不用动。跑通之后你会发现,换模型只是改配置里一行name,加治理规则只是往 yaml 里加一条,这些才是平台该有的样子。
下一步可以做的:把LocalSandbox换成DockerSandbox上生产隔离,把InMemoryTracer换成 OTel 实现接监控后端,把maxReplans调大观察 PEV 在复杂任务上的表现。Key 管理和模型列表在控制台,接入细节看文档,需要长期跑编码和 Agent 任务的话可以了解下 Coding Plan。通道先跑通,闭环再打磨,顺序别反。