🏭 构建 Agent 实践指南 — 生产级特性

Demo vs 生产级 Agent 的差异

实现细节

关键代码：src/agents/system-prompt-cache-boundary.ts (~50 行)——用 SHA256 哈希验证 stable prefix 没有被意外修改，确保 provider 的 prefix cache 始终命中。

Review Fork 继承父 Agent 的缓存

效果：实测约 26% 的成本降低。Review fork 不需要重新构建系统提示词，直接复用父 agent 已缓存的版本。

RetryMixin 实现

class LLM(BaseModel, RetryMixin, NonNativeToolCallingMixin):
    model: str = "claude-sonnet-4-20250514"
    api_key: str | SecretStr | None
    base_url: str | None
    num_retries: int = 5           # 最大重试次数
    retry_multiplier: float = 8.0  # 退避因子
    timeout: int = 300             # 单次请求超时 (秒)
    max_message_chars: int = 30_000
    max_output_tokens: int | None

# 只对这些瞬态错误重试:
LLM_RETRY_EXCEPTIONS = (
    APIConnectionError,     # 网络错误
    RateLimitError,         # 429 限流
    ServiceUnavailableError, # 503 服务不可用
)

# 不对以下错误重试:
# - AuthenticationError (401/403): key 有问题，重试没用
# - BadRequestError (400): 请求格式错误，重试没用
# - NotFoundError (404): 模型不存在，重试没用

节点级重试策略

from langgraph.pregel import RetryPolicy

graph.add_node(
    "api_call",
    call_external_api,
    retry=RetryPolicy(
        max_attempts=3,          # 最大重试次数
        initial_interval=1.0,    # 初始间隔（秒）
        backoff_factor=2.0,      # 退避因子
        max_interval=10.0,       # 最大间隔上限
        jitter=True,             # 随机抖动防惊群
    )
)

# 命令式 failover: 执行时切换节点
graph.invoke(
    Command(goto="fallback_node"),  # 跳到备用节点
    config
)

LangGraph 的独特之处：重试策略是图节点的属性，不是全局配置。不同节点可以有不同的重试参数。

Provider Runtime 的 Failover 策略

编译时绑定 vs 运行时发现

Channel 分类

统一消息模型：所有平台的消息被抽象为统一的 Message 对象，Agent 只需要处理一种格式，Channel 插件负责翻译。

Gateway 平台矩阵

运行时发现机制：Hermes 在启动时扫描已安装的 plugin 包，自动注册可用的 Gateway。不需要修改代码，pip install hermes-telegram 就能添加 Telegram 支持。

平台差异处理矩阵

传统三支柱 → Agent 三支柱

监控基础设施

# Prometheus metrics
active_runs_gauge = Gauge("execution_manager_active_runs", "Number of active graph runs")
pool_size_gauge = Gauge("execution_manager_pool_size", "Maximum number of graph workers")
utilization_gauge = Gauge("execution_manager_utilization_ratio", ...)

Trajectory 自监控

OpenClaw 有一套独特的自监控机制——Trajectory 导出：

Trajectory 包含：

• ⏱ Timeline: 用户消息 → 思考 → 回复 → 工具调用链
• 🔧 Tools: 工具使用频率 + 每次调用详情 + 耗时
• 📊 Stats: 事件分布、类型统计
• 🪙 Tokens: 每轮 token 用量、输入输出比
• 🧠 API I/O: 完整的 LLM 输入输出（system prompt + 对话 + tool results）

这种设计让运维可以回放任何一次 Agent 决策的完整上下文——不仅是结果，还有"为什么"。

原生 OTel 支持

PydanticAI 是唯一原生集成 OpenTelemetry 的项目：

• Instrumentation — 自动 trace 每个 agent run 和 tool call
• Span 属性 — 记录 model、tokens、latency
• 供应商无关 — 可对接 Jaeger、Zipkin、Datadog 等

这是最"标准"的可观测性实现——不造轮子，直接用业界标准。

Gateway 配置系统

OpenClaw 使用 gateway 工具进行配置管理，支持热更新：

• config.get — 读取当前配置
• config.patch — 增量更新，尽可能热重载
• config.apply — 应用完整配置，必要时触发安全重启
• config.schema.lookup — 查询配置 schema，了解可用选项

关键设计：config.patch 优先热重载，只有当配置变更涉及核心架构（如端口、数据库连接）时才触发安全重启。这保证了大部分配置变更零停机。

LaunchDarkly 集成

AutoGPT 在非本地环境使用 LaunchDarkly 的 Feature Flags：

• 灰度发布新 Agent 策略
• 按用户/租户开关功能
• A/B 测试不同 prompt 模板
• 紧急 kill switch 关闭危险功能

这是 SaaS 化 Agent 平台的必备能力——你不能因为一个功能变更就让所有用户重启。

SOUL.md + AGENTS.md + USER.md 三层配置

Credential Pool + Rate Limit Guard

OpenClaw 将 credential rotation 作为 failover 的一环：

与 Hermes 不同，OpenClaw 的 credential 管理更简单——它是 failover 链的一部分，而不是独立的 Pool。

从 20 个项目提炼的密钥管理原则

Agent 单轮对话的 Token 消耗构成

按任务复杂度路由到不同模型

# 概念实现：模型路由器
class ModelRouter:
    """根据任务复杂度选择最经济的模型"""

    def route(self, task: str, tools: list, history_len: int) -> str:
        # 简单问答 → 小模型
        if not tools and history_len < 5:
            return "gpt-4o-mini"     # $0.15/1M input

        # 需要工具调用 → 中等模型
        if len(tools) <= 3:
            return "claude-sonnet"    # $3/1M input

        # 复杂推理 + 多工具 → 旗舰模型
        if len(tools) > 5 or "debug" in task:
            return "claude-opus"      # $15/1M input

        # 默认中等
        return "claude-sonnet"

# 成本对比:
# 全部用 opus:  $15 × 每次请求 = 💸💸💸
# 路由策略:     80% sonnet + 15% mini + 5% opus = 省 70%+

部署架构演进

Event-Driven Gateway

Docker Sandbox 隔离

OpenHands 的生产部署亮点是 Docker sandbox 隔离：

• 每个会话一个容器 — 完全隔离，防止 agent 之间互相干扰
• 三级安全模式 — LOW (只读) / MEDIUM (容器内) / HIGH (需审批)
• WebSocket 通信 — Agent 与 sandbox 之间的双向实时通信
• 资源限制 — 每个容器有 CPU/内存/磁盘上限

对话工厂模式：LocalConversation (进程内) vs RemoteConversation (连接 agent server)，根据工作区类型自动路由。

20 个项目的生产级特性评分

⭐⭐⭐ = 业界最佳水平 | ⭐⭐ = 可用 | ⭐ = 基础/缺失

🚨 必须实现

⚡ 强烈建议

📊 逐步添加

P0 #1: 重试 + 指数退避 (5 分钟实现)

import asyncio
import random
from typing import TypeVar, Callable

T = TypeVar('T')

# 瞬态错误 — 可以重试
TRANSIENT_ERRORS = (ConnectionError, TimeoutError)

async def llm_call_with_retry(
    fn: Callable[..., T],
    max_retries: int = 5,
    base_delay: float = 1.0,
    backoff_factor: float = 2.0,
    max_delay: float = 60.0,
) -> T:
    """带指数退避 + jitter 的 LLM 调用重试"""
    for attempt in range(max_retries + 1):
        try:
            return await fn()
        except TRANSIENT_ERRORS as e:
            if attempt == max_retries:
                raise  # 重试耗尽，抛出

            # 指数退避 + 随机 jitter (防惊群)
            delay = min(base_delay * (backoff_factor ** attempt), max_delay)
            jitter = random.uniform(0, delay * 0.1)
            await asyncio.sleep(delay + jitter)
        except Exception:
            raise  # 非瞬态错误，不重试

P0 #3: Cache Boundary (15 分钟实现)

import hashlib

class PromptCacheBoundary:
    """系统提示词缓存分割点"""

    def __init__(self):
        self._stable_hash: str | None = None
        self._stable_content: str | None = None

    def build_system_prompt(
        self,
        stable_parts: dict,   # 跨 turn 不变的部分
        volatile_parts: dict,  # 每 turn 重建的部分
    ) -> str:
        # 组装 stable prefix
        stable = "\n".join(
            f"## {k}\n{v}" for k, v in stable_parts.items()
        )

        # 验证 stable 没有意外变化
        current_hash = hashlib.sha256(stable.encode()).hexdigest()
        if self._stable_hash and current_hash != self._stable_hash:
            # Stable 部分变了！prefix cache 会失效
            # 这可能是故意的（配置更新），记录一下
            import warnings
            warnings.warn("Stable prefix changed — cache boundary reset")

        self._stable_hash = current_hash
        self._stable_content = stable

        # 组装 volatile suffix
        volatile = "\n".join(
            f"## {k}\n{v}" for k, v in volatile_parts.items()
        )

        # 关键: stable 在前, volatile 在后
        # 这样 LLM provider 的 prefix cache 才能命中
        return f"{stable}\n\n{volatile}"

# 用法:
cache_boundary = PromptCacheBoundary()

# 每 turn:
system_prompt = cache_boundary.build_system_prompt(
    stable_parts={
        "identity": agent_identity,       # 不变
        "tools": tool_definitions,        # 不变
        "skills": skill_content,          # 不变
    },
    volatile_parts={
        "memory": current_memory,         # 每 turn 可能变
        "runtime": runtime_context,       # 每 turn 变
        "conversation_summary": summary,  # 每 turn 变
    },
)

P1 #6: Credential Pool (20 分钟实现)

import time
from dataclasses import dataclass, field

@dataclass
class Credential:
    key: str
    provider: str
    rpm_limit: int = 60        # requests per minute
    tpm_limit: int = 100_000   # tokens per minute
    last_used: float = 0
    request_count: int = 0
    token_count: int = 0

class CredentialPool:
    """多 key 凭证池 + 速率追踪"""

    def __init__(self):
        self._credentials: list[Credential] = []
        self._window = 60  # 速率追踪窗口 (秒)

    def add(self, key: str, provider: str, **limits):
        self._credentials.append(
            Credential(key=key, provider=provider, **limits)
        )

    def acquire(self, provider: str | None = None) -> Credential:
        """获取一个可用的 credential"""
        now = time.time()
        candidates = self._credentials

        if provider:
            candidates = [c for c in candidates
                          if c.provider == provider]

        # 筛选未限流的 credential
        available = []
        for c in candidates:
            # 清除过期窗口的计数
            if now - c.last_used > self._window:
                c.request_count = 0
                c.token_count = 0

            # 检查是否还有额度
            if (c.request_count < c.rpm_limit * 0.8 and
                c.token_count < c.tpm_limit * 0.8):
                available.append(c)

        if not available:
            raise RuntimeError("All credentials rate-limited")

        # 选最近最少使用的 (避免热点)
        selected = min(available, key=lambda c: c.last_used)
        selected.last_used = now
        selected.request_count += 1
        return selected

    def report_tokens(self, key: str, tokens: int):
        """上报 token 使用量"""
        for c in self._credentials:
            if c.key == key:
                c.token_count += tokens
                break

# 用法:
pool = CredentialPool()
pool.add("sk-key1", "openai", rpm_limit=60)
pool.add("sk-key2", "openai", rpm_limit=60)
pool.add("sk-ant-key3", "anthropic", rpm_limit=50)

cred = pool.acquire(provider="openai")
# → 自动选择使用最少的 key

• 🏠 构建 Agent 实践指南 — 总览
• 📝 提示词工程 — Cache Boundary 的完整实现
• 🛡️ 安全机制 — 沙箱与审批
• 🧠 记忆系统 — 上下文压缩策略
• OpenClaw 深度分析 — 150+ 扩展生态
• Hermes 深度分析 — Credential Pool + 30+ 平台
• OpenHands 深度分析 — RetryMixin + Condenser
• AutoGPT 深度分析 — Prometheus + RabbitMQ + Langfuse
• LangGraph 深度分析 — RetryPolicy + Checkpoint
• 架构对比 — 20 个项目的生产级架构差异