AI Agent 从 demo 进入生产环境后，最先暴露的问题通常不是模型能力，而是工程控制面。

一个可以调用工具的 Agent，背后会经过意图识别、状态更新、工具选择、参数生成、外部 API 调用、结果合并、风险判断和最终回复。任何一步缺少可观测性，线上问题都会变成“模型偶发不稳定”；任何一步缺少审计字段，后续复盘都会变成靠聊天记录猜。

这篇文章用一个生产级 Agent 的典型架构，说明如何用 LangGraph 管住状态流，用 OpenTelemetry 追踪调用链，再用结构化审计日志补齐合规和复盘能力。

1. 生产级 Agent 先解决什么问题

很多团队做 Agent 的第一步，是把 Prompt、工具列表和模型路由接起来。这个阶段能跑 demo，但还不是生产系统。

生产环境至少要回答五个问题：

1. 本次请求经过了哪些节点？
2. Agent 为什么选择这个工具？
3. 工具入参、出参、错误码和重试次数是什么？
4. 哪一步耗时最高，P95/P99 延迟是否可控？
5. 如果用户质疑结果，能不能回放证据链？

如果这些问题答不上来，系统上线后遇到超时、错答、权限失败、数据漂移时，排障成本会非常高。

2. 用 LangGraph 把 Agent 流程显式化

LangGraph 的价值不在于“多 Agent 更酷”，而在于把状态、节点和分支显式建模。

一个最小生产流程可以拆成：

• classify_intent：判断请求类型和风险等级
• plan_tools：生成工具调用计划
• execute_tool：执行外部工具并记录结果
• verify_evidence：检查引用、权限和数据完整性
• finalize_answer：生成最终回复
• human_review：高风险或证据不足时进入人工复核

示例状态可以这样定义：

from typing import TypedDict, Optional, Any


class AgentState(TypedDict):
    request_id: str
    user_id: str
    intent: Optional[str]
    risk_level: str
    tool_name: Optional[str]
    tool_input: Optional[dict[str, Any]]
    tool_output: Optional[dict[str, Any]]
    evidence_ids: list[str]
    retry_count: int
    error_code: Optional[str]
    requires_human_review: bool
    final_answer: Optional[str]

这里的关键不是字段多，而是每个字段都能进入日志和 trace。状态一旦显式化，失败路径才有办法设计。

3. 工具调用必须有统一边界

生产 Agent 最容易坏在工具层。外部 API 会超时，权限会失败，字段会变更，返回值也可能不符合模型预期。

建议所有工具封装成统一契约：

from dataclasses import dataclass
from typing import Any, Optional


@dataclass
class ToolResult:
    ok: bool
    data: Optional[dict[str, Any]]
    error_code: Optional[str]
    latency_ms: int
    evidence_ids: list[str]
    retry_count: int


def call_tool(name: str, payload: dict[str, Any]) -> ToolResult:
    # 生产代码里应加入超时、重试、权限校验、schema 校验和错误映射
    ...

不要让每个工具随意返回自然语言。工具层应该返回结构化结果，再由 Agent 决定是否继续、重试、降级或转人工。

4. 用 OpenTelemetry 串起完整链路

Agent 系统里，单看模型调用日志不够。需要把 LLM、工具、检索、权限校验、后处理都串进同一个 trace。

典型 trace 结构可以是：

• agent.request
• agent.intent_classification
• agent.plan_tools
• tool.crm_lookup
• tool.payment_status
• agent.verify_evidence
• agent.finalize_answer

示例埋点：

from opentelemetry import trace

tracer = trace.get_tracer("agentkick.agent")


def execute_tool_node(state: AgentState) -> AgentState:
    with tracer.start_as_current_span("tool.execute") as span:
        span.set_attribute("request_id", state["request_id"])
        span.set_attribute("tool.name", state["tool_name"] or "")
        span.set_attribute("risk_level", state["risk_level"])
        span.set_attribute("retry_count", state["retry_count"])

        result = call_tool(state["tool_name"], state["tool_input"] or {})

        span.set_attribute("tool.ok", result.ok)
        span.set_attribute("tool.latency_ms", result.latency_ms)
        if result.error_code:
            span.set_attribute("tool.error_code", result.error_code)

        state["tool_output"] = result.data
        state["evidence_ids"].extend(result.evidence_ids)
        state["error_code"] = result.error_code
        return state

这样排障时可以直接看到：是模型慢、工具慢、权限失败，还是证据校验失败。

5. 审计日志不要只依赖 trace

OpenTelemetry 适合排障和性能分析，但审计日志需要更稳定的业务语义。

建议单独写一条 append-only 审计流：

{
  "request_id": "req_20260612_001",
  "user_id": "u_123",
  "node": "tool.payment_status",
  "tool_name": "payment_status",
  "input_hash": "sha256:...",
  "output_hash": "sha256:...",
  "evidence_ids": ["invoice_456", "payment_789"],
  "decision": "continue",
  "risk_level": "medium",
  "created_at": "2026-06-12T10:30:00Z"
}

注意两点：

• 不要把敏感明文全部塞进日志，优先记录摘要、哈希、引用 ID 和脱敏字段。
• 审计日志要能长期保存，并且最好不可随意覆盖。

6. 上线前检查清单

我会用这张清单判断一个 Agent 是否接近生产可用：

• 每个节点都有明确职责和输入输出 schema。
• 每个工具都有超时、重试、错误码和降级策略。
• 每次请求都有唯一 request_id 和 trace id。
• LLM 调用、工具调用、检索和后处理都进入同一条 trace。
• 高风险动作有人工复核或强约束规则。
• 最终回答能关联到证据 ID。
• 审计日志可以按请求、用户、工具、时间窗口检索。
• 线上监控覆盖延迟、失败率、无证据回答率和人工接管率。

结语

生产级 AI Agent 的核心，不是让模型多调用几个工具，而是让工具调用链、状态流、证据链和恢复链都可观测、可审计、可回滚。

LangGraph 适合把复杂流程显式化，OpenTelemetry 适合把跨组件调用串起来，审计日志则负责业务复盘和合规证明。三者一起用，Agent 才更像一个可运营的生产系统，而不是一次性 demo。