企业官网建设流程全解析

• 人工智能
• AI Agent
• 多智能体
• MCP 服务
• 工具调用
• 浏览器控制

A practical guide to taking AI agents from demo to production

让一个 AI Agent 在演示环境跑通很容易，让它在上生产后依然可靠却很难。本文围绕"Demo 与生产的差距"这条主线，系统梳理了从可靠性、可扩展性、可观测性、安全、成本控制到人工监督（HITL）的完整生产就绪清单，给出可直接复制的重试、熔断、结构化日志、指标采集、分布式追踪、输入校验、审计日志等实现代码，并以 Hive（当前仓库gh_mirrors/hive48/hive，项目描述为 "Multi-Agent Harness for Production AI"）为参照物，逐一对照其 可观测性日志实现、事件总线、上游健康探测、社交速率限制器 等源码，让你既掌握通用工程方法，也能理解一个生产级多 Agent 平台是如何把这些方法落地的。

Demo vs Production：差距到底在哪里

这张表是整篇文章的基调：Demo 阶段的所有"无所谓"，在 production 阶段都会变成事故。接下来的每一节，都是在逐项填平这些差距。

生产就绪检查清单（Production Readiness Checklist）

1. 可靠性（Reliability）

• 带指数退避的重试逻辑（Retry logic with exponential backoff）
• 面向故障服务的熔断器（Circuit breakers for failing services）
• 优雅降级 / 回退（Graceful degradation / fallbacks）
• 健康检查端点（Health check endpoints）
• 崩溃后自动恢复（Automatic recovery from crashes）

2. 可扩展性（Scalability）

• 水平扩展能力（Horizontal scaling capability）
• 无状态设计（或托管状态）（Stateless design or managed state）
• 基于队列处理突发流量（Queue-based processing for bursts）
• 数据库连接池（Database connection pooling）
• 缓存层（Caching layer）

3. 可观测性（Observability）

• 结构化日志（Structured logging）
• 指标采集（Metrics collection）
• 分布式追踪（Distributed tracing）
• 告警规则（Alerting rules）
• 监控仪表盘（Dashboard for monitoring）

4. 安全（Security）

• API 认证（API authentication）
• 输入校验（Input validation）
• 输出清洗（Output sanitization）
• 密钥管理（Secrets management）
• 审计日志（Audit logging）

5. 成本控制（Cost Control）

• 预算上限（Budget limits）
• 用量追踪（Usage tracking）
• 模型降级策略（Model degradation policies）
• 异常检测（Anomaly detection）

6. 人工监督（Human Oversight）

• HITL 检查点（HITL checkpoints）
• 升级策略（Escalation policies）
• 审计追踪（Audit trails）
• 手动覆盖能力（Manual override capability）

这份清单就是整篇文章的"验收标准"。把它贴在你的项目 README 里，每完成一项就打勾，是评估 Agent 系统生产就绪度最直接的方法。

架构模式：从单体服务到全平台

模式 1：简单 Agent 服务（Simple Agent Service）

┌──────────────────────────────────────────┐ │ Agent Service │ │ ┌────────────────────────────────────┐ │ │ │ Request Handler │ │ │ │ ┌──────┐ ┌──────┐ ┌──────┐ │ │ │ │ │Validate│→│Agent │→│Format │ │ │ │ │ │ Input │ │Execute│ │Output│ │ │ │ │ └──────┘ └──────┘ └──────┘ │ │ │ └────────────────────────────────────┘ │ │ │ │ │ ┌─────────────────────────────────────┐│ │ │ Dependencies ││ │ │ • LLM API • Tools • Database ││ │ └─────────────────────────────────────┘│ └──────────────────────────────────────────┘

适用场景：简单用例、低流量。请求进、结果出，一个进程搞定，是四种模式里最容易上手的一种。

模式 2：队列化处理（Queue-Based Processing）

┌───────┐ ┌───────┐ ┌───────────────┐ │Request│───▶│ Queue │───▶│ Agent Workers │ │ API │ │ │ │ (N copies) │ └───────┘ └───────┘ └───────────────┘ │ ▼ ┌─────────┐ │ Results │ │ DB │ └─────────┘

适用场景：高流量、异步处理。请求先入队，N 个 Worker 副本消费，天然支持水平扩展，也是应对突发流量（burst）的标准手段——检查清单里"基于队列处理突发流量"指的就是它。

模式 3：事件驱动 Agent（Event-Driven Agents）

┌─────────────┐ │ Event Source│─────┐ └─────────────┘ │ ▼ ┌─────────────┐ ┌─────────┐ ┌─────────────┐ │ Event Source│─▶│ Event │─▶│ Agent │ └─────────────┘ │ Bus │ │ Processors │ └─────────┘ └─────────────┘ ┌─────────────┐ │ │ Event Source│─────┘ └─────────────┘

适用场景：响应式系统、集成场景。多个事件源把事件发布到总线，Agent 处理器订阅并响应。Hive 对这一模式的实现值得单独对照：其 事件总线模块 是一个基于 asyncio 的发布/订阅系统，EventBus支持按事件类型（如EXECUTION_STARTED、EXECUTION_COMPLETED、EXECUTION_FAILED、STREAM_STALLED等几十种EventType）订阅，还支持按filter_stream、filter_node、filter_execution、filter_colony做过滤；每个事件携带correlation_id用于关联相关事件，seq单调递增计数用于前端去重；发布时通过asyncio.Semaphore(max_concurrent_handlers)做并发限制，并对每个 handler 施加 15 秒硬超时（_HANDLER_TIMEOUT_SECONDS），防止某个订阅者死锁把整个发布者拖垮——这正是一个生产级事件总线在"优雅处理错误"上的典型设计。

模式 4：全平台（Full Platform，Hive）

┌────────────────────────────────────────────────────────┐ │ Hive Platform │ │ │ │ ┌──────────────┐ ┌──────────────┐ ┌─────────────┐ │ │ │ Coding Agent │ │Worker Agents │ │ Dashboard │ │ │ │ (Generate) │ │ (Execute) │ │ (Monitor) │ │ │ └──────────────┘ └──────────────┘ └─────────────┘ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌────────────────────────────────────────────────┐ │ │ │ Control Plane │ │ │ │ • Budget • Policies • Metrics • HITL │ │ │ └────────────────────────────────────────────────┘ │ │ │ │ │ ┌────────────────────────────────────────────────┐ │ │ │ Storage Layer │ │ │ │ • Events • Policies • Config │ │ │ └────────────────────────────────────────────────┘ │ └────────────────────────────────────────────────────────┘

适用场景：复杂系统、自我改进型 Agent。Coding Agent 负责生成，Worker Agents 负责执行，Dashboard 负责监控，Control Plane 集中管理预算、策略、指标和 HITL，Storage Layer 持久化事件、策略与配置。Hive 正是按这一结构组织的：仓库中有 orchestrator 编排器（Control Plane 的调度核心）、host 主机层（Worker 执行与事件日志）、queen 皇后 Agent（可自我改进的顶层 Agent）以及前端 dashboard 等模块。

实现可靠性（Implementing Reliability）

重试逻辑（Retry Logic）：指数退避 + 抖动

原文档给出了通用实现：

import time from functools import wraps def retry_with_backoff(max_retries=3, base_delay=1, max_delay=60): def decorator(func): @wraps(func) async def wrapper(*args, **kwargs): retries = 0 while True: try: return await func(*args, **kwargs) except (RateLimitError, TimeoutError) as e: retries += 1 if retries > max_retries: raise delay = min(base_delay * (2 ** retries), max_delay) logger.warning(f"Retry {retries}/{max_retries} after {delay}s: {e}") await asyncio.sleep(delay) return wrapper return decorator @retry_with_backoff(max_retries=3) async def call_llm(prompt): return await llm_client.complete(prompt)

源码级深化：生产级重试远不止"退避"这么简单。Hive 的 LLM 客户端 litellm.py 给出了一个更完整的参考实现，值得逐条吸收：

1. 优先尊重服务端给出的退避时间：_parse_retry_after()按优先级解析retry-after-ms（毫秒）、retry-after（秒）、retry-after（RFC 7231 HTTP-date）三种响应头，并统一min(..., max_delay)封顶，避免上游明说"等 3 秒"时客户端却按自己的指数序列傻等。
2. 指数退避带封顶与正抖动：_compute_retry_delay()在无头部可用时使用backoff_base * (2**attempt)，并加上正抖动（positive jitter）——文档注释明确说明：加入抖动是为了让 N 个并发重试的客户端不会在同一时刻一起醒来（惊群效应），这正是生产环境中重试逻辑最常见的隐形坑。
3. 区分"可重试"与"不可重试"错误：源码注释明确把计费/额度类错误（如 0 余额账号）归类为永久性错误——重试一个 0 余额的账号只是浪费钱；空流重试则使用短固定延迟而非限流退避，因为那是流级别的瞬时问题。
4. 超时预算意识：源码中还有一处值得注意的注释——避免在指数退避上"烧掉 12+ 分钟"才报错，说明生产环境要给整体重试设一个时间预算，而不是无限退避。

熔断器（Circuit Breaker）

class CircuitBreaker: def __init__(self, failure_threshold=5, recovery_time=60): self.failure_count = 0 self.failure_threshold = failure_threshold self.recovery_time = recovery_time self.last_failure_time = None self.state = "closed" # closed, open, half-open async def call(self, func, *args, **kwargs): if self.state == "open": if time.time() - self.last_failure_time > self.recovery_time: self.state = "half-open" else: raise CircuitOpenError("Circuit breaker is open") try: result = await func(*args, **kwargs) if self.state == "half-open": self.state = "closed" self.failure_count = 0 return result except Exception as e: self.failure_count += 1 self.last_failure_time = time.time() if self.failure_count >= self.failure_threshold: self.state = "open" raise

熔断器是"三态机"：closed（正常放行）→ 连续失败达到阈值后open（直接拒绝，快速失败）→ 经过恢复时间进入half-open（放一个探针请求试水），成功则复位为closed，失败则回到open。它的核心价值是避免对已经挂掉的服务继续发起注定失败的调用，把资源留给健康服务。

源码级深化：Hive 在"上游健康"上的实践。熔断解决的是"该不该继续调"，而 runtime_health.py 解决的是"要不要让用户知道上游挂了"——它是进程级的上游网络健康标志位。该模块维护一个线程安全的全局状态（degraded/reason/since_epoch/last_event_epoch），核心逻辑如下：

• is_upstream_network_error()用一组刻意收窄的字符串标记（如name resolution、connection refused、ssl handshake、connect timeout、ConnectionError）判断异常是否属于"上游 LLM 不可达"类——注释强调要收窄范围，避免把认证、schema、支付这类真实应用错误也点亮告警横幅；
• Agent 循环在遇到网络类异常时调用mark_upstream_degraded(reason)，下一次流成功时调用mark_upstream_healthy()复位；
• /api/sessions/live的 SSE 推送会在每个 tick 读取该状态并携带给前端，于是用户 WiFi 断掉约 3 秒内，界面的连通性横幅就会从绿变琥珀。

这告诉我们：重试逻辑和熔断/降级不是孤立的，它们必须和面向用户的可见信号打通——用户需要知道"系统在重试，而不是卡死了"。

优雅降级（Graceful Degradation）

async def process_with_fallback(task): try: # Try primary approach return await primary_agent.execute(task) except AgentError: try: # Fall back to simpler approach return await fallback_agent.execute(task) except AgentError: # Last resort: static response return create_static_response(task)

优雅降级的核心是降级阶梯：先尝试最强的主路径，失败则退到更简单的备用路径，最后兜底返回静态响应——保证"服务可用但可能不那么聪明"，而不是"服务不可用"。

源码级深化：Hive 的"降级"无处不在。举两个代表性例子：

• LLM 调用链：当主模型（可能昂贵或不可用）失败时，Hive 的编排层可降级到备用模型/更小预算的调用。例如 orchestrator.py 中为摘要生成设置summary_budget = max(1024, max_tokens // 2)，即对非核心任务使用减半的 token 预算——这是"成本维度"上的降级。
• 事件日志自愈：event_log.py 中的EventLogFile实现了"写失败自动重开句柄"的降级：一次写失败不会像旧实现那样留下关闭的句柄导致整个会话后续事件全部静默丢失（源码记录了 2026-07-02 01:04:30 的真实事故），而是自动_reopen()重开同一路径的文件继续追加；若重开也失败，则置空句柄让后续写入干净短路，并做限频 WARN 提示。降级的目标是"永远不让日志故障拖垮事件投递"。

实现可观测性（Implementing Observability）

结构化日志（Structured Logging）

import structlog logger = structlog.get_logger() async def execute_agent(task): logger.info("agent_execution_started", task_id=task.id, agent_id=agent.id, input_tokens=count_tokens(task.input)) try: result = await agent.run(task) logger.info("agent_execution_completed", task_id=task.id, duration_ms=duration, output_tokens=count_tokens(result), cost_usd=calculate_cost(result)) return result except Exception as e: logger.error("agent_execution_failed", task_id=task.id, error=str(e), error_type=type(e).__name__) raise

结构化日志的关键词是"事件 + 字段"：每条日志是一个命名的event（如agent_execution_started），附带键值对字段（task_id、agent_id、duration_ms、cost_usd），而不是一段人肉拼接的字符串。这样才能被日志系统按字段索引、过滤和聚合。

源码级深化：Hive 的零摩擦结构化日志。logging.py 给出了一个更工程化的设计，其核心思想是**"开发者零负担"**：

• ContextVar 自动传播：用ContextVar保存 trace 上下文（线程安全且 async 安全），框架在关键点自动注入——Runtime.start_run()生成trace_id/execution_id/goal_id，GraphExecutor.execute()追加agent_id，Node.execute()追加node_id。之后用户代码里任何一行普通的logger.info("message")都会自动携带全部上下文，无需手动传参。这正是文档中"参数化日志"的进阶版：不是靠开发者自觉写字段，而是框架自动带上。
• 双输出模式：configure_logging()支持json（生产，机器可解析）与human（开发，带颜色）两种 formatter，auto模式则根据LOG_FORMAT=json或ENV=production自动选择。JSON 模式还会设置NO_COLOR/FORCE_COLOR并让 LiteLLM、httpcore、openai 等第三方库的日志也走 JSON formatter。
• stdout/stderr 分流：DEBUG/INFO走 stdout，WARNING+走 stderr，避免常规日志被打上[runtime stderr]标签被误判为错误——这是连日志管道语义都考虑到的生产细节。

指标采集（Metrics Collection）

from prometheus_client import Counter, Histogram, Gauge # Counters agent_requests_total = Counter( 'agent_requests_total', 'Total agent requests', ['agent_id', 'status'] ) # Histograms agent_duration_seconds = Histogram( 'agent_duration_seconds', 'Agent execution duration', ['agent_id'] ) # Gauges agent_active_tasks = Gauge( 'agent_active_tasks', 'Currently running agent tasks', ['agent_id'] ) async def execute_with_metrics(agent, task): agent_active_tasks.labels(agent_id=agent.id).inc() start = time.time() try: result = await agent.run(task) agent_requests_total.labels(agent_id=agent.id, status='success').inc() return result except Exception: agent_requests_total.labels(agent_id=agent.id, status='error').inc() raise finally: duration = time.time() - start agent_duration_seconds.labels(agent_id=agent.id).observe(duration) agent_active_tasks.labels(agent_id=agent.id).dec()

指标类型选择的三条经验：Counter只增不减（请求总数、错误数），Histogram记录分布（耗时、token 数），Gauge可增可减（当前并发任务数）。注意finally块保证了耗时记录与并发数递减无论成败都会执行——这是最容易写漏的地方。

源码级深化：Hive 的"用量即指标"。Hive 把"成本"也做成了一等指标：在 tracker/llm_debug_logger.py 与 tracker/runtime_log_store.py 中，每次 LLM 调用的 token 用量、耗时、模型都被结构化记录；结构化日志的 JSON 输出里也专门预留了latency_ms、tokens_used、model字段（见 logging.py）。也就是说，"成本控制"与"可观测性"在实现层面是打通的——用量数据本身既是监控指标，又是计费与预算的依据。

分布式追踪（Distributed Tracing）

from opentelemetry import trace tracer = trace.get_tracer(__name__) async def execute_with_tracing(agent, task): with tracer.start_as_current_span("agent_execution") as span: span.set_attribute("agent.id", agent.id) span.set_attribute("task.id", task.id) # LLM call with tracer.start_as_current_span("llm_call") as llm_span: llm_span.set_attribute("model", agent.model) result = await call_llm(task.prompt) llm_span.set_attribute("tokens", result.usage.total_tokens) # Tool execution with tracer.start_as_current_span("tool_execution") as tool_span: tool_span.set_attribute("tool", tool.name) tool_result = await execute_tool(tool, result) return tool_result

分布式追踪的核心是span 树：一次 Agent 执行是一个根 span，其下的 LLM 调用、工具执行各自是子 span，层层嵌套形成调用链。当 Agent 系统拆成多个服务/进程后，靠trace_id把一次业务请求的所有日志、指标、调用串成一条链路。

源码级深化：Hive 的 trace 关联实现。logging.py 中set_trace_context()生成的trace_id是 32 位十六进制（W3C Trace Context 兼容格式），execution_id与之对齐以便关联。整个执行链路里所有日志自动携带trace_id/execution_id/agent_id/node_id，无需开发者手动把 trace id 传进每个函数——这是"零摩擦追踪"：框架层把 trace 上下文注入 ContextVar，用户代码零感知。而 event_bus.py 中的AgentEvent.correlation_id则负责跨事件关联：一次用户输入从"收到"（CLIENT_INPUT_RECEIVED）到"真正织入对话"（CLIENT_INPUT_COMMITTED）的事件，靠同一个correlation_id串起来。

安全最佳实践（Security Best Practices）

输入校验（Input Validation）

from pydantic import BaseModel, validator class AgentRequest(BaseModel): task: str context: dict = {} max_tokens: int = 1000 @validator('task') def validate_task(cls, v): if len(v) > 10000: raise ValueError('Task too long') if contains_injection_attempt(v): raise ValueError('Invalid input detected') return v @validator('max_tokens') def validate_max_tokens(cls, v): if v > 4000: raise ValueError('max_tokens too high') return v

输入校验要做两件事：长度/范围校验（防资源耗尽）和注入检测（防提示注入 prompt injection）。Agent 系统的输入校验比传统 API 更关键，因为 LLM 对恶意输入的"执行"能力远超普通字符串处理。

源码级深化：Hive 的工具输入强制转换。Hive 在工具调用入口做了类似的防御：agent_loop/internals/tool_input_coercer.py 负责把 LLM 返回的工具参数强制转换/清洗成声明类型，避免 LLM 幻觉产生的畸形参数直接进入工具执行层；此外 skills/tool_gating.py 提供工具白名单/门控能力，从"能调哪些工具"的层面做约束——这属于输入校验的更高层防线：不仅要校验参数合法，还要校验"这个 Agent 有没有权限调这个工具"。

输出清洗（Output Sanitization）

Note:以下代码片段仅用于示意，展示输出清洗的简化示例。实际实现可能有所不同。

def sanitize_output(result): # Remove any leaked secrets result = mask_patterns(result, SECRET_PATTERNS) # Validate structure if not is_valid_response(result): raise OutputValidationError("Invalid response structure") # Check for harmful content if contains_harmful_content(result): raise ContentPolicyError("Response violates content policy") return result

输出清洗要防三类问题：密钥泄露（模型可能复述训练数据里的密钥或系统提示中的 secret）、结构非法（下游无法解析）、有害内容（违反内容政策）。

源码级深化：Hive 的"敏感信息不出库"设计。一个可对照的实践来自事件总线：审计/日志场景中，Hive 明确不记录完整敏感数据——事件日志对诊断型重字段（如每次 context 用量 tick 携带的约 280KB 完整 LLM promptfull_request）做磁盘裁剪（_DISK_STRIPPED_DATA_FIELDS，见 event_bus.py），避免日志膨胀到数十 MB；凭据收集走专门的安全表单流程，用户密钥直接 POST 到存储层，"never flow back through this event or the conversation"（见CLIENT_CREDENTIAL_FORM_REQUESTED的注释）。这与审计日志中"不要记录完整输入"（用哈希代替）的原则一脉相承。

审计日志（Audit Logging）

async def audit_log(event): log_entry = { "timestamp": datetime.utcnow().isoformat(), "event_type": event.type, "agent_id": event.agent_id, "user_id": event.user_id, "action": event.action, "input_hash": hash_content(event.input), # Don't log full input "output_hash": hash_content(event.output), "metadata": event.metadata } await audit_db.insert(log_entry)

审计日志的要点：记录"谁在什么时间对什么做了什么"，同时用哈希代替原始内容——既满足审计追溯（可以验证内容未被篡改、可以比对是否一致），又不泄露敏感数据。审计日志应追加不可变（append-only），防止事后篡改。

源码级深化：Hive 的 append-only 事件日志。event_log.py 的EventLogFile正是 append-only JSONL 写入器：每次写入line + "\n"并立即flush()；写失败自动重开句柄恢复，且"绝不抛出异常"（write()的方法签名注释就是Never raises）。事件日志由 event_bus.py 的EventBus.set_session_log()启用，事件按会话持久化为events.jsonl，可跨重启重放以重建前端状态——审计追踪在这里不只是"合规要求"，更是状态恢复与故障排查的基础设施。

部署策略（Deployment Strategies）

蓝绿部署（Blue-Green Deployment）

Load Balancer │ ┌───────────┴───────────┐ │ │ ┌─────▼─────┐ ┌─────▼─────┐ │ Blue │ │ Green │ │ (Current) │ │ (New) │ └───────────┘ └───────────┘ 1. Deploy new version to Green 2. Test Green environment 3. Switch traffic Blue → Green 4. Keep Blue for rollback

蓝绿部署的要点是始终保留一套可直接回滚的旧版本：新版本先在 Green 环境完整验证，然后负载均衡器一次性切换流量，出问题时一键切回 Blue。代价是需要双倍资源。

金丝雀部署（Canary Deployment）

Load Balancer │ ┌───────────┴───────────┐ │ 95% 5% │ ┌─────▼─────┐ ┌─────▼─────┐ │ Stable │ │ Canary │ │ (v1.0) │ │ (v1.1) │ └───────────┘ └───────────┘ 1. Deploy new version as Canary 2. Route 5% traffic to Canary 3. Monitor metrics 4. Gradually increase or rollback

金丝雀部署的要点是小流量灰度 + 指标驱动决策：先把新版本放到 5% 流量上观察指标，确认无误后逐步放大比例，异常则立即回滚。对 LLM 应用尤其合适——新模型的输出质量、延迟、成本都可以先在真实流量的 5% 上验证。

特性开关（Feature Flags）

async def execute_agent(task, user): if feature_flags.is_enabled("new_agent_v2", user.id): return await agent_v2.execute(task) else: return await agent_v1.execute(task)

特性开关的价值是把"发布"和"上线"解耦：代码可以先部署（deploy），功能可以随时开/关（release），甚至能做到按用户维度定向灰度——同一个服务实例同时运行新旧两套逻辑，用开关切换。

源码级深化：Hive 的能力开关。Hive 在 llm/capabilities.py 和 config.py 中实现了能力/配置开关体系（如按模型能力、按环境变量ENV=production切换行为），并在 server/routes_config.py 中暴露了配置读取接口——这类开关体系正是"特性开关"在生产配置管理层面的落地：把可变的策略参数化、可配置化，避免每次调整都重新发版。

框架对比：生产就绪度（Framework Comparison）

对照仓库实证，这张表格里 Hive 的"内置"并非空口：

• 重试逻辑内置：litellm.py 的_compute_retry_delay()/_parse_retry_after()；
• 健康检查内置：runtime_health.py 的上游健康探测 + runtime/health 相关模块；
• 监控内置：observability/logging.py 结构化日志 + tracker/ 的 LLM debug 日志与运行时日志存储；
• 成本控制内置：rate_limiter.py 社交平台速率限制 + orchestrator 中的 token 预算控制；
• HITL 原生：事件类型CLIENT_INPUT_REQUESTED/CLIENT_INPUT_RECEIVED/ESCALATION_REQUESTED（event_bus.py）以及 sentinel/ 升级通知体系；
• 自我改进原生：agents/queen/ 皇后 Agent（含 playbooks、memory）、skills/ 技能系统（含 skill_writer.py 让 Agent 自己写技能）；
• 仪表盘内置：前端 dashboard 页面 + server/routes_events.py 事件流接口。

这张表的正确用法不是"选谁不选谁"，而是成本核算：选择 DIY 或轻量框架，意味着表格里所有"Build"格子的工程量都由你承担；选择像 Hive 这样面向生产的平台，则意味着这些能力开箱即用，但你需要接受它的架构约定。

面向生产的测试（Testing for Production）

单元测试（Unit Tests）

def test_agent_handles_rate_limit(): with mock.patch('llm.complete', side_effect=RateLimitError()): result = agent.execute(task) assert result.status == "retried" def test_agent_validates_input(): with pytest.raises(ValidationError): agent.execute({"task": "x" * 100000}) # Too long

单元测试的核心手法是mock 外部依赖：把 LLM 客户端替换成预设抛错的桩，验证重试/校验逻辑本身。注意第一类测试断言的是"状态"，不是"最终成功"——限流重试后应处于retried状态。

仓库实证：Hive 的重试/限流测试覆盖。Hive 的测试套件（core/tests/）中包括test_litellm_provider.py、test_litellm_streaming.py（验证 LLM 客户端与流式行为）、test_social_rate_limiter.py（验证速率限制器）、test_stream_stall.py（验证流停滞检测与恢复）。在 tools 侧还有 tools/tests/test_terminal_tools_* 一系列安全/健壮性测试。这印证了"生产级重试与限流必须有测试兜底"的原则。

集成测试（Integration Tests）

async def test_full_agent_flow(): # Create test task task = create_test_task() # Execute agent result = await agent.execute(task) # Verify result assert result.success assert result.output is not None # Verify monitoring assert metrics.request_count > 0 assert metrics.last_cost < 1.0

集成测试验证完整链路：任务创建 → 执行 → 结果校验，并额外断言"监控系统确实在记录"（metrics.request_count > 0、metrics.last_cost < 1.0）——如果可观测性没生效，集成测试应该失败。这条"测试监控本身"的思想非常值得推广。

负载测试（Load Tests）

async def load_test_agent(): tasks = [create_test_task() for _ in range(100)] start = time.time() results = await asyncio.gather(*[ agent.execute(task) for task in tasks ]) duration = time.time() - start success_rate = sum(1 for r in results if r.success) / len(results) avg_latency = duration / len(tasks) assert success_rate > 0.95 assert avg_latency < 5.0 # seconds

负载测试同时关注两个维度：成功率（> 95%）与平均延迟（< 5 秒）。注意这里的并发方式是用asyncio.gather在单进程内模拟并发——真正的高并发场景还应配合多进程/多实例验证水平扩展。

混沌测试（Chaos Tests）

async def test_agent_survives_llm_outage(): with mock.patch('llm.complete', side_effect=ConnectionError()): # Should use fallback or degrade gracefully result = await agent.execute(task) assert result.status in ["fallback", "degraded"] async def test_agent_survives_high_load(): # Simulate burst traffic tasks = [create_test_task() for _ in range(1000)] results = await asyncio.gather(*[ agent.execute(task) for task in tasks ], return_exceptions=True) # Should not crash, may throttle errors = [r for r in results if isinstance(r, Exception)] assert len(errors) / len(results) < 0.1 # <10% error rate

混沌测试主动注入故障验证系统韧性：LLM 完全宕机时系统应优雅降级（status in ["fallback", "degraded"]）；1000 个并发任务压测下系统"不能崩，但允许限流"，错误率必须低于 10%——注意断言的是"不崩溃 + 错误率可接受"，而不是"全部成功"，因为生产系统在过载时应当限流而非硬扛。

结论

生产级 AI Agent 需要同时具备五根支柱：

1. 可靠性（Reliability）：重试、熔断、回退——保证 LLM 和外部服务抖动时不掉链子；
2. 可观测性（Observability）：日志、指标、追踪、仪表盘——保证出问题时 30 秒内定位根因；
3. 安全（Security）：校验、清洗、审计——保证输入、输出、日志三个入口都不泄露、不被攻击；
4. 成本控制（Cost Control）：预算、追踪、降级——保证每一美元都花在刀刃上；
5. 人工监督（Human Oversight）：HITL、升级、覆盖——保证复杂决策始终有人类兜底。

像 Hive 这样的框架把上述能力大量内建（可对照 observability/logging.py、host/event_bus.py、host/runtime_health.py、rate_limiter.py、orchestrator/ 等模块验证）；而使用其他框架时，这些基础设施都需要你自己搭建。

Demo 与生产之间的鸿沟是真实且巨大的——从第一天就为它做规划，而不是上线前才补课。把本文的检查清单、代码骨架与 Hive 的源码级实现对照着消化，你就能少踩绝大多数 Agent 上生产的坑。

Last updated: January 2025

• 人工智能
• AI Agent
• 多智能体
• MCP 服务
• 工具调用
• 浏览器控制