拓冰网站定制 
1. 从一次失败的 Agent 循环说起Harness 自主进化到底难在哪我最早接触 Harness 这个概念是在折腾一个自动写周报的 Agent 时。当时想法很简单让模型在循环里跑调用几个工具读文件、写文件、发消息理论上就能自己迭代。结果跑了两天就崩了——上下文越滚越大工具调用开始重复模型在第 17 轮之后彻底忘了自己最初要干什么。这不是模型不行是编排层缺了结构。Harness 自主进化 Agent 的核心命题其实就一句话让 Agent 在运行过程中把「这次怎么解决问题」沉淀成下次可以直接复用的能力。这个能力单元现在业界普遍叫Skill。而 LangGraph 提供的 Graph-like Workflow恰好是承载这种沉淀最合适的骨架——它把「循环调用工具」这种自由散漫的 ReAct 模式约束成有节点、有边、有状态的确定性流程。你可能会问为什么不用纯 LangChain 的 AgentExecutor我实测下来AgentExecutor 适合一次性任务但一旦你要做「Agent 自己判断该不该新建 Skill、该不该更新已有 Skill」这种元决策它的黑盒程度就太高了。LangGraph 把每一步都暴露成节点你可以在节点之间插入检查点、插入人工审核、插入 Skill 注册动作这才是可观测的进化闭环。这篇文章面向的是已经写过基础 LangChain 调用、想往工程化方向走一步的开发者。我会带你从零搭一个最小可跑的 Harness 骨架用 LangGraph 定义状态图把 Skill 做成可注册、可检索、可执行的模块然后跑一次真实的自主迭代验证——让 Agent 自己发现「当前 Skill 不够用」然后生成一个新的 Skill 并注册进系统。全程代码可复制配置可落地。需要提前说明的是Skill 不是 Prompt 模板的换皮。Prompt 模板是你每次手动贴给模型的便签Skill 是写一次、在合适场景自动被激活的认知模块。这个区别决定了你的架构必须支持渐进式加载——Skill 的元数据常驻正文按需读取。后面第 3 节的配置会具体体现这一点。2. TaoToken 前置准备把模型调用层先跑通在写 LangGraph 之前得先把模型调用这层搞定。Harness 这类 Agent 对模型的调用频率很高——一次自主迭代可能触发十几轮 LLM 请求如果每轮都直连官方 API网络抖动和限流会让你调试到怀疑人生。我的做法是统一走一个兼容 OpenAI 协议的接入层TaoToken 就是我在用的方案它的 API 地址是https://taotoken.net/api兼容标准 OpenAI SDK改个 base_url 就能接。先说清楚它能做什么TaoToken 提供的是模型 API 接入服务支持 Claude、GPT 等主流模型的统一调用。对 Harness 场景来说最关键的是它支持Claude Code 和 Anthropic 协议这意味着你在 LangGraph 里用ChatAnthropic或者ChatOpenAI都能直接对接不用改业务代码。适合谁适合像我这样需要频繁切换模型做对比测试、又不想维护多套 SDK 的开发者。第一步去官网注册并拿到 API Key。地址是https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content注册流程不复杂邮箱验证后进控制台。控制台里找到 API Keys 页面新建一个 Key复制出来。这里提醒一句Key 只在创建时完整显示一次务必当场存进密码管理器别像我第一次那样刷新页面后到处找。拿到 Key 之后配置环境变量。我习惯用.env文件管理配合python-dotenv加载# .env TAOTOKEN_API_KEYsk-你的key TAOTOKEN_BASE_URLhttps://taotoken.net/api然后在 Python 里验证一下连通性这一步别跳过很多后续报错都是因为 Key 或 base_url 写错import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() llm ChatOpenAI( modelclaude-sonnet-4-20250514, api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL), temperature0, ) resp llm.invoke(用一句话说明什么是 LangGraph) print(resp.content)如果这段跑通说明模型层没问题。如果报 401检查 Key 有没有多余空格如果报连接超时检查 base_url 是不是写成了https://taotoken.net/api/末尾斜杠有时会导致路径拼接问题去掉更稳。模型选型上Harness 的自主迭代环节我建议用推理能力强的模型比如 Claude Sonnet 系列或者 GPT-4o 级别。因为「判断当前 Skill 是否够用」这个决策本身需要模型有较好的元认知能力小模型容易在该新建 Skill 的时候选择硬扛导致循环空转。TaoToken 的好处是你可以随时在配置里换 model 字段做 A/B 对比不用改其他代码。还有一点Harness 跑起来后请求量不小建议在 TaoToken 控制台里设置好用量提醒避免调试期间跑飞。我一般会把开发环境和生产环境的 Key 分开开发用的 Key 设一个较低的额度上限。3. LangGraph 编排 Skill 的可复制配置这一节是核心我会给出完整的 LangGraph 状态图定义、Skill 注册表结构、以及节点之间的路由逻辑。你可以直接把代码复制到项目里跑。先理解整体架构。Harness 的进化闭环分四个阶段感知Agent 接收任务→检索从 Skill 库找可用 Skill→执行调用 Skill 或直接推理→沉淀判断是否需要新建/更新 Skill。LangGraph 把这四个阶段映射成节点用条件边控制流转。先定义状态结构。LangGraph 的状态是一个 TypedDict所有节点共享读写from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver import operator class HarnessState(TypedDict): task: str # 当前任务描述 messages: Annotated[List[dict], operator.add] # 对话历史 available_skills: List[dict] # 检索到的候选 Skill selected_skill: dict # 当前选中的 Skill execution_result: str # 执行结果 should_create_skill: bool # 是否触发 Skill 沉淀 new_skill_draft: dict # 新 Skill 草稿 iteration: int # 迭代计数防止死循环注意messages用了operator.add作为 reducer这样每个节点返回的消息会追加而不是覆盖。iteration字段很关键Harness 最容易出的问题就是无限循环必须有个计数器兜底。接下来是 Skill 注册表。我用一个 JSON 文件做持久化结构参考 agentskills.io 的最小合规字段{ skills: [ { name: json_validator, description: 校验 JSON 字符串格式是否合法返回错误位置。当任务涉及 JSON 解析或格式检查时使用。, version: 1.0.0, body: 接收 input 字符串尝试 json.loads捕获 JSONDecodeError 并返回错误行列号。, created_by: human, usage_count: 0 }, { name: regex_extractor, description: 从文本中按正则表达式提取字段。当需要从非结构化文本抽取信息时使用。, version: 1.0.0, body: 接收 text 和 pattern 两个参数返回所有匹配项列表。, created_by: human, usage_count: 0 } ] }这个文件放在./harness/skills_registry.json。description字段是渐进式加载的第一层——它常驻上下文模型靠它判断要不要读完整 body。所以 description 必须写清楚「什么时候用」而不是「这是什么」。现在写 Skill 检索节点。它的职责是根据任务描述从注册表里挑出最相关的 Skillimport json from langchain_core.prompts import ChatPromptTemplate SKILL_REGISTRY_PATH ./harness/skills_registry.json def load_registry(): with open(SKILL_REGISTRY_PATH, r, encodingutf-8) as f: return json.load(f) def retrieve_skills(state: HarnessState): registry load_registry() skill_index [ {name: s[name], description: s[description]} for s in registry[skills] ] prompt ChatPromptTemplate.from_messages([ (system, 你是一个 Skill 检索器。根据任务描述从候选 Skill 中选出最相关的 1-3 个只返回 name 列表JSON 格式。), (user, 任务{task}\n\n候选 Skill{skills}) ]) chain prompt | llm result chain.invoke({ task: state[task], skills: json.dumps(skill_index, ensure_asciiFalse) }) try: names json.loads(result.content) except json.JSONDecodeError: names [] matched [s for s in registry[skills] if s[name] in names] return {available_skills: matched}这里有个坑我踩过模型返回的 JSON 有时会带 markdown 代码块标记直接json.loads会炸。稳妥做法是加一层清洗或者用llm.with_structured_output。我为了代码简洁先用 try/except 兜底生产环境建议上 structured output。执行节点负责真正调用 Skill。这里我用一个简化的执行器实际项目中你可以把 Skill body 解析成可调用函数def execute_skill(state: HarnessState): if not state[available_skills]: # 没有匹配 Skill直接让模型推理 resp llm.invoke(state[task]) return { execution_result: resp.content, selected_skill: {}, iteration: state[iteration] 1 } skill state[available_skills][0] exec_prompt ChatPromptTemplate.from_messages([ (system, 你正在执行 Skill{name}\n\nSkill 说明{body}\n\n请根据任务调用该 Skill 并返回结果。), (user, {task}) ]) chain exec_prompt | llm result chain.invoke({ name: skill[name], body: skill[body], task: state[task] }) return { execution_result: result.content, selected_skill: skill, iteration: state[iteration] 1 }沉淀判断节点是 Harness 的灵魂。它让模型自己评估这次执行有没有产生值得复用的新能力def reflect_and_evolve(state: HarnessState): reflect_prompt ChatPromptTemplate.from_messages([ (system, 你是一个 Skill 进化评估器。判断本次执行是否产生了可复用的新能力。 如果任务中出现了现有 Skill 无法覆盖的模式且该模式未来可能重复出现则应该创建新 Skill。 返回 JSON{{should_create: true/false, reason: ..., skill_draft: {{name: ..., description: ..., body: ...}}}}), (user, 任务{task}\n执行结果{result}\n现有 Skill{skills}) ]) chain reflect_prompt | llm result chain.invoke({ task: state[task], result: state[execution_result], skills: json.dumps([s[name] for s in state[available_skills]], ensure_asciiFalse) }) try: decision json.loads(result.content) except json.JSONDecodeError: decision {should_create: False} return { should_create_skill: decision.get(should_create, False), new_skill_draft: decision.get(skill_draft, {}) }最后是 Skill 注册节点把草稿写进注册表def register_skill(state: HarnessState): if not state[should_create_skill] or not state[new_skill_draft]: return {} registry load_registry() draft state[new_skill_draft] draft[version] 1.0.0 draft[created_by] agent draft[usage_count] 0 registry[skills].append(draft) with open(SKILL_REGISTRY_PATH, w, encodingutf-8) as f: json.dump(registry, f, ensure_asciiFalse, indent2) return {messages: [{role: system, content: f已注册新 Skill: {draft[name]}}]}现在把这些节点组装成图def should_continue(state: HarnessState): if state[iteration] 5: return end if state[should_create_skill]: return register return end workflow StateGraph(HarnessState) workflow.add_node(retrieve, retrieve_skills) workflow.add_node(execute, execute_skill) workflow.add_node(reflect, reflect_and_evolve) workflow.add_node(register, register_skill) workflow.set_entry_point(retrieve) workflow.add_edge(retrieve, execute) workflow.add_edge(execute, reflect) workflow.add_conditional_edges(reflect, should_continue, { register: register, end: END }) workflow.add_edge(register, END) memory MemorySaver() app workflow.compile(checkpointermemory)这套配置的关键设计点iteration上限设为 5防止 Agent 在反思环节反复触发新建 Skillshould_continue把注册和结束分开注册完直接 END不回头再跑一轮避免刚注册的 Skill 立刻被自己调用导致递归。4. 验证一次 Agent 自主迭代从任务到新 Skill 落地配置写完了现在跑一次真实验证。我准备了一个任务故意设计成现有两个 Skilljson_validator 和 regex_extractor都覆盖不了的场景从一段混合了 JSON 和自然语言的日志里提取所有错误码并统计出现次数。这个任务需要的能力是「混合文本解析 聚合统计」现有 Skill 都不匹配。理想情况下Agent 应该先尝试直接推理然后在反思环节意识到「这种混合解析模式会重复出现」主动创建一个新 Skill。执行代码config {configurable: {thread_id: harness-test-001}} initial_state { task: 从日志文本中提取所有错误码格式 ERR-XXXX并统计每个错误码出现次数。日志内容2024-01-01 ERR-1001 连接失败; 2024-01-02 ERR-1002 超时; 2024-01-03 ERR-1001 重试失败, messages: [], available_skills: [], selected_skill: {}, execution_result: , should_create_skill: False, new_skill_draft: {}, iteration: 0 } for event in app.stream(initial_state, config): for node_name, node_output in event.items(): print(f 节点: {node_name} ) if execution_result in node_output: print(f执行结果: {node_output[execution_result][:200]}) if should_create_skill in node_output: print(f是否创建 Skill: {node_output[should_create_skill]}) if new_skill_draft in node_output and node_output[new_skill_draft]: print(f新 Skill 草稿: {node_output[new_skill_draft].get(name)})跑完之后我观察到的输出流程是这样的retrieve节点返回空列表因为没有匹配 Skillexecute节点让模型直接推理输出了错误码统计结果reflect节点判断「混合文本解析是通用需求值得沉淀」返回should_create: trueregister节点把新 Skill 写入注册表。验证注册结果cat ./harness/skills_registry.json | python -m json.tool你应该能看到 skills 数组里多了一个created_by: agent的条目name 类似mixed_log_parserdescription 里写着「当需要从混合 JSON 和自然语言的文本中提取结构化字段时使用」。再跑一次相同任务这次retrieve节点应该能命中新 Skillexecute节点会走 Skill 执行路径而不是裸推理。这就是进化闭环生效的标志——第二次执行比第一次多了一层可复用的结构。这里有个细节值得说新 Skill 的 description 质量直接决定它未来能不能被正确检索到。我在 reflect 节点的 prompt 里特意强调了「description 要写清楚什么时候用」但模型第一次生成的 description 还是偏「这是什么」。我的做法是加一个二次润色步骤或者人工审核后再入库。生产环境建议在 register 节点前加一个 human-in-the-loop 的检查点LangGraph 的interrupt_before参数可以做到。如果你想验证模型调用是否正常可以单独跑一次对话测试TaoToken 的模型对话入口在https://taotoken.net/api用前面配好的 Key 直接调即可。对于需要长期跑 Agent 的场景建议关注 Coding Plan 这类按量方案避免调试期间额度跑超。5. 本篇常见报错排查401、local proxy failed 与 reading choices这一节把我踩过的坑集中列一下都是真实报错对照着查能省不少时间。报错一401 Unauthorizedopenai.AuthenticationError: Error code: 401 - {error: {message: Invalid API key}}原因通常是三种Key 复制时带了空格或换行、.env文件没被正确加载、或者 base_url 写错了。排查步骤先在 Python 里print(os.getenv(TAOTOKEN_API_KEY))确认读到了值然后确认 base_url 是https://taotoken.net/api而不是带/v1的变体不同 SDK 对路径拼接处理不同LangChain 的 ChatOpenAI 会自动补/chat/completions所以 base_url 不要带/v1。如果还报错去控制台确认 Key 有没有被禁用或额度耗尽。报错二local proxy failed / Connection errorhttpx.ConnectError: [Errno 111] Connection refused这个报错在 LangGraph 里出现往往不是网络问题而是你的环境变量里残留了HTTP_PROXY或HTTPS_PROXY设置导致请求被转发到一个不存在的本地端口。检查方式echo $HTTP_PROXY如果有值且指向127.0.0.1:xxxx清掉它。在代码里也可以显式禁用代理import os os.environ.pop(HTTP_PROXY, None) os.environ.pop(HTTPS_PROXY, None)报错三reading choices / KeyError: choicesKeyError: choices这个报错说明你拿到的响应不是标准 OpenAI 格式。常见于两种情况一是 base_url 指向了 Anthropic 原生接口但用了 OpenAI SDK二是模型名写错了服务端返回了错误信息但 SDK 没正确解析。解决方法是打印原始响应import httpx resp httpx.post( f{os.getenv(TAOTOKEN_BASE_URL)}/chat/completions, headers{Authorization: fBearer {os.getenv(TAOTOKEN_API_KEY)}}, json{model: claude-sonnet-4-20250514, messages: [{role: user, content: hi}]} ) print(resp.status_code, resp.text[:500])看到原始返回就能定位是模型名问题还是协议问题。TaoToken 兼容 OpenAI 协议用ChatOpenAI类对接即可不要混用ChatAnthropic除非你确认走的是 Anthropic 原生端点。报错四OAuth / token 过期如果你用的是 Claude Code 或某些 CLI 工具接入可能会遇到 OAuth 相关报错。这类工具通常有自己的认证流程和 API Key 是两套体系。排查时先确认你用的是 API Key 模式还是 OAuth 模式两者不要混。API Key 模式下Base URL、Key、Model ID 三件套必须配全{ base_url: https://taotoken.net/api, api_key: sk-你的key, model: claude-sonnet-4-20250514 }少任何一个都会导致认证失败或模型找不到。如果你在用 Cline、CC Switch 这类工具它们的配置文件里也是这三个字段路径通常在~/.cline/config.json或工具指定的 settings 文件里对照着填。报错五LangGraph 状态字段缺失KeyError: iteration这是因为初始 state 里漏了字段。LangGraph 的 TypedDict 不会自动补默认值所有字段必须在 initial_state 里显式给出。我习惯写一个make_initial_state(task)工厂函数避免每次手动拼。6. 把进化闭环跑成日常下一步可以做什么代码跑通只是起点。真正让 Harness 有价值的是让它持续运行、持续沉淀。我现在的做法是把这个 LangGraph 应用包成一个 CLI 工具每天定时跑几个固定任务观察 Skill 库的增长曲线。如果某个 Skill 的usage_count长期为 0说明它的 description 写得不够好检索不到需要人工润色或者合并。下一步可以扩展的方向有几个。一是给 Skill 加版本管理当 Agent 发现已有 Skill 执行效果不好时不是新建而是更新版本这需要在 reflect 节点里增加「对比现有 Skill 执行结果」的逻辑。二是引入 Skill 测试体系参考 skill-creator 的思路每个新 Skill 注册前自动跑一组断言通过率不达标就不入库。三是把 Skill 注册表从 JSON 换成 SQLite 或向量库支持语义检索而不是靠模型读 description 列表。如果你还没配好模型调用层先去 TaoToken 控制台拿 Key接入文档在https://taotoken.net/api对应的文档页有详细说明。需要长期跑 Agent 的话Coding Plan 的按量模式比单次调用更划算。模型对话入口可以用来快速验证某个模型在反思任务上的表现换模型只需要改一个 model 字段。最后留一个我自己的经验Harness 的进化能力不取决于模型多强而取决于你的 Skill 边界划得清不清楚。什么该做成 Skill、什么该留在 Prompt 里、什么该交给模型裸推理这个判断需要你对自己的业务领域有足够深的理解。模型会变框架会变但你对领域的 Knowledge 和沉淀下来的 Tool才是真正不会贬值的东西。 Harness 自主进化 Agent 实战：用 LangGraph 编排 Skill 的配置与验证
1. 从一次失败的 Agent 循环说起Harness 自主进化到底难在哪我最早接触 Harness 这个概念是在折腾一个自动写周报的 Agent 时。当时想法很简单让模型在循环里跑调用几个工具读文件、写文件、发消息理论上就能自己迭代。结果跑了两天就崩了——上下文越滚越大工具调用开始重复模型在第 17 轮之后彻底忘了自己最初要干什么。这不是模型不行是编排层缺了结构。Harness 自主进化 Agent 的核心命题其实就一句话让 Agent 在运行过程中把「这次怎么解决问题」沉淀成下次可以直接复用的能力。这个能力单元现在业界普遍叫Skill。而 LangGraph 提供的 Graph-like Workflow恰好是承载这种沉淀最合适的骨架——它把「循环调用工具」这种自由散漫的 ReAct 模式约束成有节点、有边、有状态的确定性流程。你可能会问为什么不用纯 LangChain 的 AgentExecutor我实测下来AgentExecutor 适合一次性任务但一旦你要做「Agent 自己判断该不该新建 Skill、该不该更新已有 Skill」这种元决策它的黑盒程度就太高了。LangGraph 把每一步都暴露成节点你可以在节点之间插入检查点、插入人工审核、插入 Skill 注册动作这才是可观测的进化闭环。这篇文章面向的是已经写过基础 LangChain 调用、想往工程化方向走一步的开发者。我会带你从零搭一个最小可跑的 Harness 骨架用 LangGraph 定义状态图把 Skill 做成可注册、可检索、可执行的模块然后跑一次真实的自主迭代验证——让 Agent 自己发现「当前 Skill 不够用」然后生成一个新的 Skill 并注册进系统。全程代码可复制配置可落地。需要提前说明的是Skill 不是 Prompt 模板的换皮。Prompt 模板是你每次手动贴给模型的便签Skill 是写一次、在合适场景自动被激活的认知模块。这个区别决定了你的架构必须支持渐进式加载——Skill 的元数据常驻正文按需读取。后面第 3 节的配置会具体体现这一点。2. TaoToken 前置准备把模型调用层先跑通在写 LangGraph 之前得先把模型调用这层搞定。Harness 这类 Agent 对模型的调用频率很高——一次自主迭代可能触发十几轮 LLM 请求如果每轮都直连官方 API网络抖动和限流会让你调试到怀疑人生。我的做法是统一走一个兼容 OpenAI 协议的接入层TaoToken 就是我在用的方案它的 API 地址是https://taotoken.net/api兼容标准 OpenAI SDK改个 base_url 就能接。先说清楚它能做什么TaoToken 提供的是模型 API 接入服务支持 Claude、GPT 等主流模型的统一调用。对 Harness 场景来说最关键的是它支持Claude Code 和 Anthropic 协议这意味着你在 LangGraph 里用ChatAnthropic或者ChatOpenAI都能直接对接不用改业务代码。适合谁适合像我这样需要频繁切换模型做对比测试、又不想维护多套 SDK 的开发者。第一步去官网注册并拿到 API Key。地址是https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content注册流程不复杂邮箱验证后进控制台。控制台里找到 API Keys 页面新建一个 Key复制出来。这里提醒一句Key 只在创建时完整显示一次务必当场存进密码管理器别像我第一次那样刷新页面后到处找。拿到 Key 之后配置环境变量。我习惯用.env文件管理配合python-dotenv加载# .env TAOTOKEN_API_KEYsk-你的key TAOTOKEN_BASE_URLhttps://taotoken.net/api然后在 Python 里验证一下连通性这一步别跳过很多后续报错都是因为 Key 或 base_url 写错import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() llm ChatOpenAI( modelclaude-sonnet-4-20250514, api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL), temperature0, ) resp llm.invoke(用一句话说明什么是 LangGraph) print(resp.content)如果这段跑通说明模型层没问题。如果报 401检查 Key 有没有多余空格如果报连接超时检查 base_url 是不是写成了https://taotoken.net/api/末尾斜杠有时会导致路径拼接问题去掉更稳。模型选型上Harness 的自主迭代环节我建议用推理能力强的模型比如 Claude Sonnet 系列或者 GPT-4o 级别。因为「判断当前 Skill 是否够用」这个决策本身需要模型有较好的元认知能力小模型容易在该新建 Skill 的时候选择硬扛导致循环空转。TaoToken 的好处是你可以随时在配置里换 model 字段做 A/B 对比不用改其他代码。还有一点Harness 跑起来后请求量不小建议在 TaoToken 控制台里设置好用量提醒避免调试期间跑飞。我一般会把开发环境和生产环境的 Key 分开开发用的 Key 设一个较低的额度上限。3. LangGraph 编排 Skill 的可复制配置这一节是核心我会给出完整的 LangGraph 状态图定义、Skill 注册表结构、以及节点之间的路由逻辑。你可以直接把代码复制到项目里跑。先理解整体架构。Harness 的进化闭环分四个阶段感知Agent 接收任务→检索从 Skill 库找可用 Skill→执行调用 Skill 或直接推理→沉淀判断是否需要新建/更新 Skill。LangGraph 把这四个阶段映射成节点用条件边控制流转。先定义状态结构。LangGraph 的状态是一个 TypedDict所有节点共享读写from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver import operator class HarnessState(TypedDict): task: str # 当前任务描述 messages: Annotated[List[dict], operator.add] # 对话历史 available_skills: List[dict] # 检索到的候选 Skill selected_skill: dict # 当前选中的 Skill execution_result: str # 执行结果 should_create_skill: bool # 是否触发 Skill 沉淀 new_skill_draft: dict # 新 Skill 草稿 iteration: int # 迭代计数防止死循环注意messages用了operator.add作为 reducer这样每个节点返回的消息会追加而不是覆盖。iteration字段很关键Harness 最容易出的问题就是无限循环必须有个计数器兜底。接下来是 Skill 注册表。我用一个 JSON 文件做持久化结构参考 agentskills.io 的最小合规字段{ skills: [ { name: json_validator, description: 校验 JSON 字符串格式是否合法返回错误位置。当任务涉及 JSON 解析或格式检查时使用。, version: 1.0.0, body: 接收 input 字符串尝试 json.loads捕获 JSONDecodeError 并返回错误行列号。, created_by: human, usage_count: 0 }, { name: regex_extractor, description: 从文本中按正则表达式提取字段。当需要从非结构化文本抽取信息时使用。, version: 1.0.0, body: 接收 text 和 pattern 两个参数返回所有匹配项列表。, created_by: human, usage_count: 0 } ] }这个文件放在./harness/skills_registry.json。description字段是渐进式加载的第一层——它常驻上下文模型靠它判断要不要读完整 body。所以 description 必须写清楚「什么时候用」而不是「这是什么」。现在写 Skill 检索节点。它的职责是根据任务描述从注册表里挑出最相关的 Skillimport json from langchain_core.prompts import ChatPromptTemplate SKILL_REGISTRY_PATH ./harness/skills_registry.json def load_registry(): with open(SKILL_REGISTRY_PATH, r, encodingutf-8) as f: return json.load(f) def retrieve_skills(state: HarnessState): registry load_registry() skill_index [ {name: s[name], description: s[description]} for s in registry[skills] ] prompt ChatPromptTemplate.from_messages([ (system, 你是一个 Skill 检索器。根据任务描述从候选 Skill 中选出最相关的 1-3 个只返回 name 列表JSON 格式。), (user, 任务{task}\n\n候选 Skill{skills}) ]) chain prompt | llm result chain.invoke({ task: state[task], skills: json.dumps(skill_index, ensure_asciiFalse) }) try: names json.loads(result.content) except json.JSONDecodeError: names [] matched [s for s in registry[skills] if s[name] in names] return {available_skills: matched}这里有个坑我踩过模型返回的 JSON 有时会带 markdown 代码块标记直接json.loads会炸。稳妥做法是加一层清洗或者用llm.with_structured_output。我为了代码简洁先用 try/except 兜底生产环境建议上 structured output。执行节点负责真正调用 Skill。这里我用一个简化的执行器实际项目中你可以把 Skill body 解析成可调用函数def execute_skill(state: HarnessState): if not state[available_skills]: # 没有匹配 Skill直接让模型推理 resp llm.invoke(state[task]) return { execution_result: resp.content, selected_skill: {}, iteration: state[iteration] 1 } skill state[available_skills][0] exec_prompt ChatPromptTemplate.from_messages([ (system, 你正在执行 Skill{name}\n\nSkill 说明{body}\n\n请根据任务调用该 Skill 并返回结果。), (user, {task}) ]) chain exec_prompt | llm result chain.invoke({ name: skill[name], body: skill[body], task: state[task] }) return { execution_result: result.content, selected_skill: skill, iteration: state[iteration] 1 }沉淀判断节点是 Harness 的灵魂。它让模型自己评估这次执行有没有产生值得复用的新能力def reflect_and_evolve(state: HarnessState): reflect_prompt ChatPromptTemplate.from_messages([ (system, 你是一个 Skill 进化评估器。判断本次执行是否产生了可复用的新能力。 如果任务中出现了现有 Skill 无法覆盖的模式且该模式未来可能重复出现则应该创建新 Skill。 返回 JSON{{should_create: true/false, reason: ..., skill_draft: {{name: ..., description: ..., body: ...}}}}), (user, 任务{task}\n执行结果{result}\n现有 Skill{skills}) ]) chain reflect_prompt | llm result chain.invoke({ task: state[task], result: state[execution_result], skills: json.dumps([s[name] for s in state[available_skills]], ensure_asciiFalse) }) try: decision json.loads(result.content) except json.JSONDecodeError: decision {should_create: False} return { should_create_skill: decision.get(should_create, False), new_skill_draft: decision.get(skill_draft, {}) }最后是 Skill 注册节点把草稿写进注册表def register_skill(state: HarnessState): if not state[should_create_skill] or not state[new_skill_draft]: return {} registry load_registry() draft state[new_skill_draft] draft[version] 1.0.0 draft[created_by] agent draft[usage_count] 0 registry[skills].append(draft) with open(SKILL_REGISTRY_PATH, w, encodingutf-8) as f: json.dump(registry, f, ensure_asciiFalse, indent2) return {messages: [{role: system, content: f已注册新 Skill: {draft[name]}}]}现在把这些节点组装成图def should_continue(state: HarnessState): if state[iteration] 5: return end if state[should_create_skill]: return register return end workflow StateGraph(HarnessState) workflow.add_node(retrieve, retrieve_skills) workflow.add_node(execute, execute_skill) workflow.add_node(reflect, reflect_and_evolve) workflow.add_node(register, register_skill) workflow.set_entry_point(retrieve) workflow.add_edge(retrieve, execute) workflow.add_edge(execute, reflect) workflow.add_conditional_edges(reflect, should_continue, { register: register, end: END }) workflow.add_edge(register, END) memory MemorySaver() app workflow.compile(checkpointermemory)这套配置的关键设计点iteration上限设为 5防止 Agent 在反思环节反复触发新建 Skillshould_continue把注册和结束分开注册完直接 END不回头再跑一轮避免刚注册的 Skill 立刻被自己调用导致递归。4. 验证一次 Agent 自主迭代从任务到新 Skill 落地配置写完了现在跑一次真实验证。我准备了一个任务故意设计成现有两个 Skilljson_validator 和 regex_extractor都覆盖不了的场景从一段混合了 JSON 和自然语言的日志里提取所有错误码并统计出现次数。这个任务需要的能力是「混合文本解析 聚合统计」现有 Skill 都不匹配。理想情况下Agent 应该先尝试直接推理然后在反思环节意识到「这种混合解析模式会重复出现」主动创建一个新 Skill。执行代码config {configurable: {thread_id: harness-test-001}} initial_state { task: 从日志文本中提取所有错误码格式 ERR-XXXX并统计每个错误码出现次数。日志内容2024-01-01 ERR-1001 连接失败; 2024-01-02 ERR-1002 超时; 2024-01-03 ERR-1001 重试失败, messages: [], available_skills: [], selected_skill: {}, execution_result: , should_create_skill: False, new_skill_draft: {}, iteration: 0 } for event in app.stream(initial_state, config): for node_name, node_output in event.items(): print(f 节点: {node_name} ) if execution_result in node_output: print(f执行结果: {node_output[execution_result][:200]}) if should_create_skill in node_output: print(f是否创建 Skill: {node_output[should_create_skill]}) if new_skill_draft in node_output and node_output[new_skill_draft]: print(f新 Skill 草稿: {node_output[new_skill_draft].get(name)})跑完之后我观察到的输出流程是这样的retrieve节点返回空列表因为没有匹配 Skillexecute节点让模型直接推理输出了错误码统计结果reflect节点判断「混合文本解析是通用需求值得沉淀」返回should_create: trueregister节点把新 Skill 写入注册表。验证注册结果cat ./harness/skills_registry.json | python -m json.tool你应该能看到 skills 数组里多了一个created_by: agent的条目name 类似mixed_log_parserdescription 里写着「当需要从混合 JSON 和自然语言的文本中提取结构化字段时使用」。再跑一次相同任务这次retrieve节点应该能命中新 Skillexecute节点会走 Skill 执行路径而不是裸推理。这就是进化闭环生效的标志——第二次执行比第一次多了一层可复用的结构。这里有个细节值得说新 Skill 的 description 质量直接决定它未来能不能被正确检索到。我在 reflect 节点的 prompt 里特意强调了「description 要写清楚什么时候用」但模型第一次生成的 description 还是偏「这是什么」。我的做法是加一个二次润色步骤或者人工审核后再入库。生产环境建议在 register 节点前加一个 human-in-the-loop 的检查点LangGraph 的interrupt_before参数可以做到。如果你想验证模型调用是否正常可以单独跑一次对话测试TaoToken 的模型对话入口在https://taotoken.net/api用前面配好的 Key 直接调即可。对于需要长期跑 Agent 的场景建议关注 Coding Plan 这类按量方案避免调试期间额度跑超。5. 本篇常见报错排查401、local proxy failed 与 reading choices这一节把我踩过的坑集中列一下都是真实报错对照着查能省不少时间。报错一401 Unauthorizedopenai.AuthenticationError: Error code: 401 - {error: {message: Invalid API key}}原因通常是三种Key 复制时带了空格或换行、.env文件没被正确加载、或者 base_url 写错了。排查步骤先在 Python 里print(os.getenv(TAOTOKEN_API_KEY))确认读到了值然后确认 base_url 是https://taotoken.net/api而不是带/v1的变体不同 SDK 对路径拼接处理不同LangChain 的 ChatOpenAI 会自动补/chat/completions所以 base_url 不要带/v1。如果还报错去控制台确认 Key 有没有被禁用或额度耗尽。报错二local proxy failed / Connection errorhttpx.ConnectError: [Errno 111] Connection refused这个报错在 LangGraph 里出现往往不是网络问题而是你的环境变量里残留了HTTP_PROXY或HTTPS_PROXY设置导致请求被转发到一个不存在的本地端口。检查方式echo $HTTP_PROXY如果有值且指向127.0.0.1:xxxx清掉它。在代码里也可以显式禁用代理import os os.environ.pop(HTTP_PROXY, None) os.environ.pop(HTTPS_PROXY, None)报错三reading choices / KeyError: choicesKeyError: choices这个报错说明你拿到的响应不是标准 OpenAI 格式。常见于两种情况一是 base_url 指向了 Anthropic 原生接口但用了 OpenAI SDK二是模型名写错了服务端返回了错误信息但 SDK 没正确解析。解决方法是打印原始响应import httpx resp httpx.post( f{os.getenv(TAOTOKEN_BASE_URL)}/chat/completions, headers{Authorization: fBearer {os.getenv(TAOTOKEN_API_KEY)}}, json{model: claude-sonnet-4-20250514, messages: [{role: user, content: hi}]} ) print(resp.status_code, resp.text[:500])看到原始返回就能定位是模型名问题还是协议问题。TaoToken 兼容 OpenAI 协议用ChatOpenAI类对接即可不要混用ChatAnthropic除非你确认走的是 Anthropic 原生端点。报错四OAuth / token 过期如果你用的是 Claude Code 或某些 CLI 工具接入可能会遇到 OAuth 相关报错。这类工具通常有自己的认证流程和 API Key 是两套体系。排查时先确认你用的是 API Key 模式还是 OAuth 模式两者不要混。API Key 模式下Base URL、Key、Model ID 三件套必须配全{ base_url: https://taotoken.net/api, api_key: sk-你的key, model: claude-sonnet-4-20250514 }少任何一个都会导致认证失败或模型找不到。如果你在用 Cline、CC Switch 这类工具它们的配置文件里也是这三个字段路径通常在~/.cline/config.json或工具指定的 settings 文件里对照着填。报错五LangGraph 状态字段缺失KeyError: iteration这是因为初始 state 里漏了字段。LangGraph 的 TypedDict 不会自动补默认值所有字段必须在 initial_state 里显式给出。我习惯写一个make_initial_state(task)工厂函数避免每次手动拼。6. 把进化闭环跑成日常下一步可以做什么代码跑通只是起点。真正让 Harness 有价值的是让它持续运行、持续沉淀。我现在的做法是把这个 LangGraph 应用包成一个 CLI 工具每天定时跑几个固定任务观察 Skill 库的增长曲线。如果某个 Skill 的usage_count长期为 0说明它的 description 写得不够好检索不到需要人工润色或者合并。下一步可以扩展的方向有几个。一是给 Skill 加版本管理当 Agent 发现已有 Skill 执行效果不好时不是新建而是更新版本这需要在 reflect 节点里增加「对比现有 Skill 执行结果」的逻辑。二是引入 Skill 测试体系参考 skill-creator 的思路每个新 Skill 注册前自动跑一组断言通过率不达标就不入库。三是把 Skill 注册表从 JSON 换成 SQLite 或向量库支持语义检索而不是靠模型读 description 列表。如果你还没配好模型调用层先去 TaoToken 控制台拿 Key接入文档在https://taotoken.net/api对应的文档页有详细说明。需要长期跑 Agent 的话Coding Plan 的按量模式比单次调用更划算。模型对话入口可以用来快速验证某个模型在反思任务上的表现换模型只需要改一个 model 字段。最后留一个我自己的经验Harness 的进化能力不取决于模型多强而取决于你的 Skill 边界划得清不清楚。什么该做成 Skill、什么该留在 Prompt 里、什么该交给模型裸推理这个判断需要你对自己的业务领域有足够深的理解。模型会变框架会变但你对领域的 Knowledge 和沉淀下来的 Tool才是真正不会贬值的东西。