生产级 Agent 项目实战

前面 13 篇文章学习了各种技术。现在把它们组合起来，从零构建一个生产级 AI Agent。本文是完整的实战项目指南——需求分析、架构设计、核心代码、错误恢复、日志监控、安全防护。

你将学到什么

• 生产级 Agent 的完整架构设计
• 核心组件的实现（Agent Loop、工具管理、记忆）
• 错误恢复和重试策略
• 日志、监控和安全防护

项目概述：智能客服 Agent

我们构建一个智能客服 Agent，它能够：

• 回答产品相关问题（基于知识库）
• 查询订单状态（调用内部 API）
• 处理退款请求（多步骤工作流）
• 升级到人工客服（兜底方案）

架构设计

┌─────────────────────────────────────────────┐
│                   Gateway                    │
│              (认证/限流/日志)                  │
├─────────────────────────────────────────────┤
│                Agent Loop                    │
│  ┌─────────┐  ┌──────────┐  ┌────────────┐ │
│  │ 推理引擎 │  │ 工具管理  │  │ 记忆系统   │ │
│  │(Claude)  │  │          │  │            │ │
│  └─────────┘  └──────────┘  └────────────┘ │
├─────────────────────────────────────────────┤
│                  Tools                       │
│  ┌──────┐ ┌──────┐ ┌──────┐ ┌───────────┐ │
│  │知识库 │ │订单API│ │退款API│ │升级人工   │ │
│  └──────┘ └──────┘ └──────┘ └───────────┘ │
└─────────────────────────────────────────────┘

核心实现

Agent 类

import anthropic
import logging
from datetime import datetime

logger = logging.getLogger("agent")

class CustomerServiceAgent:
    def __init__(self):
        self.client = anthropic.Anthropic()
        self.tools = self._define_tools()
        self.conversation_history = []
        self.max_iterations = 10

    def _define_tools(self):
        return [
            {
                "name": "search_knowledge_base",
                "description": "搜索产品知识库回答用户问题。输入关键词，返回相关文档片段。",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "query": {"type": "string", "description": "搜索关键词"}
                    },
                    "required": ["query"]
                }
            },
            {
                "name": "query_order",
                "description": "查询订单状态。输入订单号，返回订单详情。",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "order_id": {"type": "string", "description": "订单号"}
                    },
                    "required": ["order_id"]
                }
            },
            {
                "name": "process_refund",
                "description": "处理退款请求。需要订单号和退款原因。",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "order_id": {"type": "string"},
                        "reason": {"type": "string"}
                    },
                    "required": ["order_id", "reason"]
                }
            },
            {
                "name": "escalate_to_human",
                "description": "升级到人工客服。当问题超出 AI 能力范围时使用。",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "summary": {"type": "string", "description": "问题摘要"}
                    },
                    "required": ["summary"]
                }
            }
        ]

    def _execute_tool(self, name, inputs):
        """执行工具并处理错误"""
        try:
            if name == "search_knowledge_base":
                return self._search_kb(inputs["query"])
            elif name == "query_order":
                return self._query_order(inputs["order_id"])
            elif name == "process_refund":
                return self._process_refund(inputs["order_id"], inputs["reason"])
            elif name == "escalate_to_human":
                return self._escalate(inputs["summary"])
            return "未知工具"
        except Exception as e:
            logger.error(f"工具执行失败: {name}, 错误: {e}")
            return f"工具执行出错: {str(e)}"

    def handle_message(self, user_message):
        """处理用户消息的主循环"""
        self.conversation_history.append({
            "role": "user", "content": user_message
        })

        system_prompt = """你是一个专业的客服助手。请遵循以下原则：
1. 友好、专业地回答用户问题
2. 需要查询信息时主动使用工具
3. 不确定的信息不要猜测，使用知识库搜索
4. 复杂问题无法解决时，升级到人工客服
5. 回复要简洁明了"""

        for iteration in range(self.max_iterations):
            try:
                response = self.client.messages.create(
                    model="claude-sonnet-4-6",
                    max_tokens=2048,
                    system=system_prompt,
                    tools=self.tools,
                    messages=self.conversation_history,
                )
            except anthropic.RateLimitError:
                logger.warning("触发速率限制，等待重试")
                import time
                time.sleep(2)
                continue
            except anthropic.APIError as e:
                logger.error(f"API 错误: {e}")
                return "抱歉，系统暂时出现问题，请稍后再试。"

            if response.stop_reason == "tool_use":
                tool_results = []
                for block in response.content:
                    if block.type == "tool_use":
                        logger.info(f"调用工具: {block.name}")
                        result = self._execute_tool(block.name, block.input)
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": str(result),
                        })

                self.conversation_history.append({
                    "role": "assistant", "content": response.content
                })
                self.conversation_history.append({
                    "role": "user", "content": tool_results
                })
            else:
                # 生成最终回复
                reply = ""
                for block in response.content:
                    if hasattr(block, "text"):
                        reply += block.text

                self.conversation_history.append({
                    "role": "assistant", "content": reply
                })
                return reply

        return "抱歉，处理超时，请联系人工客服。"

    # 工具实现（示意）
    def _search_kb(self, query):
        return f"知识库搜索结果：关于'{query}'的信息..."

    def _query_order(self, order_id):
        return f"订单 {order_id}：已发货，预计明天送达"

    def _process_refund(self, order_id, reason):
        return f"退款已提交，订单 {order_id}，预计 3-5 个工作日到账"

    def _escalate(self, summary):
        return f"已升级到人工客服，摘要: {summary}"

日志和监控

import logging

# 配置结构化日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)

# 记录关键指标
class AgentMetrics:
    def __init__(self):
        self.total_requests = 0
        self.tool_calls = 0
        self.errors = 0
        self.total_tokens = 0

    def record_request(self, response):
        self.total_requests += 1
        self.total_tokens += response.usage.input_tokens
        self.total_tokens += response.usage.output_tokens

    def report(self):
        return {
            "total_requests": self.total_requests,
            "tool_calls": self.tool_calls,
            "errors": self.errors,
            "total_tokens": self.total_tokens,
        }

安全防护

1. 输入验证： 对用户输入做基本的清洗和长度限制。

2. 输出过滤： 确保 Agent 不泄露内部系统信息。

3. 权限控制： 敏感操作（如退款）需要额外验证。

4. 速率限制： 防止单用户滥用。

部署清单

Warning: 上线前确认以下事项。

• API Key 安全存储（环境变量，不是代码中）
• 错误处理和重试机制完备
• 日志和监控已配置
• 速率限制已设置
• 敏感操作有权限控制
• 有人工兜底方案
• 测试覆盖了常见场景和边界情况

实战练习

Tip: 基于本文框架构建你自己的 Agent。

1. 选择一个你熟悉的业务场景（客服、数据分析、运维）
2. 定义 3-5 个核心工具
3. 实现 Agent Loop + 错误处理
4. 添加日志和监控

关键要点

Note: 本文核心总结

• 生产级 Agent = Agent Loop + 工具管理 + 错误恢复 + 监控
• 每个工具调用都需要 try/catch 和日志
• API 错误用指数退避重试
• 一定要有人工兜底方案
• 安全防护（输入验证、权限控制、速率限制）不可少

延伸阅读

恭喜你完成了全部高级教程！你现在已经掌握了构建生产级 AI 应用的核心技能。继续实践，持续学习！