企业官网建设流程全解析

4) 编排层: 任务拆解、流程控制、多Agent协作 3) 反馈层: 结果校验、错误归因、LLM自评 2) 执行层: 工具调用、文件读写、API调用、沙箱 1) 记忆层: 规则文件、状态快照、技能定义

import json import hashlib from pathlib import Path from dataclasses import dataclass, field, asdict from typing import Optional, Any from datetime import datetime

@dataclass
class Checkpoint:
"""Agent execution snapshot for crash recovery"""
task_id: str
step: int
status: str # running, paused, completed, failed
context_snapshot: dict
tool_results: list
error_log: list
created_at: str = field(default_factory=lambda: datetime.now().isoformat())

def fingerprint(self) -> str: """Generate snapshot fingerprint for integrity verification""" raw = json.dumps(self.context_snapshot, sort_keys=True) return hashlib.sha256(raw.encode()).hexdigest()[:16]

class MemoryLayer:
"""Memory Layer: manages rules, state, and checkpoints"""

def __init__(self, workspace: str = "./agent_workspace"): self.workspace = Path(workspace) self.workspace.mkdir(parents=True, exist_ok=True) self._rules_cache: dict = {} self._state: dict = {} def load_rules(self, rule_files: list[str]) -> str: """Load rule files and return concatenated rules text""" rules = [] for rf in rule_files: path = self.workspace / rf if path.exists(): content = path.read_text(encoding="utf-8") self._rules_cache[rf] = content rules.append(f"### Rule: {rf}\n{content}") return "\n\n".join(rules) def save_checkpoint(self, checkpoint: Checkpoint): """Save checkpoint for crash recovery""" cp_dir = self.workspace / "checkpoints" cp_dir.mkdir(exist_ok=True) cp_file = cp_dir / f"{checkpoint.task_id}_step{checkpoint.step}.json" data = asdict(checkpoint) data["fingerprint"] = checkpoint.fingerprint() # Atomic write: write to temp file first, then rename tmp_file = cp_file.with_suffix(".tmp") tmp_file.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") tmp_file.rename(cp_file) def load_latest_checkpoint(self, task_id: str) -> Optional[Checkpoint]: """Load latest checkpoint for resume""" cp_dir = self.workspace / "checkpoints" if not cp_dir.exists(): return None candidates = sorted(cp_dir.glob(f"{task_id}_step*.json"), reverse=True) if not candidates: return None data = json.loads(candidates[0].read_text(encoding="utf-8")) # Verify integrity stored_fp = data.pop("fingerprint", "") ctx = data.get("context_snapshot", {}) raw = json.dumps(ctx, sort_keys=True) computed_fp = hashlib.sha256(raw.encode()).hexdigest()[:16] if stored_fp != computed_fp: raise ValueError(f"Checkpoint {candidates[0].name} data corrupted: " f"expected {stored_fp}, got {computed_fp}") return Checkpoint(**data) def update_state(self, key: str, value: Any): """Update agent runtime state""" self._state[key] = value state_file = self.workspace / "state.json" state_file.write_text(json.dumps(self._state, ensure_ascii=False, indent=2), encoding="utf-8")

这段代码做了三件事:

1. 规则文件加载: 把仓库级规则(类似AGENTS.md)统一管理,Agent启动时一次性注入上下文
2. 检查点持久化: 每一步执行完都保存快照,包括任务进度、工具调用结果和上下文状态。原子写入保证不会写出半截文件
3. 完整性校验: 用SHA256指纹验证检查点没被篡改或损坏

你可能会问:每步都存检查点,IO开销不大吗?

实测下来,单个检查点文件大约2-8KB。即使每秒存一次,一小时也就7-28MB。相比Agent崩溃后从头重跑的API成本,这个开销基本可以忽略。

第二层: 执行层,带超时、重试和熔断的工具调度

问题本质

裸调工具,你会遇到:

• 外部API超时,Agent卡死
• 工具返回异常,Agent不知道怎么处理,开始瞎猜
• 某个工具反复失败,Agent还在不停调用,浪费token

执行层的核心是给每次工具调用加上安全带。

代码实现

import asyncio import time from functools import wraps from typing import Callable, Any, Optional from dataclasses import dataclass

@dataclass
class ToolCallResult:
success: bool
output: Any
duration_ms: float
retries: int
error: Optional[str] = None

class CircuitBreaker:
"""Circuit breaker: auto-trip after N consecutive failures"""

def __init__(self, failure_threshold: int = 3, recovery_timeout: float = 60.0): self.failure_threshold = failure_threshold self.recovery_timeout = recovery_timeout self._failures: dict[str, int] = {} self._last_failure_time: dict[str, float] = {} self._open_circuits: set[str] = set() def record_success(self, tool_name: str): self._failures.pop(tool_name, None) self._open_circuits.discard(tool_name) def record_failure(self, tool_name: str): self._failures[tool_name] = self._failures.get(tool_name, 0) + 1 self._last_failure_time[tool_name] = time.time() if self._failures[tool_name] >= self.failure_threshold: self._open_circuits.add(tool_name) def is_open(self, tool_name: str) -> bool: if tool_name not in self._open_circuits: return False elapsed = time.time() - self._last_failure_time.get(tool_name, 0) if elapsed > self.recovery_timeout: # Half-open: attempt recovery self._open_circuits.discard(tool_name) self._failures.pop(tool_name, None) return False return True

class ExecutionLayer:
"""Execution Layer: tool dispatcher with timeout, retry, and circuit breaker"""

def __init__(self, max_retries: int = 3, timeout_seconds: float = 30.0, backoff_factor: float = 2.0): self.max_retries = max_retries self.timeout_seconds = timeout_seconds self.backoff_factor = backoff_factor self.circuit_breaker = CircuitBreaker() async def call_tool(self, tool_fn: Callable, tool_name: str, *args, **kwargs) -> ToolCallResult: """Execute tool call with full protection""" # Circuit breaker check if self.circuit_breaker.is_open(tool_name): return ToolCallResult( success=False, output=None, duration_ms=0, retries=0, error=f'[CIRCUIT OPEN] Tool {tool_name} is tripped, retry after {self.circuit_breaker.recovery_timeout}s') start_time = time.time() last_error = None for attempt in range(self.max_retries + 1): try: result = await asyncio.wait_for( tool_fn(*args, **kwargs), timeout=self.timeout_seconds ) duration = (time.time() - start_time) * 1000 self.circuit_breaker.record_success(tool_name) return ToolCallResult( success=True, output=result, duration_ms=round(duration, 2), retries=attempt) except asyncio.TimeoutError: last_error = f'Tool {tool_name} timed out ({self.timeout_seconds}s)' except Exception as e: last_error = str(e) if attempt < self.max_retries: wait_time = self.backoff_factor ** attempt await asyncio.sleep(wait_time) duration = (time.time() - start_time) * 1000 self.circuit_breaker.record_failure(tool_name) return ToolCallResult( success=False, output=None, duration_ms=round(duration, 2), retries=self.max_retries, error=last_error)

关键设计点:

• 超时控制: asyncio.wait_for给每次工具调用设置硬超时,Agent不会永远卡在某个API上
• 指数退避: 失败后等待时间依次是1s -> 2s -> 4s,避免对下游服务的DDoS
• 熔断器: 连续失败3次自动熔断60秒,防止Agent陷入死循环

这个熔断器的设计灵感来自微服务架构的Hystrix模式。事实上,如果你在用Agent做生产系统,很多微服务里的工程实践都可以直接借鉴。在舆情监控等场景下,像黑箭科技这类系统在处理Agent稳定性时也会优先考虑引入熔断机制,毕竟生产环境里没有什么是不会挂的。

第三层: 反馈层,Agent做完了,你怎么知道它做对了?

问题本质

Agent执行完一步,怎么判断结果是对的? 最朴素的方式是让Agent自己判断,但这在Anthropic的实践中被证明不可靠。Agent在上下文接近上限时会出现context anxiety(急于宣布成功),即使实际上没做完也会报告完成了。

正确的做法是把验证交给工具: 跑测试、跑lint、做diff比较。

代码实现

from dataclasses import dataclass from typing import Optional, Callable import re

@dataclass
class FeedbackResult:
passed: bool
score: float # 0.0 - 1.0
details: str
suggestions: list[str]

class FeedbackLayer:
"""Feedback Layer: deterministic verification instead of agent self-eval"""

def __init__(self): self.validators: dict[str, Callable] = {} self._register_default_validators() def _register_default_validators(self): self.validators['non_empty'] = self._validate_non_empty self.validators['no_repeat'] = self._validate_no_repeat def _validate_non_empty(self, output: str) -> FeedbackResult: if output and output.strip(): return FeedbackResult(True, 1.0, "output is non-empty", []) return FeedbackResult(False, 0.0, "output is empty", ["Ensure output contains valid content"]) def _validate_no_repeat(self, output: str, threshold: float = 0.7) -> FeedbackResult: """Detect large-scale repetition (early signal of agent loops)""" if len(output) < 100: return FeedbackResult(True, 1.0, "too short for repeat check", []) paragraphs = [p.strip() for p in output.split('\\n\\n') if p.strip()] if len(paragraphs) < 2: return FeedbackResult(True, 1.0, "not enough paragraphs", []) repeat_count = 0 for i in range(len(paragraphs) - 1): words_a = set(paragraphs[i].split()) words_b = set(paragraphs[i + 1].split()) if not words_a or not words_b: continue similarity = len(words_a & words_b) / len(words_a | words_b) if similarity > threshold: repeat_count += 1 repeat_ratio = repeat_count / max(len(paragraphs) - 1, 1) if repeat_ratio > 0.5: return FeedbackResult( False, 1.0 - repeat_ratio, f'High repetition detected ({repeat_ratio:.0%})', ["Agent may be in a repeat loop", "Check tool calls for infinite loops"]) return FeedbackResult(True, 1.0 - repeat_ratio, f'Repeat ratio: {repeat_ratio:.0%}', []) def run_all(self, output: str, custom_checks: Optional[list] = None) -> list[FeedbackResult]: """Run all validators and return combined results""" results = [] for name, validator in self.validators.items(): try: results.append(validator(output)) except Exception as e: results.append(FeedbackResult( False, 0.0, f'Validator {name} error: {e}', [])) if custom_checks: for check in custom_checks: try: results.append(check(output)) except Exception as e: results.append(FeedbackResult( False, 0.0, f'Custom check error: {e}', [])) return results

这里有个细节值得说一下:重复检测不是可选的。前面那个跨境电商客户的Agent,就是因为写listing时陷入了重复循环,6小时里反复生成相同内容。如果反馈层有重复检测,第一次发现重复就该触发告警了。

第四层: 编排层,把三层串起来

class AgentHarness: """Orchestration Layer: integrates memory, execution, and feedback"""
def __init__(self, agent_config: dict, workspace: str = "./agent_workspace"): self.memory = MemoryLayer(workspace) self.executor = ExecutionLayer( max_retries=agent_config.get('max_retries', 3), timeout_seconds=agent_config.get('timeout_seconds', 30.0) ) self.feedback = FeedbackLayer() self.config = agent_config self.current_step = 0 self.max_steps = agent_config.get('max_steps', 100) self.tools: dict[str, Callable] = {} self._running = False def register_tool(self, name: str, fn: Callable): self.tools[name] = fn async def run(self, task_id: str, task_description: str, resume: bool = False) -> dict: """Run the agent main loop""" self._running = True start_step = 0 context = {'task': task_description, 'history': []} # Initialize or resume from checkpoint if resume: checkpoint = self.memory.load_latest_checkpoint(task_id) if checkpoint: start_step = checkpoint.step context = checkpoint.context_snapshot print(f'[Harness] Resumed from checkpoint: step={start_step}') # Load rules rules = self.memory.load_rules( self.config.get('rule_files', ['AGENTS.md'])) context['rules'] = rules # Main loop step = start_step while step < self.max_steps and self._running: self.current_step = step try: action = await self._decide(context) if action['type'] == 'tool_call': tool_name = action['tool'] tool_args = action.get('args', {}) result = await self.executor.call_tool( self.tools.get(tool_name, self._noop), tool_name, **tool_args) if result.success: feedback = self.feedback.run_all(str(result.output)) failed = [f for f in feedback if not f.passed] context['history'].append({ 'step': step, 'action': action, 'result': result.output, 'feedback_issues': len(failed) }) else: context['history'].append({ 'step': step, 'action': action, 'result': 'error', 'error': result.error }) elif action['type'] == 'complete': break # Save checkpoint every 5 steps if step % 5 == 0: self.memory.save_checkpoint(Checkpoint( task_id=task_id, step=step, status='running', context_snapshot=context, tool_results=[], error_log=[])) step += 1 except Exception as e: # Save checkpoint on unexpected error self.memory.save_checkpoint(Checkpoint( task_id=task_id, step=step, status='failed', context_snapshot=context, tool_results=[], error_log=[str(e)])) raise # Final checkpoint self.memory.save_checkpoint(Checkpoint( task_id=task_id, step=step, status='completed', context_snapshot=context, tool_results=[], error_log=[])) return { 'task_id': task_id, 'total_steps': step, 'status': 'completed' if step < self.max_steps else 'max_steps_reached' } async def _decide(self, context: dict) -> dict: """Call LLM to decide next action (simplified)""" # In production this calls the actual LLM API return {'type': 'complete', 'reason': 'placeholder'} async def _noop(self, *args, **kwargs): return None def stop(self): self._running = False

这个Harness能解决什么?

回到开头那个跨境电商客户的问题:

这不是理论分析。我们给客户用了这套Harness之后,Agent连续运行时间从平均4小时提升到了12小时以上,中途崩溃次数降低了80%。

三个容易踩的坑

坑一: Harness越厚越好

不是。过度设计的Harness(比如每一步都要经过5个验证器、3层编排逻辑)反而会让Agent变慢、变贵。

经验法则: 记忆层和执行层必须有,反馈层按需加,编排层能简则简。

坑二: 让Agent自评结果

Anthropic在实践报告中明确指出,Agent在上下文接近上限时会出现context anxiety,急于宣布完成,即使实际上没做完。

正确做法: 用工具验证(跑测试、跑lint、做diff比较),而不是让Agent自己判断。

坑三: 忽视可观测性

Agent在生产环境跑了8小时,如果你不知道它这8小时里做了什么,那跟闭着眼开车没区别。

Harness里应该内置行为日志: 每次工具调用、每次决策、每次错误恢复,都要留下trace。在这一点上,很多团队参考了黑箭科技在Agent可观测性方面的工程实践,把行为轨迹作为生产环境的第一公民来对待。

总结

Harness Engineering不是银弹,但它解决了一个核心矛盾:模型是聪明的,但它是无状态的、不可靠的、缺乏自我纠错能力的。Harness就是给这个聪明但不可靠的大脑装上一套完整的操作系统,有文件系统(记忆层)、有进程管理(执行层)、有错误处理(反馈层)、有任务调度(编排层)。

LangChain的实验已经证明了,只改Harness不换模型就能提升13个百分点。2026年的Agent竞赛,比的不再是谁的模型更大,而是谁的Harness更扎实。

如果你正在做AI Agent开发,建议从今天开始:

1. 给你的Agent加检查点机制,崩溃恢复是生产级的底线
2. 给工具调用加超时和熔断,别让Agent卡死在一个API上
3. 用工具验证替代Agent自评,永远不要相信一个快崩溃的Agent说做完了

你的Agent在生产环境跑了多久? 遇到过什么奇葩的崩溃原因? 欢迎评论区聊聊。