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

2.2 核心代码：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}")

2.3 记忆管理系统

记忆分为短期会话记忆和长期语义记忆。短期用 Redis 存储，长期用向量数据库。

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]

三、三大核心技术实现

3.1 ReAct 框架：推理 + 行动协同

ReAct（Reasoning + Acting）是目前 Agent 最主流的推理范式。它让模型在思考和行动之间交替，通过观察工具结果来修正下一步计划。

代码实现：

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: 我现在知道最终答案了
Answer: 对原始问题的最终答案
开始!
Question: {question}
Thought:"""

    async def run(self, user_input: str) -> str:
        prompt = self._build_react_prompt(user_input)
        for _ in range(self.max_iterations):
            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:
        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

3.2 工具调用系统

我们需要一个统一的工具注册和管理机制，支持装饰器快速定义工具。

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:
        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 else "string"
            parameters[param_name] = {
                "type": param_type.__name__ if hasattr(param_type, "__name__") else "string",
                "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):
    return f"找到 {num_results} 条关于 '{query}' 的结果"

@tool(name="get_weather", description="获取天气信息")
async def get_weather(location: str):
    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]:
        return [tool.to_openai_schema() for tool in self._tools.values()]

3.3 任务规划器

对于复杂任务，需要将其拆解为子任务序列。

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

    async def plan(self, goal: str) -> List[Dict]:
        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:
        completed = []
        failed = []
        for task in tasks:
            print(f"\n执行任务 {task['order']}: {task['task']}")
            try:
                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

4.1 场景分析

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

4.2 完整实现

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):
            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):
            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"):
            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):
            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, "你是智能客服助手...")

if __name__ == "__main__":
    # 初始化逻辑略，实际运行需配置 LLM 客户端和环境变量
    pass

4.3 性能对比

五、性能优化与成本控制

5.1 成本分析

AI Agent 的主要成本来源通常是 LLM Token 消耗，其次是向量数据库和 API 调用费用。

5.2 优化策略

智能缓存实现：

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):
        return await super()._llm_inference(prompt)

六、总结

AI Agent 正在从概念验证走向生产力工具。掌握 Agent 开发已成为 AI 工程师的核心竞争力。

关键要点

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

参考资源

• OpenAI Function Calling 文档
• LangChain Agent 文档
• AutoGen 论文