生产环境中的AI智能体：架构模式与可靠性工程

智能体可靠性挑战

AI智能体引入了一类新的生产工程问题。与相同输入总是产生相同输出的确定性软件不同，智能体：

做出概率性决策，每次运行可能不同

执行长动作序列，早期错误会累积

使用可能失败、返回意外结果或改变行为的外部工具

可能陷入循环、死胡同或产生意外结果

可能造成现实世界副作用（发送邮件、修改数据库、调用API）

构建可靠的智能体需要重新思考若干软件工程基础。

基础智能体架构

感知-决策-行动循环

所有智能体遵循相同的基本循环：

感知当前状态（上下文、先前动作、工具结果）

决定下一步行动（LLM基于可用工具进行推理）

执行行动（调用工具，记录结果）

更新状态（将动作+结果添加到上下文）

检查完成（目标达成？达到最大步数？需要人工审核？）

循环或返回

最小足迹原则

Anthropic的智能体安全指南：采取必要的最小行动，优先选择可逆行动而非不可逆行动，不确定时升级到人工处理。

实现：设计工具时尽量可逆（移至回收站 vs. 永久删除），对高影响行动要求明确确认（发送邮件？发送前确认），实现“试运行”模式用于调试。

工具设计的可靠性

工具接口设计

设计良好的智能体工具：

清晰、具体的函数名（search_customer_records 而非 query_db）

描述性文档字符串，准确说明工具能做什么和不能做什么

带验证的类型化参数

尽可能幂等（调用两次与调用一次效果相同）

返回结构化数据（而非需要解析的自由文本）

在返回值中包含错误信息（不要仅抛出异常）

python
from pydantic import BaseModel, Field
from typing import Optional
class CustomerSearchResult(BaseModel):
    found: bool
    customer_id: Optional[str] = None
    name: Optional[str] = None
    email: Optional[str] = None
    error: Optional[str] = None
def search_customer_by_email(email: str) -> CustomerSearchResult:
    """
    通过电子邮件地址搜索客户。
    如果找到客户则返回客户详情，如果不存在则返回found=False。
    不会创建新客户或修改任何数据。
    """
    try:
        customer = db.customers.find_one({"email": email})
        if customer:
            return CustomerSearchResult(
                found=True,
                customer_id=str(customer["_id"]),
                name=customer["name"],
                email=customer["email"]
            )
        return CustomerSearchResult(found=False)
    except Exception as e:
        return CustomerSearchResult(found=False, error=str(e))

工具中的错误处理

工具会失败。为此设计：

将错误作为数据返回（而非异常），以便智能体能够推理

包含足够的错误上下文，供智能体决定下一步行动

区分可重试与不可重试的错误

对所有外部调用实现超时

工具权限模型

并非所有智能体都需要所有工具。使用最小权限原则：

按智能体角色定义工具集（客服智能体获得读取工具+发送邮件；计费智能体获得读取+写入计费工具）

明确分离读取和写入工具

高风险工具类别（删除、大额金融交易）需要人工批准

长时间运行智能体的状态管理

为什么状态管理很重要

LLM上下文窗口有限。一个100步的智能体工作流无法容纳在单个上下文中。长时间运行的智能体需要持久状态。

状态组件：

工作记忆：当前任务、最近动作、工具结果

长期记忆：执行过程中学到的事实、用户偏好

情景记忆：所有已执行动作的日志（用于审计和调试）

外部状态：对外部系统所做的更改（以便智能体推理当前状态）

检查点

在每个重要动作后保存智能体状态。好处：从中断处恢复、所有决策的审计跟踪、调试（在任何点重现智能体状态）。

LangGraph：通过SQLite、Redis或自定义后端内置检查点功能。智能体可以从任何检查点中断、恢复和分支。

python
from langgraph.checkpoint.sqlite import SqliteSaver
memory = SqliteSaver.from_conn_string("agent_checkpoints.db")
graph = StateGraph(AgentState)
... 添加节点和边 ...
app = graph.compile(checkpointer=memory)
从之前的检查点恢复智能体
config = {"configurable": {"thread_id": "task-123"}}
result = app.invoke(input, config=config)

人机协同模式

何时插入人工审核

并非所有智能体决策都应自主进行。在以下情况插入人工审核：

不可逆的高影响行动（删除客户数据、发送群发邮件、大额金融交易）

低置信度决策（智能体表示不确定、存在多个有效路径）

升级的客户服务（沮丧的用户、复杂边缘情况）

计划中的审核关卡（自主完成阶段A，阶段B前审核）

实现

LangGraph interrupt：内置机制，暂停执行、序列化状态、等待人工决策、恢复。

python
智能体在此暂停，等待人工批准
tool_result = interrupt({
    "type": "approval_required",
    "action": "send_email",
    "details": {"to": customer_email, "subject": "...", "body": "..."},
    "risk_level": "medium"
})
当人工批准/拒绝/修改后，执行恢复

用户界面：构建一个简单的审批仪表板，显示：待处理决策、上下文、推荐操作、批准/拒绝/修改选项。

智能体的可观测性

追踪什么

标准APM（应用性能监控）无法捕获智能体特定的信息。你需要：

每次LLM调用：输入提示、输出、模型、令牌使用量、延迟

每次工具调用：工具名称、参数、结果、延迟、成功/失败

智能体决策：智能体决定做什么以及为什么

状态变更：智能体状态的变化

错误和恢复尝试

追踪栈

LangSmith：专为LLM应用可观测性构建。每次智能体运行的完整追踪树。令牌使用量、延迟、模型调用、工具调用。生产监控与告警。

OpenTelemetry：标准可观测性协议。LangChain和LangGraph支持OTEL导出。发送到Jaeger、Zipkin、Datadog或任何OTEL后端。

AgentOps：较新的可观测性平台，专门针对智能体部署。

智能体特定指标

任务完成率：达到目标的智能体运行百分比

步数分布：智能体通常需要多少步？（异常值=问题）

按工具划分的工具失败率：哪些工具失败最多？

人工升级率：需要人工审核的运行百分比

每任务成本：令牌+时间

测试非确定性智能体

测试挑战

你无法为智能体编写确定性单元测试。相同的输入可能产生不同的输出。如何测试？

行为测试：测试智能体是否达成目标，而非具体路径。“智能体应成功完成订单退款”，允许多个有效执行路径。

模拟工具测试：用模拟替换真实工具。在隔离外部服务的情况下测试智能体推理。通过注入工具失败来测试错误处理。

轨迹分析：记录成功的智能体轨迹。测试新版本是否遵循相似的高级模式（访问相同的工具、到达相同的决策点）。

评估套件：定义50-100个代表性任务，附带明确的成功标准。定期运行。跨版本衡量通过率。

混沌测试：注入随机的工具失败、超时、意外响应。验证智能体能够优雅处理，而不会发生灾难性故障。

安全考量

提示注入：用户控制的输入将智能体重定向到攻击者的指令。缓解措施：明确分离系统和用户内容，根据原始任务范围验证智能体行动。

工具滥用：具有广泛权限的智能体在被操纵时可能造成损害。缓解措施：最小权限工具权限，高风险操作需操作确认。

数据泄露：能够读取和写入的智能体可能被操纵以泄露数据。缓解措施：分离读取和写入权限，审计所有写入操作。

资源耗尽：智能体可能无限循环。缓解措施：步数限制、成本限制、时间限制并优雅终止。