AI Agent 生产级框架设计与实战落地

核心代码：Agent 基类

这里我们定义一个生产级的 Agent 基类。注意状态管理和迭代控制，这是防止死循环的关键。

from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from enum import Enum
import asyncio
import json

class AgentState(Enum):
    """Agent 状态枚举"""
    IDLE = "idle"          # 空闲
    THINKING = "thinking"  # 思考中
    ACTING = "acting"      # 执行中
    WAITING = "waiting"    # 等待外部输入
    ERROR = "error"        # 错误状态

@dataclass
class Message:
    """消息数据结构"""
    role: str              # user / assistant / system / tool
    content: str           # 消息内容
    tool_calls: Optional[List[Dict]] = None
    timestamp: float = None
    metadata: Dict[str, Any] = None

class BaseAgent:
    """生产级 Agent 基类"""
    def __init__(self, 
                 llm_client: Any,
                 memory_manager: Any = None,
                 tool_registry: Any = None,
                 max_iterations: int = 10,
                 verbose: bool = True):
        self.llm = llm_client
        self.memory = memory_manager
        self.tools = tool_registry
        self.max_iterations = max_iterations
        self.verbose = verbose
        self.state = AgentState.IDLE
        self.conversation_history: List[Message] = []

    async def run(self, user_input: str) -> str:
        """Agent 主执行循环"""
        # 添加用户消息到历史
        self.conversation_history.append(
            Message(role="user", content=user_input)
        )
        self.state = AgentState.THINKING

        for iteration in range(self.max_iterations):
            self._log(f"迭代 {iteration + 1}/{self.max_iterations}")
            
            # 1. 从记忆中检索相关信息
            context = await self._retrieve_context(user_input)
            
            # 2. 构建提示词
            prompt = self._build_prompt(context)
            
            # 3. LLM 推理
            response = await self._llm_inference(prompt)
            
            # 4. 检查是否需要调用工具
            if response.tool_calls:
                self.state = AgentState.ACTING
                # 执行工具调用
                tool_results = await self._execute_tools(response.tool_calls)
                # 将工具结果添加到历史
                for result in tool_results:
                    self.conversation_history.append(
                        Message(
                            role="tool", 
                            content=result["content"], 
                            tool_name=result["tool_name"]
                        )
                    )
            else:
                # 5. 无需工具调用，返回最终答案
                self.state = AgentState.IDLE
                self.conversation_history.append(
                    Message(role="assistant", content=response.content)
                )
                return response.content

        return "超过最大迭代次数，任务未完成"

    async def _retrieve_context(self, query: str) -> str:
        """从记忆中检索上下文"""
        if not self.memory:
            return ""
        # 检索相关记忆（这里简化处理）
        return await self.memory.search(query, top_k=3)

    def _build_prompt(self, context: str) -> str:
        """构建系统提示词"""
        system_prompt = f"""
        你是一个智能 AI 助手。
        # 可用工具
        {self.tools.get_tool_descriptions() if self.tools else '无'}
        # 相关记忆
        {context}
        # 任务要求
        1. 分析用户需求
        2. 如需信息查询或执行操作，调用相应工具
        3. 基于工具结果给出准确答案
        4. 如无法完成，明确说明原因
        开始工作！
        """
        return system_prompt

    async def _llm_inference(self, prompt: str) -> Any:
        """LLM 推理（示例使用 OpenAI 格式）"""
        messages = [
            {"role": "system", "content": prompt},
            *[{"role": m.role, "content": m.content} for m in self.conversation_history]
        ]
        response = await self.llm.chat.completions.create(
            model="gpt-4",
            messages=messages,
            tools=self.tools.get_tool_schemas() if self.tools else None,
            temperature=0.7
        )
        return response.choices[0].message

    async def _execute_tools(self, tool_calls: List[Dict]) -> List[Dict]:
        """执行工具调用"""
        results = []
        for call in tool_calls:
            tool_name = call["function"]["name"]
            arguments = json.loads(call["function"]["arguments"])
            self._log(f"调用工具：{tool_name} | 参数：{arguments}")
            try:
                # 从工具注册表获取工具并执行
                tool = self.tools.get_tool(tool_name)
                result = await tool.execute(**arguments)
                results.append({
                    "tool_name": tool_name,
                    "content": json.dumps(result, ensure_ascii=False)
                })
            except Exception as e:
                results.append({
                    "tool_name": tool_name,
                    "content": json.dumps({"error": str(e)})
                })
        return results

    def _log(self, message: str):
        """日志输出"""
        if self.verbose:
            print(f"[Agent] {message}")

记忆管理系统

记忆是 Agent 的'灵魂'。我们需要区分短期会话记忆和长期语义记忆。

from abc import ABC, abstractmethod
from typing import List, Dict, Any
import redis
import numpy as np
from datetime import datetime, timedelta
import json

class MemoryBackend(ABC):
    """记忆后端抽象基类"""
    @abstractmethod
    async def add(self, content: str, metadata: Dict = None) -> str:
        """添加记忆"""
        pass

    @abstractmethod
    async def search(self, query: str, top_k: int = 5) -> List[Dict]:
        """搜索记忆"""
        pass

class ShortTermMemory(MemoryBackend):
    """短期记忆：基于 Redis 的会话记忆"""
    def __init__(self, redis_url: str = "redis://localhost:6379", ttl: int = 3600):
        self.client = redis.from_url(redis_url)
        self.ttl = ttl  # 记忆过期时间（秒）

    async def add(self, content: str, metadata: Dict = None) -> str:
        """添加会话记忆"""
        memory_id = f"mem:{datetime.now().timestamp()}"
        memory_data = {
            "content": content,
            "metadata": metadata or {},
            "timestamp": datetime.now().isoformat()
        }
        self.client.setex(memory_id, self.ttl, json.dumps(memory_data, ensure_ascii=False))
        return memory_id

    async def search(self, query: str, top_k: int = 5) -> List[Dict]:
        """检索最近的相关记忆"""
        # 简化版：返回最近的记忆
        keys = self.client.keys("mem:*")
        memories = []
        for key in keys[-top_k:]:
            data = json.loads(self.client.get(key))
            memories.append(data)
        return memories

class LongTermMemory(MemoryBackend):
    """长期记忆：基于向量数据库的语义记忆"""
    def __init__(self, embedding_model: Any, vector_db: Any):
        self.embedding_model = embedding_model
        self.vector_db = vector_db

    async def add(self, content: str, metadata: Dict = None) -> str:
        """添加长期记忆"""
        # 生成向量嵌入
        embedding = await self.embedding_model.embed(content)
        # 存储到向量数据库
        memory_id = self.vector_db.insert(
            vector=embedding, 
            payload={"content": content, "metadata": metadata or {}}
        )
        return memory_id

    async def search(self, query: str, top_k: int = 5) -> List[Dict]:
        """语义搜索长期记忆"""
        # 查询向量嵌入
        query_embedding = await self.embedding_model.embed(query)
        # 向量检索
        results = self.vector_db.search(
            vector=query_embedding, 
            top_k=top_k, 
            score_threshold=0.7
        )
        return results

class HybridMemory:
    """混合记忆管理器：整合短期和长期记忆"""
    def __init__(self, short_term: ShortTermMemory, long_term: LongTermMemory):
        self.short_term = short_term
        self.long_term = long_term

    async def remember(self, content: str, importance: float = 0.5, metadata: Dict = None):
        """存储记忆（根据重要性决定存储位置）"""
        # 始终存入短期记忆
        await self.short_term.add(content, metadata)
        # 重要记忆存入长期记忆
        if importance > 0.7:
            await self.long_term.add(content, metadata)

    async def recall(self, query: str, top_k: int = 5) -> List[Dict]:
        """回忆相关记忆（整合短期和长期）"""
        # 并行检索
        short_results = await self.short_term.search(query, top_k // 2)
        long_results = await self.long_term.search(query, top_k // 2)
        # 合并去重
        all_results = short_results + long_results
        # 按相关性排序（这里简化）
        return all_results[:top_k]

三、三大核心技术实现

ReAct 框架：推理 + 行动协同

ReAct（Reasoning + Acting）是目前 Agent 最主流的推理范式。它让模型在思考和行动之间交替，而不是盲目猜测。

流程逻辑：

1. Thought: 分析问题，决定下一步做什么。
2. Action: 选择工具并准备参数。
3. Observation: 获取工具执行结果。
4. Answer: 综合信息给出最终回答。

代码实现：

class ReActAgent(BaseAgent):
    """基于 ReAct 范式的 Agent"""
    def _build_react_prompt(self, question: str) -> str:
        return f"""
        使用以下格式回答问题：
        Question: {question}
        Thought: 你应该思考做什么
        Action: 要采取的操作，应该是 [{self.tools.get_tool_names()}] 中的一个
        Observation: 操作的结果
        ... (这个 Thought/Action/Observation 可以重复 N 次)
        Thought: 我现在知道最终答案了
        Answer: 对原始问题的最终答案
        开始!
        Question: {question}
        Thought:
        """

    async def run(self, user_input: str) -> str:
        """ReAct 循环执行"""
        prompt = self._build_react_prompt(user_input)
        for _ in range(self.max_iterations):
            # LLM 生成下一步动作
            response = await self.llm.generate(prompt)
            
            # 解析响应
            thought, action, action_input = self._parse_react_response(response)
            
            if not action:
                # 没有动作，说明已有答案
                return thought
            
            # 执行动作
            observation = await self._execute_action(action, action_input)
            
            # 更新提示词
            prompt += f"\n{response}\nObservation: {observation}\nThought:"
        
        return "无法在指定迭代次数内完成"

    def _parse_react_response(self, response: str) -> tuple:
        """解析 ReAct 响应"""
        # 解析 Thought、Action、Action Input
        # 这里简化处理，实际需要更复杂的解析
        lines = response.strip().split('\n')
        thought = ""
        action = None
        action_input = None
        
        for line in lines:
            if line.startswith("Thought:"):
                thought = line.replace("Thought:", "").strip()
            elif line.startswith("Action:"):
                action = line.replace("Action:", "").strip()
            elif line.startswith("Action Input:"):
                action_input = line.replace("Action Input:", "").strip()
        
        return thought, action, action_input

工具调用系统

工具是 Agent 能力的边界。我们通过装饰器快速注册工具，并自动生成 Schema 供 LLM 识别。

from typing import Callable, Dict, Any, List
import inspect
from pydantic import BaseModel, Field

class Tool(BaseModel):
    """工具基类"""
    name: str = Field(description="工具名称")
    description: str = Field(description="工具功能描述")
    parameters: Dict[str, Any] = Field(default_factory=dict, description="参数 schema")
    function: Callable = Field(description="工具执行函数")

    class Config:
        arbitrary_types_allowed = True

    async def execute(self, **kwargs) -> Any:
        """执行工具"""
        return await self.function(**kwargs)

    def to_openai_schema(self) -> Dict:
        """转换为 OpenAI 函数调用格式"""
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters
            }
        }

def tool(name: str = None, description: str = None):
    """工具装饰器"""
    def decorator(func: Callable) -> Tool:
        # 提取函数签名
        sig = inspect.signature(func)
        parameters = {}
        for param_name, param in sig.parameters.items():
            param_type = param.annotation
            if param.annotation != inspect.Parameter.empty:
                param_type = param.annotation
            else:
                param_type = "string"
            
            type_name = param_type.__name__ if hasattr(param_type, "__name__") else "string"
            parameters[param_name] = {
                "type": type_name,
                "description": f"参数 {param_name}"
            }

        return Tool(
            name=name or func.__name__,
            description=description or func.__doc__ or "",
            parameters={
                "type": "object",
                "properties": parameters,
                "required": [p for p in sig.parameters if p.default == inspect.Parameter.empty]
            },
            function=func
        )
    return decorator

# 工具使用示例
@tool(name="search_web", description="搜索网络信息")
async def search_web(query: str, num_results: int = 5):
    """搜索网络信息
    Args:
        query: 搜索关键词
        num_results: 返回结果数量
    """
    # 实际实现调用搜索 API
    return f"找到 {num_results} 条关于 '{query}' 的结果"

@tool(name="get_weather", description="获取天气信息")
async def get_weather(location: str):
    """获取指定地点的天气信息
    Args:
        location: 城市名称
    """
    # 实际实现调用天气 API
    return f"{location} 今天晴，温度 25°C"

class ToolRegistry:
    """工具注册中心"""
    def __init__(self):
        self._tools: Dict[str, Tool] = {}

    def register(self, tool: Tool):
        """注册工具"""
        self._tools[tool.name] = tool

    def get_tool(self, name: str) -> Tool:
        """获取工具"""
        return self._tools.get(name)

    def get_tool_names(self) -> List[str]:
        """获取所有工具名称"""
        return list(self._tools.keys())

    def get_tool_descriptions(self) -> str:
        """获取工具描述文本"""
        descriptions = []
        for tool in self._tools.values():
            descriptions.append(f"- {tool.name}: {tool.description}")
        return "\n".join(descriptions)

    def get_tool_schemas(self) -> List[Dict]:
        """获取 OpenAI 格式的工具 schema"""
        return [tool.to_openai_schema() for tool in self._tools.values()]

任务规划器

对于复杂任务，单轮对话不够用，需要规划器将目标拆解为子任务序列。

class TaskPlanner:
    """任务分解与规划器"""
    def __init__(self, llm_client: Any):
        self.llm = llm_client

    async def plan(self, goal: str) -> List[Dict]:
        """将目标分解为子任务列表
        Returns:
            [ {"task": "任务描述", "order": 1, "dependencies": []}, ... ]
        """
        prompt = f"""
        将以下目标分解为具体的、可执行的子任务列表。
        目标：{goal}
        请按以下格式输出：
        1. [任务描述]
        2. [任务描述]
        ...
        要求：
        - 每个任务应该独立且可执行
        - 任务之间应该有逻辑顺序
        - 尽量细化到可以直接执行
        """
        response = await self.llm.generate(prompt)
        
        # 解析任务列表
        tasks = []
        for line in response.strip().split('\n'):
            if line.strip():
                # 提取任务描述
                task_desc = line.split('.', 1)[1].strip() if '.' in line else line.strip()
                tasks.append({
                    "task": task_desc,
                    "order": len(tasks) + 1,
                    "status": "pending"
                })
        return tasks

    async def execute_plan(self, agent: BaseAgent, tasks: List[Dict]) -> Dict:
        """执行任务计划
        Returns:
            { "success": bool, "completed_tasks": List[Dict], "failed_tasks": List[Dict], "final_result": Any }
        """
        completed = []
        failed = []
        
        for task in tasks:
            print(f"\n执行任务 {task['order']}: {task['task']}")
            try:
                # 使用 Agent 执行单个任务
                result = await agent.run(task['task'])
                task['status'] = 'completed'
                task['result'] = result
                completed.append(task)
            except Exception as e:
                task['status'] = 'failed'
                task['error'] = str(e)
                failed.append(task)
        
        return {
            "success": len(failed) == 0,
            "completed_tasks": completed,
            "failed_tasks": failed,
            "final_result": completed[-1]['result'] if completed else None
        }

四、实战案例：智能客服 Agent

智能客服是 AI Agent 最典型的应用场景。我们来实现一个政务大厅智能客服，具备政策问答、办事流程引导、工单生成及人工转接能力。

场景分析

• 政策咨询：RAG 检索知识库。
• 办事指引：流程引导 Agent。
• 投诉建议：情绪安抚 + 记录。
• 复杂问题：意图识别后转人工。

完整实现

import asyncio
from typing import Optional

class CustomerServiceAgent(ReActAgent):
    """智能客服 Agent"""
    def __init__(self, knowledge_base, ticket_system, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.knowledge_base = knowledge_base
        self.ticket_system = ticket_system
        # 注册客服专用工具
        self._register_customer_service_tools()

    def _register_customer_service_tools(self):
        """注册客服工具"""
        @self.tools.register
        @tool(name="search_policy", description="搜索政策信息")
        async def search_policy(query: str):
            """从知识库搜索相关政策
            Args:
                query: 政策关键词
            """
            results = await self.knowledge_base.search(query, top_k=3)
            return "\n".join([r['content'] for r in results])

        @self.tools.register
        @tool(name="get_process_guide", description="获取办事流程")
        async def get_process_guide(service_type: str):
            """获取指定业务的办事流程
            Args:
                service_type: 业务类型（如：身份证办理、社保卡申领）
            """
            guide = await self.knowledge_base.get_guide(service_type)
            return guide

        @self.tools.register
        @tool(name="create_ticket", description="创建工单")
        async def create_ticket(category: str, description: str, priority: str = "normal"):
            """创建服务工单
            Args:
                category: 工单类别
                description: 问题描述
                priority: 优先级（low/normal/high）
            """
            ticket_id = await self.ticket_system.create(
                category=category, 
                description=description, 
                priority=priority
            )
            return f"工单已创建，编号：{ticket_id}，我们将在 1 个工作日内处理"

        @self.tools.register
        @tool(name="transfer_to_human", description="转人工客服")
        async def transfer_to_human(reason: str):
            """转接到人工客服
            Args:
                reason: 转接原因
            """
            queue_number = await self.ticket_system.human_transfer(reason)
            return f"已为您转接人工客服，当前排队人数：{queue_number}人，预计等待时间：{queue_number * 2}分钟"

    async def handle_customer_query(self, user_input: str) -> str:
        """处理客户咨询"""
        # 意图识别
        intent = await self._detect_intent(user_input)
        # 根据意图调整系统提示
        system_prompt = self._get_system_prompt(intent)
        # 执行
        return await self.run(user_input)

    async def _detect_intent(self, user_input: str) -> str:
        """意图识别"""
        intent_prompt = f"""
        分类以下用户咨询的意图类型：
        用户输入：{user_input}
        意图类型：
        1. policy_inquiry - 政策咨询
        2. process_guide - 办事流程咨询
        3. complaint - 投诉建议
        4. complex - 复杂问题需人工
        只返回意图类型代码：
        """
        response = await self.llm.generate(intent_prompt)
        return response.strip()

    def _get_system_prompt(self, intent: str) -> str:
        """根据意图获取系统提示"""
        prompts = {
            "policy_inquiry": "你是政策咨询专员，请准确引用政策文件内容...",
            "process_guide": "你是办事引导员，请给出清晰的办事步骤...",
            "complaint": "你是投诉处理专员，请先安抚情绪，再记录问题...",
            "complex": "你是客服助理，对于复杂问题，请主动建议转人工..."
        }
        return prompts.get(intent, "你是智能客服助手...")

# 使用示例
async def main():
    from openai import AsyncOpenAI
    # 初始化
    llm_client = AsyncOpenAI(api_key="your-api-key")
    knowledge_base = MockKnowledgeBase()  # 模拟知识库
    ticket_system = MockTicketSystem()    # 模拟工单系统
    
    agent = CustomerServiceAgent(
        llm_client=llm_client,
        memory_manager=HybridMemory(
            short_term=ShortTermMemory(), 
            long_term=LongTermMemory()
        ),
        tool_registry=ToolRegistry(),
        knowledge_base=knowledge_base,
        ticket_system=ticket_system
    )
    
    # 处理咨询
    response = await agent.handle_customer_query("我想办理社保卡，需要准备什么材料？")
    print(response)

if __name__ == "__main__":
    asyncio.run(main())

性能对比

五、性能优化与成本控制

成本分析

AI Agent 的主要成本来源通常是 Token 消耗，其次是向量数据库和缓存服务。

• LLM Token 消耗: 约 45%
• 向量数据库: 约 25%
• Redis 缓存: 约 12%
• API 调用: 约 10%
• 其他: 约 8%

优化策略

智能缓存实现：

import hashlib
import time
from functools import wraps

def smart_cache(ttl: int = 3600):
    """智能缓存装饰器"""
    cache = {}
    
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # 生成缓存键
            key = hashlib.md5(f"{func.__name__}{args}{kwargs}".encode()).hexdigest()
            # 检查缓存
            if key in cache:
                cache_data = cache[key]
                if time.time() - cache_data['timestamp'] < ttl:
                    print("缓存命中!")
                    return cache_data['result']
            
            # 执行函数
            result = await func(*args, **kwargs)
            
            # 存储缓存
            cache[key] = {'result': result, 'timestamp': time.time()}
            return result
        return wrapper
    return decorator

# 使用示例
class OptimizedAgent(BaseAgent):
    @smart_cache(ttl=1800)
    async def _llm_inference(self, prompt: str):
        """带缓存的 LLM 推理"""
        return await super()._llm_inference(prompt)

总结

AI Agent 正在从'玩具'向'生产力工具'转变。掌握 Agent 开发已成为 AI 工程师的核心竞争力。

关键要点：

1. 记忆是核心竞争力：混合记忆架构（短期 + 长期）是最佳实践。
2. 工具调用决定能力边界：丰富的工具生态意味着更强的 Agent 能力。
3. 成本控制是关键：智能缓存 + 模型混合可节省 50%+ 成本。
4. 评估体系必不可少：建立完善的监控和评估体系。

下一步学习：

• 深入学习 LangChain / LangGraph
• 研究多 Agent 协作模式
• 探索 Agent + RAG 最佳实践
• 建立 Agent 评估体系

参考资源：

• OpenAI Function Calling 文档
• LangChain Agent 文档
• AutoGen 论文
• AgentScope 框架