AI Agent实战：构建自主决策的智能体系统

AI Agent是能够自主感知环境、做出决策并执行行动的智能体。本文将深入探讨AI Agent的架构设计、实现方法和最佳实践。

AI Agent核心架构

基本组件

from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional
from dataclasses import dataclass
from enum import Enum

class ActionType(Enum):
    THINK = "think"
    ACT = "act"
    OBSERVE = "observe"
    WAIT = "wait"

@dataclass
class Observation:
    """环境观察"""
    state: Dict[str, Any]
    timestamp: float
    metadata: Optional[Dict[str, Any]] = None

@dataclass
class Action:
    """行动"""
    type: ActionType
    parameters: Dict[str, Any]
    reasoning: Optional[str] = None

@dataclass
class Goal:
    """目标"""
    description: str
    priority: float
    deadline: Optional[float] = None
    success_criteria: Optional[Dict[str, Any]] = None

class BaseAgent(ABC):
    """Agent基类"""
    
    def __init__(self, name: str, config: Dict[str, Any]):
        self.name = name
        self.config = config
        self.goals: List[Goal] = []
        self.memory = []
        self.tools = {}
    
    @abstractmethod
    def perceive(self) -> Observation:
        """感知环境"""
        pass
    
    @abstractmethod
    def decide(self, observation: Observation) -> Action:
        """决策"""
        pass
    
    @abstractmethod
    def act(self, action: Action) -> Any:
        """执行行动"""
        pass
    
    def update_memory(self, observation: Observation, action: Action, result: Any):
        """更新记忆"""
        self.memory.append({
            'observation': observation,
            'action': action,
            'result': result,
            'timestamp': observation.timestamp
        })
    
    def run(self, max_steps: int = 100):
        """运行Agent主循环"""
        for step in range(max_steps):
            # 感知
            observation = self.perceive()
            
            # 决策
            action = self.decide(observation)
            
            # 行动
            result = self.act(action)
            
            # 更新记忆
            self.update_memory(observation, action, result)
            
            # 检查目标是否完成
            if self.check_goals():
                break
​

基于LLM的Agent实现

ReAct模式（推理+行动）

import openai
from typing import List, Tuple

class ReActAgent(BaseAgent):
    """ReAct模式Agent：推理和行动交替"""
    
    def __init__(self, name: str, llm_config: Dict[str, Any]):
        super().__init__(name, llm_config)
        self.client = openai.OpenAI(api_key=llm_config['api_key'])
        self.model = llm_config.get('model', 'gpt-4')
    
    def perceive(self) -> Observation:
        """从环境中获取信息"""
        # 实现具体的感知逻辑
        return Observation(
            state={'env_status': 'active'},
            timestamp=time.time()
        )
    
    def think(self, observation: Observation) -> str:
        """LLM推理阶段"""
        prompt = f"""
        作为AI助手，基于以下信息进行思考：
        
        当前环境状态：{observation.state}
        当前目标：{self.goals[0].description if self.goals else '无'}
        历史记忆：{self.memory[-3:] if self.memory else '无'}
        
        请分析当前情况并制定下一步行动计划。
        """
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7
        )
        
        return response.choices[0].message.content
    
    def decide(self, observation: Observation) -> Action:
        """结合推理和工具使用进行决策"""
        # 思考阶段
        thought = self.think(observation)
        
        # 工具选择
        tool_prompt = f"""
        基于以下思考，选择最合适的工具和参数：
        
        思考：{thought}
        
        可用工具：{list(self.tools.keys())}
        
        请返回JSON格式：{{"tool": "工具名", "parameters": {{}}}}
        """
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": tool_prompt}],
            response_format={"type": "json_object"}
        )
        
        decision = json.loads(response.choices[0].message.content)
        
        return Action(
            type=ActionType.ACT,
            parameters=decision,
            reasoning=thought
        )
    
    def act(self, action: Action) -> Any:
        """执行行动"""
        tool_name = action.parameters.get('tool')
        if tool_name in self.tools:
            return self.tools[tool_name](**action.parameters.get('parameters', {}))
        return None
    
    def register_tool(self, name: str, func):
        """注册工具"""
        self.tools[name] = func
    
    def check_goals(self) -> bool:
        """检查目标是否完成"""
        return len(self.goals) == 0
​

多Agent协作

class MultiAgentSystem:
    """多Agent协作系统"""
    
    def __init__(self):
        self.agents: Dict[str, BaseAgent] = []
        self.communication_hub = CommunicationHub()
    
    def add_agent(self, agent: BaseAgent):
        """添加Agent"""
        self.agents[agent.name] = agent
        agent.communication = self.communication_hub
    
    def orchestrate_task(self, task: Dict[str, Any]) -> Any:
        """协调多个Agent完成任务"""
        # 任务分解
        subtasks = self.decompose_task(task)
        
        # Agent选择
        agent_assignments = self.assign_agents(subtasks)
        
        # 执行和协调
        results = {}
        for subtask, agent_name in agent_assignments.items():
            agent = self.agents[agent_name]
            result = self.execute_with_monitoring(agent, subtask)
            results[subtask['id']] = result
        
        # 结果整合
        return self.integrate_results(results)
    
    def decompose_task(self, task: Dict[str, Any]) -> List[Dict]:
        """任务分解"""
        prompt = f"""
        将以下任务分解为多个子任务：
        
        任务描述：{task['description']}
        
        请返回JSON格式的子任务列表。
        """
        
        # 使用LLM进行任务分解
        response = self.llm_client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"}
        )
        
        return json.loads(response.choices[0].message.content)['subtasks']
    
    def assign_agents(self, subtasks: List[Dict]) -> Dict[str, str]:
        """为子任务分配最合适的Agent"""
        assignments = {}
        
        for subtask in subtasks:
            # 基于Agent能力匹配
            best_agent = self.find_best_agent(subtask)
            assignments[subtask['id']] = best_agent
        
        return assignments

class CommunicationHub:
    """Agent间通信中心"""
    
    def __init__(self):
        self.message_queue = []
        self.message_history = []
    
    def send_message(self, sender: str, receiver: str, message: Dict[str, Any]):
        """发送消息"""
        msg = {
            'sender': sender,
            'receiver': receiver,
            'content': message,
            'timestamp': time.time()
        }
        self.message_queue.append(msg)
    
    def receive_messages(self, receiver: str) -> List[Dict]:
        """接收消息"""
        messages = [msg for msg in self.message_queue if msg['receiver'] == receiver]
        self.message_queue = [msg for msg in self.message_queue if msg['receiver'] != receiver]
        self.message_history.extend(messages)
        return messages
​

工具使用能力

工具定义和注册

from typing import Callable, Any
import inspect

class ToolRegistry:
    """工具注册表"""
    
    def __init__(self):
        self.tools: Dict[str, Dict[str, Any]] = {}
    
    def register(self, name: str, func: Callable, description: str):
        """注册工具"""
        self.tools[name] = {
            'function': func,
            'description': description,
            'parameters': self._extract_parameters(func)
        }
    
    def _extract_parameters(self, func: Callable) -> Dict[str, Any]:
        """提取函数参数信息"""
        sig = inspect.signature(func)
        parameters = {}
        
        for name, param in sig.parameters.items():
            param_info = {
                'type': str(param.annotation) if param.annotation != inspect.Parameter.empty else 'any',
                'required': param.default == inspect.Parameter.empty,
                'default': param.default if param.default != inspect.Parameter.empty else None
            }
            parameters[name] = param_info
        
        return parameters
    
    def get_tool(self, name: str) -> Optional[Callable]:
        """获取工具函数"""
        if name in self.tools:
            return self.tools[name]['function']
        return None
    
    def list_tools(self) -> List[str]:
        """列出所有工具"""
        return list(self.tools.keys())

# 使用示例
def search_web(query: str, num_results: int = 5) -> List[Dict[str, str]]:
    """搜索网络"""
    # 实现搜索逻辑
    return [
        {'title': f"Result {i}", 'url': f"http://example.com/{i}"}
        for i in range(num_results)
    ]

def calculate(expression: str) -> float:
    """计算数学表达式"""
    return eval(expression)

registry = ToolRegistry()
registry.register('search_web', search_web, '在互联网上搜索信息')
registry.register('calculate', calculate, '计算数学表达式')
​

智能工具选择

class ToolSelector:
    """智能工具选择器"""
    
    def __init__(self, llm_client, tool_registry: ToolRegistry):
        self.llm_client = llm_client
        self.registry = tool_registry
    
    def select_tool(self, task: str) -> Tuple[str, Dict[str, Any]]:
        """为任务选择最合适的工具"""
        tools_info = self._get_tools_info()
        
        prompt = f"""
        为以下任务选择最合适的工具：
        
        任务：{task}
        
        可用工具：
        {tools_info}
        
        请返回JSON格式：{{"tool": "工具名", "parameters": {{}}}}
        """
        
        response = self.llm_client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"}
        )
        
        result = json.loads(response.choices[0].message.content)
        return result['tool'], result.get('parameters', {})
    
    def _get_tools_info(self) -> str:
        """获取工具信息"""
        info = []
        for name, tool_data in self.registry.tools.items():
            info.append(f"- {name}: {tool_data['description']}")
        return '\n'.join(info)
​

记忆系统

短期记忆与长期记忆

from typing import List, Dict, Any
import numpy as np
from datetime import datetime, timedelta

class MemorySystem:
    """Agent记忆系统"""
    
    def __init__(self, short_term_capacity: int = 10, long_term_capacity: int = 1000):
        self.short_term: List[Dict[str, Any]] = []
        self.long_term: List[Dict[str, Any]] = []
        self.short_term_capacity = short_term_capacity
        self.long_term_capacity = long_term_capacity
    
    def add_memory(self, memory: Dict[str, Any]):
        """添加记忆"""
        memory['timestamp'] = datetime.now()
        
        # 添加到短期记忆
        self.short_term.append(memory)
        
        # 限制短期记忆容量
        if len(self.short_term) > self.short_term_capacity:
            # 将最旧的记忆转移到长期记忆
            old_memory = self.short_term.pop(0)
            self._consolidate_to_long_term(old_memory)
    
    def _consolidate_to_long_term(self, memory: Dict[str, Any]):
        """将记忆固化到长期记忆"""
        memory['importance'] = self._calculate_importance(memory)
        
        self.long_term.append(memory)
        
        # 限制长期记忆容量
        if len(self.long_term) > self.long_term_capacity:
            # 移除重要性最低的记忆
            self.long_term.sort(key=lambda x: x['importance'])
            self.long_term.pop(0)
    
    def _calculate_importance(self, memory: Dict[str, Any]) -> float:
        """计算记忆重要性"""
        # 基于多个因素计算重要性
        factors = []
        
        # 最近性
        age = (datetime.now() - memory['timestamp']).total_seconds()
        factors.append(1.0 / (1.0 + age / 3600))  # 1小时衰减
        
        # 访问频率（如果有）
        factors.append(memory.get('access_count', 0) * 0.1)
        
        # 情感强度（如果有）
        factors.append(memory.get('emotional_intensity', 0) * 0.5)
        
        return sum(factors) / len(factors)
    
    def retrieve_relevant(self, query: str, top_k: int = 5) -> List[Dict[str, Any]]:
        """检索相关记忆"""
        # 简单实现：基于关键词匹配
        # 实际应用中可以使用向量搜索
        relevant = []
        
        for memory in self.short_term + self.long_term:
            score = self._similarity_score(query, memory)
            if score > 0.5:  # 阈值
                relevant.append((memory, score))
        
        # 按相似度排序
        relevant.sort(key=lambda x: x[1], reverse=True)
        
        return [mem for mem, score in relevant[:top_k]]
    
    def _similarity_score(self, query: str, memory: Dict[str, Any]) -> float:
        """计算相似度分数"""
        # 简单实现：关键词重叠
        query_words = set(query.lower().split())
        memory_text = str(memory).lower()
        
        overlap = sum(1 for word in query_words if word in memory_text)
        return overlap / len(query_words) if query_words else 0.0
​

实战案例

客服Agent

class CustomerServiceAgent(ReActAgent):
    """客服Agent"""
    
    def __init__(self, knowledge_base, llm_config):
        super().__init__("客服Agent", llm_config)
        self.knowledge_base = knowledge_base
        
        # 注册工具
        self.register_tool('search_knowledge', self._search_knowledge, '搜索知识库')
        self.register_tool('create_ticket', self._create_ticket, '创建工单')
        self.register_tool('check_order_status', self._check_order_status, '查询订单状态')
    
    def _search_knowledge(self, query: str) -> str:
        """搜索知识库"""
        results = self.knowledge_base.search(query)
        return results[0] if results else "未找到相关信息"
    
    def _create_ticket(self, issue: str, customer_id: str) -> str:
        """创建工单"""
        ticket_id = f"TKT-{int(time.time())}"
        return f"工单{ticket_id}已创建"
    
    def _check_order_status(self, order_id: str) -> str:
        """查询订单状态"""
        # 模拟查询
        return f"订单{order_id}状态：已发货"
    
    def handle_customer_query(self, query: str, customer_id: str) -> str:
        """处理客户查询"""
        # 感知
        observation = Observation(
            state={'query': query, 'customer_id': customer_id},
            timestamp=time.time()
        )
        
        # 决策
        action = self.decide(observation)
        
        # 行动
        result = self.act(action)
        
        return result
​

研究Agent

class ResearchAgent(ReActAgent):
    """研究Agent"""
    
    def __init__(self, llm_config):
        super().__init__("研究Agent", llm_config)
        
        # 注册工具
        self.register_tool('search_paper', self._search_paper, '搜索学术论文')
        self.register_tool('summarize_paper', self._summarize_paper, '总结论文')
        self.register_tool('analyze_trends', self._analyze_trends, '分析趋势')
    
    def conduct_research(self, topic: str) -> Dict[str, Any]:
        """进行研究"""
        # 搜索相关论文
        papers = self._search_paper(topic)
        
        # 总结论文
        summaries = [self._summarize_paper(paper) for paper in papers]
        
        # 分析趋势
        trends = self._analyze_trends(summaries)
        
        return {
            'topic': topic,
            'papers': papers,
            'summaries': summaries,
            'trends': trends
        }
​

最佳实践

1. 明确目标定义：为Agent设置清晰、可衡量的目标
2. 工具设计：工具应该是简单、可靠、可组合的
3. 记忆管理：合理设计记忆的存储和检索机制
4. 错误处理：Agent应该能够优雅地处理失败
5. 监控和调试：建立完善的监控体系
6. 安全考虑：限制Agent的权限和访问范围

总结

AI Agent通过结合LLM的推理能力和工具使用能力，可以自主完成复杂的任务。理解Agent的架构设计、记忆系统和工具使用机制，对于构建实用的AI应用至关重要。随着技术的不断发展，AI Agent将在越来越多的领域发挥重要作用。