当前位置：首页 > news >正文

生产级 AI Agent Harness 实战：6 大核心组件与完整 Python 实现

news 2026/7/26 1:11:06

大多数 AI agent 问题并不是模型问题，而是 harness 问题。

你的 demo 能跑。agent 找到了正确的文件，调用了正确的工具，产出了让所有人点头认可的东西。你把它上线了。然后它在生产环境跑到第 10 步时出问题了。模型忘记了三步之前自己做过什么。一次 tool call 静默失败，而 agent 还是继续往下走。context 里塞满了噪声。整个系统崩了。

你切换了模型。GPT-5，然后 Claude，再到 Gemini。根本没有发生本质变化。

因为模型从来都不是瓶颈。包裹在它外面的基础设施才是。

LangChain 在 Terminal Bench 2.0 上直接证明了这一点。Terminal Bench 2.0 是一个包含 89 个任务的 benchmark，覆盖软件工程、debugging、machine learning 和 security。起始分数：52.8%，排在前 30 名之外。他们没有做任何模型改动。全程使用同一套 weights。改变的是围绕模型的基础设施：orchestration loop、context management、verification middleware 和 reasoning budget allocation。结果：66.5%，提升 13.7 分，进入排行榜前 5。仅靠基础设施决策就提升了 25 个名次。

这种基础设施有一个名字：agent harness。有意识地工程化构建它，也有一个名字：harness engineering。

公式是Agent = Model + Harness。模型正在越来越商品化。Harness engineering 才是真正构建产品的地方。它决定了一个 agent 是只能在 demo 中工作，还是能够可靠处理 50 步生产任务。

继续读下去，你将准确理解 harness 是什么、六个基础组件如何工作，以及每个组件背后的 Python 代码。

Agent Harness 到底是什么？

agent harness 是除模型本身之外的所有代码、配置和执行逻辑。

这是一个干净的定义。但要真正理解它的含义，用操作系统类比会更合适。

一个原始 LLM 就像一颗没有操作系统的 CPU。CPU 可以计算，可以处理指令。但它不能自己从磁盘读取数据，不能在屏幕上绘制窗口，也不能直接与网络接口通信。你需要 OS 来管理这些资源、调度任务、处理 I/O，并保护内存边界。

harness 就是你的语言模型的 OS：

Context window = RAM。快速、有限、昂贵。放进去的内容决定了模型能推理什么。
External databases = disk。容量大、速度慢、持久化。模型可以读写，但需要一个系统来管理访问。
Tools = device drivers。模型与外部系统交互所通过的接口。
Harness = operating system。管理什么进入 context、持久化 memory、路由 tool calls、从错误中恢复，并执行安全约束。

没有 harness，模型无法在多次交互之间维护状态，无法执行代码，无法访问实时知识，无法安装 packages，无法在失败后恢复，也无法与其他 agents 协作。我们认为“强大 agent”具备的每一种能力都来自 harness，而不是模型。

Beren Millidge 在 2023 年将 scaffolded LLMs 称为“natural language computers”，从而形式化了这一点。计算的基本单位是 NLOP（Natural Language Operation），大致相当于生成 100 个 tokens。当前 frontier models 的运行速度约为每秒 1–10 NLOPs，比 CPU 指令慢几个数量级，但每次操作的表达能力要强得多。

agent harness 是一整套软件基础设施，它把无状态的语言模型转变为能够执行多步骤任务的 autonomous agent。它管理 orchestration loop、tool execution、context window、memory persistence、error recovery、safety enforcement 和 output verification。没有 harness，LLM 可以回答问题，但不能采取行动、维护状态或从失败中恢复。第 1 部分和第 2 部分覆盖的 11 个组件代表了 production-grade agent harness 的完整结构，并为每个组件提供 Python 实现。Cursor、Claude Code、Codex 和 Windsurf 都是围绕 large language models 构建的 agent harness 示例。

Harness vs Framework：一个值得区分的概念

在继续之前先澄清一个词，因为这两个术语经常被混用，但它们不应该被混用。

Frameworks，比如 LangChain、LangGraph、AutoGen 和 CrewAI，给你的是抽象：state graphs、chains、memory containers、retrievers。你这个人类架构师需要把它们连接起来。它们的基本假设是：开发者会把这些组件配置成一个 agent。

harness则来自相反方向。没有组装步骤。它交付的是一个可工作的 agent。其核心是一个带有 tool registry 和 permission layer 的 while loop，所有东西都已经连接好。你提供目标，harness 处理剩下的事情。

framework 是为人类组装 agent 而构建的。harness 是为 agent 自己完成任务而构建的。

这个区别很重要，因为其收敛模式非常醒目。Cursor、Codex、Claude Code 和 Windsurf 都从一个具体问题出发（让模型在真实 repository 中编写和编辑代码），并且各自独立地收敛到非常相似的 harness 架构。这种收敛说明了架构真正需要什么，无论你是否使用了某个 framework 来构建它。

围绕模型的三层工程

可以把它想象成同心圆：Prompt Engineering 在中心（模型读取什么），Context Engineering 在中间（什么内容在什么时候进入窗口），Harness Engineering 在外层（完整的应用基础设施：tool execution、state persistence、error recovery、verification、safety 和 multi-agent coordination）。

大多数团队几乎把全部投入都放在最内层。这就是为什么无论换什么模型，他们的 agents 都以同样的方式失败。那句正在流传的 harness 引言是：“If you’re not the model, you’re the harness.” 让 agent 变得有用的一切都在那里。

Harness 才是产品的证据

Terminal Bench 的结果很惊人，但并非个例。看看还有哪些内容已经被公开记录。

Meta-Harness 发现。一个研究团队构建了一个自动化 harness optimizer，使用 LLM 自身在 harness configurations 空间中搜索。在 Claude Opus 4.6 上运行且没有模型改动的情况下，它在 Terminal Bench 2.0 上达到 76.4%，在所有 Opus 4.6 agents 中排名第二。同一个 meta-harness 运行 Claude Haiku 4.5，在所有 Haiku 4.5 agents 中排名第一。harness 比模型层级更重要。

Vercel 工具削减。Vercel 的 text-to-SQL agent 最初有 15+ 个专用工具。他们移除了其中 80%，只留下一个 bash command execution tool。成功率从 80% 提升到 100%。执行时间下降 3.5 倍。Token usage 下降 37%。关键洞察是：他们因为不信任模型自由推理，反而限制了模型的 reasoning。

复合失败数学。如果单个 agent step 有 99% 的可靠性，而一个任务需要 10 个连续步骤，端到端成功率是 90.4%。到了 50 步：60.5%。即使单步“99% reliable”，10 个任务中也会有 4 个失败。真实生产任务经常跨越 30–60 次 tool calls。在这个尺度上，你衡量的是 error handling 有多好，而不是模型有多聪明。

Credit: OpenAI

Context rot。Chroma 测试了 18 个 frontier models，包括 Claude Opus 4、GPT-4.1 和 Gemini 2.5 Pro。每一个模型都会随着 context 增长而退化，远早于达到 token limits。Stanford 的 “Lost in the Middle” 研究显示，放在 context window 中间的内容准确率下降超过 30%。管理 context 完全是 harness 的责任。

Manus 的五次重写。Manus 是 2025 年爆火并以约 20 亿美元被收购的 autonomous agent，它在六个月内重建了自己的 agent framework 五次。每次重写都是移除复杂性，而不是增加复杂性。随着模型变得更好，harness 变得更薄。

Claude Code 的 512,000 行 harness。当 Anthropic 在 2026 年 3 月意外暴露 Claude Code 的源代码时，泄露内容显示有 1,906 个文件、513,000 行 TypeScript。那就是 harness。模型 weights 并不在那里。竞争护城河完全在周围的工程里。

两个产品，同一个模型，结果天差地别。harness 才是差异化因素。

机器内部：前六个组件

harness 不是单一事物。它是由 11 个相互联锁的组件组成的系统。这里是六个基础组件：控制 agent 如何运行的组件。

(Harness layer components) Credits: OpenAI

组件 1：编排循环

orchestration loop 实现 Thought-Action-Observation（TAO）循环，也被称为 ReAct loop（Reason + Act）。循环本身并不智能：智能存在于模型中，harness 管理这个循环。终止会在以下情况下触发：输出中没有 tool calls（任务完成）、达到 turn limit、token budget 耗尽、guardrail tripwire 被触发，或用户中断。

import anthropicfrom typing importAnyclient = anthropic.Anthropic()defrun_agent( system_prompt: str, user_message: str, tools: list[dict], max_turns: int = 20,) -> str: messages = [{"role": "user", "content": user_message}] for turn inrange(max_turns): response = client.messages.create( model="claude-opus-4-7", max_tokens=4096, system=system_prompt, tools=tools, messages=messages, ) # Append model response to history messages.append({"role": "assistant", "content": response.content}) # Check termination: no tool calls means task is complete tool_calls = [b for b in response.content if b.type == "tool_use"] ifnot tool_calls: # Extract final text response text_blocks = [b.text for b in response.content ifhasattr(b, "text")] return"\n".join(text_blocks) # Execute tools and collect results tool_results = [] for call in tool_calls: result = execute_tool(call.name, call.input) tool_results.append({ "type": "tool_result", "tool_use_id": call.id, "content": str(result), }) messages.append({"role": "user", "content": tool_results}) return"Max turns reached without completion."

一个关键细节：循环的终止条件应该在 harness 中检查，而不是委托给模型。模型经常“想要”继续。harness 负责强制停止。

组件 2：工具层

tools 并不神秘。从 harness 的视角看，一个 tool 就是描述模型可以调用什么的 JSON schema，再配上一段在被调用时实际运行的可执行代码。tool layer 管理 registration、validation、sandboxed execution 和 result formatting。

工具数量悖论是真实存在的：更多 tools 会降低性能。每个 tool schema 都会消耗 context tokens，并增加模型的决策空间。Vercel 的案例表明，移除 80% 的 tools 改善了结果。设计时应追求最小可行工具集。

import functoolsimport inspectimport jsonfrom typing importCallable, AnyTOOL_REGISTRY: dict[str, dict] = {}deftool(name: str = None, description: str = ""): """Decorator that registers a function as an agent tool.""" defdecorator(fn: Callable) -> Callable: tool_name = name or fn.__name__ schema = _infer_schema(fn) TOOL_REGISTRY[tool_name] = { "name": tool_name, "description": description or fn.__doc__ or"", "input_schema": schema, "fn": fn, } @functools.wraps(fn) defwrapper(*args, **kwargs): return fn(*args, **kwargs) return wrapper return decoratordefexecute_tool(name: str, args: dict) -> Any: """Validate and execute a tool call in a sandboxed context.""" if name notin TOOL_REGISTRY: returnf"Error: Unknown tool '{name}'" entry = TOOL_REGISTRY[name] # Validate required arguments exist required = entry["input_schema"].get("required", []) missing = [r for r in required if r notin args] if missing: returnf"Error: Missing required arguments: {missing}" try: return entry["fn"](**args) except Exception as e: returnf"Error executing {name}: {str(e)}"defget_tool_schemas() -> list[dict]: """Return tool schemas in Anthropic API format.""" return [ { "name": v["name"], "description": v["description"], "input_schema": v["input_schema"], } for v in TOOL_REGISTRY.values() ]# Example usage@tool(description="Read a file from the workspace")defread_file(path: str) -> str: """Read and return file contents.""" withopen(path) as f: return f.read()@tool(description="Write content to a file")defwrite_file(path: str, content: str) -> str: """Write content to a file and return confirmation.""" withopen(path, "w") as f: f.write(content) returnf"Written {len(content)} characters to {path}"

_infer_schema函数使用 Python 的inspectmodule 根据 type annotations 自动构建 JSON schema，从而让 tool schemas 与 implementation 保持同步。

Tools vs Skills。Tools 是通用原语：读取文件、运行 bash、调用 API。Skills 是编码为 Markdown 文件（AGENTS.md、CLAUDE.md）的组织知识，并被注入 system prompt。Skills 是团队特定的：commit message conventions、branching rules、如何解读你的测试失败。每个 harness 都会内置 tools。Skills 则是让通用 harness 适配你的 workflow 的方式。

内置 skills。每个生产级 harness 还会提供一套不可协商的高级能力基线，而不只是原始 primitives。不只是“read a file”，而是“make a git commit with a proper message”、“open a pull request”、“run the test suite and interpret what the failures mean”。这些内置 skills 正是将通用 LLM wrapper 与 coding harness 区分开来的东西。Claude Code、Cursor、Codex 都开箱即带这些能力。如果你的 agent 不能在你每次手写 instructions 的情况下自己 commit code 或 run tests，那它还不能算是真正的 harness。一个重要的实现注意事项：内置 primitives 应该使用纯 standard library code，不依赖任何 framework。agent 需要能够在它落地的任何环境中执行它们。

Lifecycle Hooks。pre-tool hook 在执行前触发，可以 allow、deny 或 modify 调用。post-tool hook 在执行后触发，用于 logging 和 auditing。Hooks 是 extensibility seam：无需触碰 harness internals，就能围绕 tool execution 添加自定义逻辑。企业安全采用 third-party harnesses 时，就是通过 hooks 在边界处注入 approval workflows 或 audit logging。

from typing importCallable, OptionalclassHookResult: def__init__(self, allowed: bool, modified_args: dict = None, reason: str = ""): self.allowed = allowed self.modified_args = modified_args self.reason = reasonclassHookRegistry: def__init__(self): self._pre_hooks: list[Callable] = [] self._post_hooks: list[Callable] = [] defregister_pre(self, fn: Callable): self._pre_hooks.append(fn) defregister_post(self, fn: Callable): self._post_hooks.append(fn) defrun_pre(self, tool_name: str, args: dict) -> HookResult: """Run all pre-tool hooks. First deny wins.""" for hook inself._pre_hooks: result = hook(tool_name, args) ifnot result.allowed: return result if result.modified_args: args = result.modified_args return HookResult(allowed=True, modified_args=args) defrun_post(self, tool_name: str, result: str): """Run all post-tool hooks (audit/logging only, cannot block).""" for hook inself._post_hooks: hook(tool_name, result)# Example: audit hook that logs every tool callhook_registry = HookRegistry()hook_registry.register_post( lambda name, result: print(f"[AUDIT] {name} -> {result[:80]}"))

组件 3：记忆系统

语言模型是无状态的。memory systems 让 agents 能够跨 turns 和 sessions 持久化。三种实用类型：

短期：存在于 context window 中。当前任务、最近的 tool outputs、活跃对话。速度快但临时。

长期：外部存储（vector database 或 filesystem）。事实、偏好、过去的决策。通过 semantic search 检索。

情景式：记录特定过去交互的完整 context：情境、采取的方法、是否有效。episodic memory 让 agent 能够说出“上次我尝试 X 失败了，因为 Y”。

Claude Code 的三层层级结构是一个实用的生产模式：always-loaded index（轻量 memory pointers）、on-demand topic files（相关时获取）以及 raw search 作为 fallback。

import jsonimport osfrom pathlib import Pathfrom datetime import datetimeimport anthropicMEMORY_DIR = Path(".agent_memory")MEMORY_INDEX = MEMORY_DIR / "MEMORY.md"EPISODES_DIR = MEMORY_DIR / "episodes"definit_memory(): MEMORY_DIR.mkdir(exist_ok=True) EPISODES_DIR.mkdir(exist_ok=True) ifnot MEMORY_INDEX.exists(): MEMORY_INDEX.write_text("# Agent Memory Index\n\n")defsave_fact(key: str, content: str, topic: str = "general"): """Save a fact to long-term memory.""" topic_file = MEMORY_DIR / f"{topic}.md" # Append to topic file withopen(topic_file, "a") as f: f.write(f"\n## {key}\n{content}\n") # Update index withopen(MEMORY_INDEX, "a") as f: f.write(f"- [{key}]({topic}.md): {topic}\n")defsave_episode(task: str, outcome: str, approach: str): """Save an episodic memory of what worked (or didn't).""" episode_id = datetime.now().strftime("%Y%m%d_%H%M%S") episode_file = EPISODES_DIR / f"{episode_id}.md" episode_file.write_text( f"# Episode {episode_id}\n\n" f"**Task:** {task}\n\n" f"**Approach:** {approach}\n\n" f"**Outcome:** {outcome}\n\n" f"**Date:** {datetime.now().isoformat()}\n" )defretrieve_relevant_memory(query: str, max_results: int = 3) -> str: """Retrieve relevant memory using Claude as the retriever.""" ifnot MEMORY_INDEX.exists(): return"" index_content = MEMORY_INDEX.read_text() # Lazy-load: only read index first, fetch topic files on demand client = anthropic.Anthropic() response = client.messages.create( model="claude-haiku-4-5-20251001", max_tokens=512, messages=[{ "role": "user", "content": ( f"Given this memory index:\n{index_content}\n\n" f"For query: '{query}'\n" f"List up to {max_results} most relevant memory file names." ) }] ) # In production, parse response and load the named files return response.content[0].text

关键设计洞察是：index 始终加载（轻量 pointers），但实际 memory files 只有在 query 使其相关时才加载。这就是 Claude Code 通过 just-in-time retrieval 实现高达 95% context reduction 的方式。

组件 4：上下文管理

Context rot 是真实且可测量的。随着输入长度增长，所有被测试模型在每个长度增量上都会退化，远早于达到 token limits。Stanford 的 “Lost in the Middle” 研究是标志性发现：在 20-document context 中放在第 5 到第 15 位的信息，相比放在第 1 位或第 20 位，准确率下降超过 30%。

管理它的五种生产策略：

Compaction：总结旧的 conversation turns，用高密度 summaries 替代冗长历史。
Observation masking：隐藏旧的 tool outputs，同时保留 tool calls 本身（JetBrains 的 Junie 使用这种方式）。
JIT retrieval：只在需要时获取信息，而不是 upfront 加载全部内容。
Sub-agent delegation：将重型探索工作卸载给隔离的 subagents，由它们返回压缩 summaries。
Structured note-taking：让 agent 维护一个持续更新的todo.md，每一步都更新，把当前目标推入 recency bias zone。

import tiktokenfrom typing importAnyCONTEXT_BUDGET = 100_000# tokensCOMPACTION_THRESHOLD = 0.75# compact when 75% fulldefestimate_tokens(text: str) -> int: """Fast token estimation without API call.""" enc = tiktoken.get_encoding("cl100k_base") returnlen(enc.encode(text))defformat_messages_for_context(messages: list[dict]) -> int: """Estimate total tokens in message history.""" total = 0 for msg in messages: content = msg.get("content", "") ifisinstance(content, list): content = " ".join( block.get("text", "") orstr(block.get("content", "")) for block in content ) total += estimate_tokens(str(content)) return totaldefcompact_messages( messages: list[dict], system_prompt: str, keep_recent: int = 6,) -> list[dict]: """ Compress old message history while preserving recent context. Summarizes everything except the last `keep_recent` turns. """ iflen(messages) return messages old_messages = messages[:-keep_recent] recent_messages = messages[-keep_recent:] # Build a summary of old context old_text = "\n".join( f"{m['role'].upper()}: {m['content']}" for m in old_messages ifisinstance(m.get("content"), str) ) import anthropic client = anthropic.Anthropic() summary_response = client.messages.create( model="claude-haiku-4-5-20251001", max_tokens=1024, messages=[{ "role": "user", "content": ( f"Summarize this agent conversation history in 3-5 key points. " f"Focus on decisions made and important findings:\n\n{old_text[:8000]}" ) }] ) summary = summary_response.content[0].text # Replace old messages with compact summary summary_message = { "role": "user", "content": f"[CONTEXT SUMMARY - earlier conversation]\n{summary}" } return [summary_message] + recent_messagesdefinject_with_position_awareness( context_items: list[dict], token_budget: int,) -> list[dict]: """ Inject context items with awareness of the lost-in-the-middle effect. Most critical items go at start and end; secondary items fill the middle. """ critical = [item for item in context_items if item.get("priority") == "high"] secondary = [item for item in context_items if item.get("priority") != "high"] # Critical items at the boundary (start and end), secondary in middle ordered = ( critical[:len(critical)//2] + secondary + critical[len(critical)//2:] ) return ordered

ACON framework（2025）显示，它能在保持 95%+ accuracy 的同时，将 peak token usage 降低 26–54%。这可能就是任务完成与撞上 context wall 之间的区别。

组件 5：Prompt 构建

system prompt 不是一个静态字符串。它是一条 pipeline。

启动时，harness 会沿 ancestor directory tree 查找 instruction files（CLAUDE.md、AGENTS.md、.cursorrules，或你的 harness convention 使用的任何文件）。它按从 root 到 current directory 的顺序读取这些文件，让更本地的文件具有更高 specificity。project-level AGENTS.md 会添加到 global one，而不是替换它。每个文件都贡献一层。

这里有一条关键排序规则：static content 先放，dynamic content 后放。static prefix（safety rules、agent identity、tool descriptions）需要在各 turns 之间保持一致，这样模型的 KV cache 才能保持 warm。把 dynamic memory files 插到 prompt 前面会破坏 prefix caching，并可能让 inference cost 增加 10 倍。OpenAI 的 Codex 使用五级 priority stack 进行 conflict resolution。Claude Code 的 AGENTS.md cascade 通过 directory hierarchy 实现了同样的效果。

关键原则：当 layers 冲突时，harness 需要一个明确的 resolution order。没有显式 priority，模型会做出任意选择。

from dataclasses import dataclass, fieldfrom typing importOptional@dataclassclassPromptLayer: priority: int # Lower number = higher priority source: str # Where this came from (e.g., "system", "agent_md", "user") content: str token_estimate: int = 0classHierarchicalPromptBuilder: def__init__(self, token_budget: int = 8000): self.token_budget = token_budget self.layers: list[PromptLayer] = [] defadd_layer(self, priority: int, source: str, content: str): """Add a prompt layer with explicit priority.""" tokens = estimate_tokens(content) self.layers.append(PromptLayer(priority, source, content, tokens)) defbuild(self) -> str: """ Assemble the final system prompt respecting priority and budget. Higher priority layers always included; lower priority layers dropped if over budget. """ # Sort by priority (ascending = highest priority first) sorted_layers = sorted(self.layers, key=lambda l: l.priority) included = [] tokens_used = 0 for layer in sorted_layers: if tokens_used + layer.token_estimate included.append(layer) tokens_used += layer.token_estimate else: # Skip lower priority content that won't fit print(f"Dropped layer '{layer.source}' (budget exceeded)") # Build the actual prompt, with conflict resolution comment parts = [] for layer in included: parts.append(f"# From: {layer.source} (priority {layer.priority})\n{layer.content}") return"\n\n---\n\n".join(parts)# Usagebuilder = HierarchicalPromptBuilder(token_budget=8000)builder.add_layer(1, "safety_system", "Never execute destructive operations without confirmation.")builder.add_layer(2, "agent_identity", "You are a software engineering assistant.")builder.add_layer(3, "project_context", load_agents_md()) # From AGENTS.md filebuilder.add_layer(4, "task_instructions", task_prompt)builder.add_layer(5, "user_preferences", user_config)system_prompt = builder.build()

AGENTS.md cascading pattern 意味着每个 subdirectory 都可以添加 local context，而不会覆盖 global rules。/project/AGENTS.md中的文件会添加到来自/AGENTS.md的 context，而不是替换它。Priority ordering 保证 global safety rules 不会被 local task instructions 覆盖。

组件 6：输出解析

现代 harnesses 使用原生 tool calling APIs，而不是对 free text 做 regex parsing。API responses 中结构化的tool_callsfield 包含经过验证的 JSON，harness 可以不做任何字符串操作就进行路由。

决策树很简单：存在 tool calls 表示 loop 继续；没有 tool calls 表示 final answer；handoff marker 表示切换 agents。

from pydantic import BaseModel, ValidationErrorfrom typing importUnionimport jsonclassFinalAnswer(BaseModel): content: str confidence: str = "high"classToolCall(BaseModel): name: str arguments: dictclassParsedOutput(BaseModel): output_type: str # "tool_calls", "final_answer", "handoff" tool_calls: list[ToolCall] = [] final_answer: Optional[FinalAnswer] = None handoff_target: Optional[str] = Nonedefparse_model_output(response) -> ParsedOutput: """ Parse model response into structured routing decision. Falls back gracefully rather than crashing on unexpected formats. """ tool_calls = [ block for block in response.content ifhasattr(block, "type") and block.type == "tool_use" ] if tool_calls: parsed_calls = [ ToolCall(name=call.name, arguments=call.input) for call in tool_calls ] return ParsedOutput(output_type="tool_calls", tool_calls=parsed_calls) # Check for handoff marker in text text_blocks = [ block.text for block in response.content ifhasattr(block, "text") ] full_text = "\n".join(text_blocks) if""in full_text: target = extract_between(full_text, "", "") return ParsedOutput(output_type="handoff", handoff_target=target.strip()) return ParsedOutput( output_type="final_answer", final_answer=FinalAnswer(content=full_text) )defparse_with_retry( model_fn, prompt: str, output_schema: type[BaseModel], max_retries: int = 2,) -> BaseModel: """ Parse structured output with retry-and-error-feedback pattern. On failure, sends the error back to the model for self-correction. """ messages = [{"role": "user", "content": prompt}] for attempt inrange(max_retries + 1): raw_response = model_fn(messages) try: data = json.loads(extract_json(raw_response)) return output_schema(**data) except (json.JSONDecodeError, ValidationError) as e: if attempt == max_retries: raise # Feed the error back: model gets to correct itself messages.append({"role": "assistant", "content": raw_response}) messages.append({ "role": "user", "content": f"The output failed validation: {e}. Please fix it and return valid JSON." })

把 validation error 发回给模型，让它能够 self-correct，这通常比盲目 retry 更容易成功。

Harness 还没有完成

六个组件已经讲完。还有五个。而剩下的五个，正是大多数生产 agents 崩溃的地方。

前六个组件描述的是当事情顺利时 agent 如何运行。接下来的五个决定的是当事情不顺利时它能否存活。

组件 7：State Management。当你的 agent 在一个 60-turn task 的第 47 轮崩溃时，checkpointing 决定了你是失去一切，还是几秒内恢复。Append-only event logs、crash recovery 和 time-travel debugging。

组件 8：Error Handling。这就是改变一切的数学：单步 99% 可靠性，跨 50 个连续步骤，端到端成功率只有 60.5%。即使单个步骤几乎完美，10 个生产任务中也有 4 个会失败。改变这个数字的是 harness，而不是模型。

组件 9：Guardrails and Safety。模型决定尝试什么。tool system 决定什么被允许。立即停止的 tripwires。Dynamic bash classification，其中ls是 read-only，而rm -rf被完全阻止。

组件 10：Verification Loops。领导 Claude Code 开发的 Boris Cherny 记录过，给 agent 一种验证自己工作的方式，可以让最终输出质量提升 2–3 倍。不是靠更好的 prompting，而是靠运行真实测试，并把 failures 作为具体、可执行的输入反馈回去。

组件 11：Subagent Orchestration。agents 之间的每一次 handoff 都是有损压缩。三种执行模型、parallel agent coordination，以及如何围绕每次 handoff 都会产生的信息损失进行设计。

学AI大模型的正确顺序，千万不要搞错了

🤔2026年AI风口已来！各行各业的AI渗透肉眼可见，超多公司要么转型做AI相关产品，要么高薪挖AI技术人才，机遇直接摆在眼前！

有往AI方向发展，或者本身有后端编程基础的朋友，直接冲AI大模型应用开发转岗超合适！

就算暂时不打算转岗，了解大模型、RAG、Prompt、Agent这些热门概念，能上手做简单项目，也绝对是求职加分王🔋

📝给大家整理了超全最新的AI大模型应用开发学习清单和资料，手把手帮你快速入门！👇👇

学习路线:

✅大模型基础认知—大模型核心原理、发展历程、主流模型（GPT、文心一言等）特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架（LangChain等）实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经

以上6大模块，看似清晰好上手，实则每个部分都有扎实的核心内容需要吃透！

我把大模型的学习全流程已经整理📚好了！抓住AI时代风口，轻松解锁职业新可能，希望大家都能把握机遇，实现薪资/职业跃迁～