资讯详情

深入了解 · 专业解读

CAR三位一体框架《从0到1实现一个企业级Harness平台》——用TaoToken统一Key打通Control/Agency/Runtime

发布时间：2026/10/3 6:54:23
CAR三位一体框架《从0到1实现一个企业级Harness平台》——用TaoToken统一Key打通Control/Agency/Runtime 1. 为什么企业级 Harness 平台需要 CAR 三位一体框架如果你正在做 Agent 平台大概率遇到过这种局面模型能跑通 demo但一上生产就失控——工具乱调、预算爆表、出错没人知道、审计查不到记录。问题不在模型本身而在于你缺一个把「控制、能动性、运行时」捏合在一起的骨架。这个骨架就是 Harness而 CARControl / Agency / Runtime是它最实用的拆解方式。Control 层负责治理哪些动作允许、花多少钱、目标有没有跑偏、每一步有没有留痕。Agency 层负责自主在允许的边界内Agent 怎么推理、怎么拆任务、出错怎么自己兜底。Runtime 层负责执行事件怎么流转、状态怎么存、异构工具怎么统一调用。三者不是三个独立模块串起来而是同一个 Harness 实体的三个面向——就像同一个人的「守法意识、决策能力、身体执行」同时存在而不是先守法再决策再动手。这篇按「从 0 到 1 搭最小可运行骨架」来写模型调用层用 TaoToken 统一 Key 打通避免你在多家模型 API 之间来回切配置。目标很明确读完你能跑通一次端到端任务从 Control 下发、Agency 编排、Runtime 执行到回执全链路有日志、有预算、有降级。适合正在做 Agent 平台、智能客服、自动化运维、代码 Agent 的后端和平台工程师。2. TaoToken 统一 Key 接入把模型调用层先固定下来在搭 CAR 之前先把模型调用层固定住。否则你会在 Control 写一套 OpenAI 配置、Agency 写一套 Claude 配置、Runtime 再写一套Key 散落各处预算和审计根本没法统一。TaoToken 的作用就是提供一个统一的 API 通道和 Key让三层都走同一个出口。2.1 为什么模型层要单独抽出来CAR 三层里Control 要统计 token 消耗、Agency 要切换模型做降级、Runtime 要记录每次调用的耗时。如果每层各自持有不同厂商的 Key你会遇到三个麻烦预算统计口径不一致、降级时改配置要动多处、审计日志里模型来源对不上。把模型调用收敛成一个内部 client三层都调它问题就消失了。TaoToken 的 API 地址是https://taotoken.net/api兼容 OpenAI 风格的接口。你只需要一个 Key就能在 Control 的策略里引用模型 ID、在 Agency 的降级链里切换模型、在 Runtime 的工具总线里统一记录调用。2.2 拿到 Key 并写入环境变量先去控制台创建 API Keyhttps://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentconsole创建后不要硬编码进代码写进环境变量。Linux/macOSexport TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/apiWindows PowerShell$env:TAOTOKEN_API_KEYsk-你的key $env:TAOTOKEN_BASE_URLhttps://taotoken.net/api2.3 封装一个三层共用的模型 client下面这个 client 是后面 Control、Agency、Runtime 都会引用的基础件。它做了三件事统一走 TaoToken、返回 token 用量、抛出结构化错误供降级使用。# harness/model_client.py import os import time from openai import OpenAI class ModelClient: def __init__(self): self.client OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ.get(TAOTOKEN_BASE_URL, https://taotoken.net/api), ) def chat(self, model: str, messages: list, **kwargs) - dict: start time.time() try: resp self.client.chat.completions.create( modelmodel, messagesmessages, **kwargs, ) return { ok: True, content: resp.choices[0].message.content, tokens: resp.usage.total_tokens, model: model, duration_ms: int((time.time() - start) * 1000), } except Exception as e: return { ok: False, error_type: type(e).__name__, error: str(e), model: model, duration_ms: int((time.time() - start) * 1000), }这里model参数就是你在 TaoToken 上可用的模型 ID。Control 层做预算时读tokensAgency 层降级时根据error_type决定切哪个模型Runtime 层把duration_ms写进事件日志。一个 client 解决三层需求。2.4 验证模型通道是否通先跑一个最小请求确认 Key 和地址没问题from harness.model_client import ModelClient mc ModelClient() result mc.chat( modelgpt-4o-mini, messages[{role: user, content: 只回复两个字通了}], ) print(result)返回里ok为 True、content有内容、tokens是数字说明通道正常。如果这里就报 401先别往下搭 CAR回到 2.2 检查 Key 是否写对、环境变量是否在当前 shell 生效。模型 ID 具体有哪些可以在模型对话页确认https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmodels3. Control 层路由配置策略、预算、审计的可复制片段Control 层是 Harness 的治理面。它不阻止 Agent 干活而是给 Agent 划出「合法行为空间」。这一层要落地四件事许可性策略、预算控制、目标对齐、审计留痕。下面给出可直接复制的配置片段。3.1 用 JSON 定义许可性策略策略用声明式 JSON 写方便热更新和审计。放到harness/config/control_policy.json{ version: 1.0, rules: [ { id: deny_dangerous_sql, name: 禁止危险 SQL, priority: 100, condition: { AND: [ { field: action.type, operator: eq, value: execute_sql }, { field: action.params.query, operator: regex, value: (?i)(DROP|DELETE|TRUNCATE|ALTER) } ] }, action: deny, audit: true }, { id: review_large_transfer, name: 大额转账需审批, priority: 90, condition: { AND: [ { field: action.type, operator: eq, value: transfer }, { field: action.params.amount, operator: gt, value: 10000 } ] }, action: review, audit: true }, { id: allow_trusted_user, name: 高信任用户放行, priority: 70, condition: { AND: [ { field: user.trustLevel, operator: gte, value: 0.9 }, { field: action.riskLevel, operator: lte, value: 3 } ] }, action: allow, audit: false } ] }action有四种取值allow放行、deny拒绝、review转人工、warn告警但放行。优先级高的先匹配deny和review优先于allow。3.2 预算控制配置预算用 TOML 写放到harness/config/budget.toml[tokens] per_request 10000 per_hour 100000 per_day 1000000 [api_calls] per_minute 20 per_hour 500 per_day 5000 [time] per_request_ms 30000 per_task_ms 300000 [cost] per_request_usd 0.50 per_day_usd 50.0 per_month_usd 1000.03.3 Control 层加载与决策# harness/control.py import json import time import tomllib from pathlib import Path class ControlLayer: def __init__(self, policy_path: str, budget_path: str): self.rules json.loads(Path(policy_path).read_text())[rules] self.rules.sort(keylambda r: r[priority], reverseTrue) self.budget tomllib.loads(Path(budget_path).read_text()) self.records [] def check_action(self, action: dict, user: dict) - dict: ctx {action: action, user: user, time: {hour: time.localtime().tm_hour}} for rule in self.rules: if self._match(rule[condition], ctx): return {decision: rule[action], rule_id: rule[id], audit: rule.get(audit, False)} return {decision: allow, rule_id: None, audit: False} def check_budget(self, estimated: dict) - dict: violations [] if estimated.get(tokens, 0) self.budget[tokens][per_request]: violations.append(token_per_request_exceeded) if estimated.get(cost_usd, 0) self.budget[cost][per_request_usd]: violations.append(cost_per_request_exceeded) return {approved: len(violations) 0, violations: violations} def _match(self, cond: dict, ctx: dict) - bool: if AND in cond: return all(self._match(c, ctx) for c in cond[AND]) if OR in cond: return any(self._match(c, ctx) for c in cond[OR]) if NOT in cond: return not self._match(cond[NOT], ctx) actual self._get(ctx, cond[field]) op, val cond[operator], cond[value] if op eq: return actual val if op gt: return actual val if op gte: return actual val if op lt: return actual val if op lte: return actual val if op regex: import re return bool(re.search(val, str(actual))) return False def _get(self, obj, path): for k in path.split(.): obj obj.get(k) if isinstance(obj, dict) else None return obj3.4 审计日志用哈希链审计记录要防篡改用简单的哈希链即可# harness/audit.py import hashlib import json import time class AuditChain: def __init__(self): self.chain [{index: 0, event: {msg: genesis}, prev: 0 * 64, hash: }] self.chain[0][hash] self._hash(self.chain[0]) def append(self, event: dict): prev self.chain[-1] block { index: len(self.chain), event: {**event, ts: time.time()}, prev: prev[hash], hash: , } block[hash] self._hash(block) self.chain.append(block) return block def verify(self) - bool: for i in range(1, len(self.chain)): cur, prev self.chain[i], self.chain[i - 1] if cur[hash] ! self._hash(cur): return False if cur[prev] ! prev[hash]: return False return True def _hash(self, block): data f{block[index]}{json.dumps(block[event], sort_keysTrue)}{block[prev]} return hashlib.sha256(data.encode()).hexdigest()到这里 Control 层就齐了策略决定动作能不能做预算决定资源够不够审计决定事后查不查得到。这三样是后面 Agency 和 Runtime 的地基。4. Agency 层任务编排意图分解与降级链Agency 层解决的是「在 Control 划定的边界内Agent 怎么自主干活」。核心是三件事把高层意图拆成可执行子任务、在推理循环里感知边界、出错时按降级链自己兜底。4.1 意图分解成子目标树用户说「帮我搭一个电商网站」Agency 要把它拆成有依赖关系的子目标。下面这个分解器把意图转成 DAG# harness/agency/decomposer.py from dataclasses import dataclass, field import uuid dataclass class SubGoal: id: str field(default_factorylambda: str(uuid.uuid4())[:8]) desc: str priority: int 1 status: str pending deps: list field(default_factorylist) class IntentTree: def __init__(self, root: str): self.root root self.goals {} def add(self, goal: SubGoal) - str: self.goals[goal.id] goal return goal.id def ready(self) - list: out [] for g in self.goals.values(): if g.status ! pending: continue if all(self.goals[d].status completed for d in g.deps if d in self.goals): out.append(g) return sorted(out, keylambda g: g.priority, reverseTrue) def progress(self) - float: if not self.goals: return 0.0 done sum(1 for g in self.goals.values() if g.status completed) return done / len(self.goals) def decompose(intent: str) - IntentTree: tree IntentTree(intent) specs [ (需求分析, 5, []), (数据库 Schema, 4, [0]), (用户认证模块, 4, [1]), (商品管理模块, 3, [1]), (订单处理模块, 3, [1, 2]), (支付集成, 2, [4]), (前端 UI, 2, [2, 3, 4]), (集成测试, 1, [6, 5]), (部署上线, 1, [7]), ] ids [] for desc, pri, deps in specs: g SubGoal(descdesc, prioritypri) ids.append(tree.add(g)) for i, (_, _, deps) in enumerate(specs): tree.goals[ids[i]].deps [ids[d] for d in deps] return tree4.2 带边界感知的 ReAct 循环Agency 的推理循环和普通 ReAct 的区别在于每次要执行动作前先问 Control 层「这个动作允许吗」。被拒绝不是报错而是作为观察结果喂回给模型让它换个方案。# harness/agency/react_loop.py class ConstrainedReAct: def __init__(self, control, model_client, max_steps10): self.control control self.mc model_client self.max_steps max_steps async def run(self, question: str, tools: dict) - dict: context f问题: {question}\n for step in range(self.max_steps): # 1. 让模型决定下一步 plan self.mc.chat( modelgpt-4o-mini, messages[{role: user, content: context 输出 JSON: {thought, action, params}}], ) if not plan[ok]: return {success: False, error: plan[error]} # 2. 解析动作这里简化实际用 json 解析 action {type: tool_call, tool: search, params: {}} # 3. Control 层检查 decision self.control.check_action(action, user{trustLevel: 0.8}) if decision[decision] deny: context f步骤{step}: 动作被拒绝({decision[rule_id]})请换方案\n continue if decision[decision] review: return {success: False, need_human: True, reason: decision[rule_id]} # 4. 执行工具 tool_fn tools.get(action[tool]) observation await tool_fn(action[params]) if tool_fn else 工具不存在 context f步骤{step}: 动作{action[tool]} 观察{observation}\n return {success: False, error: 达到最大步数}4.3 降级链配置Agency 遇到模型报错、超时、预算不足时按降级链逐级兜底。用 JSON 配置{ degradation_chain: [ { level: L0, name: 重试, trigger: { error_type: RateLimitError, max_attempts: 3 }, action: retry_with_backoff }, { level: L1, name: 切换模型, trigger: { error_type: ModelError }, action: switch_model, fallback_map: { gpt-4o: gpt-4o-mini, claude-3-opus: claude-3-sonnet } }, { level: L2, name: 简化执行, trigger: { error_type: TimeoutError }, action: simplified_response }, { level: L3, name: 转人工, trigger: { severity: critical }, action: escalate_to_human } ] }降级链的关键是每一级失败后自动落到下一级而不是直接抛错给用户。L1 的fallback_map里模型 ID 都走 TaoToken 同一个 Key切换时不用改任何认证配置。5. Runtime 层执行器事件循环、状态管理与工具总线Runtime 是 Harness 的物理基础。Control 和 Agency 再完善没有 Runtime 执行都是纸上谈兵。这一层要落地三件事事件循环驱动 Agent 运转、事件溯源管理状态、工具总线统一异构调用。5.1 带优先级的异步事件循环Agent 运行时会产生多种事件用户消息、工具调用、工具结果、推理、响应、错误。不同事件优先级不同错误要插队处理。# harness/runtime/event_loop.py import asyncio from collections import defaultdict class EventLoop: PRIORITIES [critical, high, normal, low] def __init__(self, backpressure1000): self.queues {p: asyncio.Queue() for p in self.PRIORITIES} self.handlers {} self.running False self.backpressure backpressure self.processed 0 def on(self, event_type: str, handler): self.handlers[event_type] handler async def enqueue(self, event: dict): pending sum(q.qsize() for q in self.queues.values()) if pending self.backpressure and event[priority] low: return False await self.queues[event[priority]].put(event) return True async def start(self): self.running True while self.running: event await self._dequeue() if event is None: await asyncio.sleep(0.01) continue handler self.handlers.get(event[type]) if handler: try: await handler(event) self.processed 1 except Exception as e: await self.enqueue({ type: error, priority: high, data: {original: event.get(id), error: str(e)}, }) async def _dequeue(self): for p in self.PRIORITIES: if not self.queues[p].empty(): return await self.queues[p].get() return None def stop(self): self.running False5.2 事件溯源状态管理Agent 的状态不是直接改的而是由事件累积出来的。这样你可以回放任意时刻的状态也能 fork 出分支做 what-if 分析。# harness/runtime/state.py import copy class EventSourcedState: def __init__(self): self.events [] self.state {} self.reducers {} self.snapshots {} def register(self, event_type: str, reducer): self.reducers[event_type] reducer def apply(self, event: dict): self.events.append(event) reducer self.reducers.get(event[type]) if reducer: self.state reducer(self.state, event) if len(self.events) % 100 0: self.snapshots[len(self.events)] copy.deepcopy(self.state) return self.state def state_at(self, version: int) - dict: snap_v max((v for v in self.snapshots if v version), default0) state copy.deepcopy(self.snapshots.get(snap_v, {})) for e in self.events[snap_v:version]: r self.reducers.get(e[type]) if r: state r(state, e) return state def fork(self, from_version: int): new EventSourcedState() new.reducers self.reducers for e in self.events[:from_version]: new.apply(e) return new5.3 工具总线异构工具HTTP API、数据库、文件系统、人工接口统一成同一个调用接口中间件做日志、鉴权、限流。# harness/runtime/tool_bus.py import asyncio import time class ToolBus: def __init__(self): self.tools {} self.middleware [] def register(self, name: str, schema: dict, adapter): self.tools[name] {schema: schema, adapter: adapter} def use(self, mw): self.middleware.append(mw) async def execute(self, name: str, params: dict) - dict: tool self.tools.get(name) if not tool: return {success: False, error: funknown tool: {name}} async def core(): start time.time() try: timeout tool[schema].get(timeout, 30) result await asyncio.wait_for(tool[adapter](params), timeouttimeout) return {**result, duration_ms: int((time.time() - start) * 1000)} except asyncio.TimeoutError: return {success: False, error: timeout, duration_ms: int((time.time() - start) * 1000)} except Exception as e: return {success: False, error: str(e), duration_ms: int((time.time() - start) * 1000)} idx 0 async def run_mw(): nonlocal idx if idx len(self.middleware): return await core() mw self.middleware[idx] idx 1 return await mw(name, params, run_mw) return await run_mw()5.4 把三层串起来跑一次端到端任务现在把 Control、Agency、Runtime 组装起来跑一次完整任务# harness/main.py import asyncio from harness.control import ControlLayer from harness.audit import AuditChain from harness.model_client import ModelClient from harness.runtime.event_loop import EventLoop from harness.runtime.state import EventSourcedState from harness.runtime.tool_bus import ToolBus async def main(): # 初始化三层 control ControlLayer(harness/config/control_policy.json, harness/config/budget.toml) audit AuditChain() mc ModelClient() loop EventLoop() state EventSourcedState() bus ToolBus() # 注册状态 reducer state.register(user_message, lambda s, e: {**s, last_msg: e[data][content]}) state.register(tool_call, lambda s, e: {**s, last_tool: e[data][tool]}) # 注册工具 bus.register(search, {timeout: 10}, lambda p: asyncio.sleep(0.1) or {success: True, data: 搜索结果}) # 注册事件处理器 async def on_user_message(event): # 1. Control 检查预算 budget control.check_budget({tokens: 5000, cost_usd: 0.1}) if not budget[approved]: audit.append({type: budget_denied, violations: budget[violations]}) return # 2. 调模型 result mc.chat(gpt-4o-mini, [{role: user, content: event[data][content]}]) audit.append({type: model_call, ok: result[ok], tokens: result.get(tokens, 0)}) # 3. 执行工具 tool_result await bus.execute(search, {q: test}) state.apply({type: tool_call, data: {tool: search}}) print(f模型回复: {result.get(content, result.get(error))}) print(f工具结果: {tool_result}) loop.on(user_message, on_user_message) # 启动事件循环 task asyncio.create_task(loop.start()) # 下发任务 await loop.enqueue({ id: evt-1, type: user_message, priority: normal, data: {content: 帮我查一下订单状态}, }) await asyncio.sleep(2) loop.stop() await task # 验证审计链 print(f审计链完整: {audit.verify()}) print(f当前状态: {state.state}) asyncio.run(main())跑通后你会看到模型回复、工具结果、审计链完整、状态里有last_tool。这就是最小可运行的 Harness 骨架——Control 管住了预算和策略Agency 的推理循环在边界内工作Runtime 把事件、状态、工具都串起来了。6. 本篇常见报错排查搭的过程中最容易卡在几个地方这里按真实报错对照排查。6.1 401 Unauthorized现象ModelClient.chat返回ok: Falseerror_type: AuthenticationError错误信息里有 401。原因通常是三种Key 没写进环境变量、Key 复制时带了空格、base_url写错。排查顺序先在终端echo $TAOTOKEN_API_KEY确认有值再确认base_url是https://taotoken.net/api注意结尾没有多余的斜杠或/v1最后确认 Key 没有过期或被删除。如果用的是 IDE 内置终端环境变量可能没继承重启 IDE 或改用系统终端。6.2 local proxy failed / connection refused现象请求直接抛连接错误error_type: APIConnectionError。这类报错一般是本机网络配置问题不是 TaoToken 侧的问题。检查你的 shell 里有没有设置HTTP_PROXY/HTTPS_PROXY环境变量指向一个已经失效的本地端口。如果有unset HTTP_PROXY HTTPS_PROXY后再跑。另外确认本机能正常访问外网curl -I https://taotoken.net/api看有没有响应。6.3 reading choices of undefined现象resp.choices[0]报Cannot read properties of undefined。这是响应结构和你预期不一致。常见原因是模型 ID 写错了服务端返回的是错误对象而不是正常的 completion 结构。先打印完整响应resp mc.client.chat.completions.create(modelgpt-4o-mini, messages[...]) print(resp)如果返回里有error字段说明模型 ID 不对或该模型不可用。去模型对话页确认可用模型 ID别凭记忆写。6.4 OAuth / 认证方式混淆现象你按某个工具的文档配了 OAuth但 TaoToken 用的是 API Key两者对不上。TaoToken 的接入方式是 API Key Base URL不需要 OAuth 流程。如果你在用 Claude Code、Cline、Codex 这类工具配置项要写全三件套Base URLhttps://taotoken.net/apiAPI Key你的sk-开头的 KeyModel ID具体模型名如gpt-4o-mini或claude-3-sonnet以 Claude Code 的settings.json为例{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_API_KEY: sk-你的key, ANTHROPIC_MODEL: claude-3-sonnet } }Cline 的 MCP 配置里同理baseUrl、apiKey、model三个字段都要填。Codex 的auth.json里对应base_url、api_key、model。少填任何一个都会报认证失败或模型找不到。6.5 预算检查一直拒绝现象check_budget总是返回approved: False。先看violations里具体是哪条超了。如果是token_per_request_exceeded说明你估算的 token 超过了budget.toml里的per_request。要么调大配置要么在调用前先做一次粗略的 token 估算。注意per_hour和per_day是滑动窗口统计如果你在测试时反复跑很容易把小时额度打满等窗口滑过去再试。6.6 事件循环卡住不处理现象enqueue成功但 handler 不执行。检查loop.start()是不是真的在跑。如果你用asyncio.create_task启动主协程要await它或者await asyncio.sleep给它执行时间。另外确认enqueue的priority值是critical/high/normal/low之一写错的话会KeyError。背压阈值设太小也会导致低优先级事件被静默丢弃测试时先把backpressure调大。7. 继续往下走从最小骨架到生产可用到这里你已经有了一个能跑通端到端任务的最小 Harness。Control 有策略、预算、审计Agency 有意图分解、边界感知推理、降级链Runtime 有事件循环、事件溯源、工具总线。模型调用层统一走 TaoToken三层共用一个 Key预算和审计口径一致。接下来可以按这个顺序加固先把审计链接到你的日志系统让每次 Control 决策都可查再把降级链的每一级都配上真实模型 ID用 TaoToken 的模型对话页验证每个模型可用然后给工具总线加鉴权和限流中间件最后把事件溯源的状态快照持久化到数据库支持重启恢复。如果你要长期跑编码类 Agent 或复杂任务编排建议直接上 Coding Plan省去自己管额度和并发https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentcoding-plan接入文档在这里配置项和错误码都有说明https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentdocAPI Key 管理在控制台https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentapi-keys我自己的经验是先把 Control 的审计跑通再往上加功能因为后面所有排障都依赖审计日志。没有审计的 Harness出问题只能靠猜。

相关新闻