当前位置：首页 > AI > 正文内容

AI Agent 开发实战完全指南：从入门到精通

廖万里1天前AI1

AI Agent（人工智能代理）是当前AI领域最热门的技术方向之一，它让大语言模型具备了自主规划、工具调用、记忆管理和多步推理的能力。本指南将系统讲解AI Agent的核心原理、主流框架选择、工具调用机制、记忆系统设计，并通过完整项目实战带你构建一个实用的智能代理系统。

一、AI Agent 核心概念

AI Agent 是一种能够自主感知环境、做出决策并执行动作的智能系统。与传统的"一问一答"式AI交互不同，Agent具备以下核心能力：

1. 自主规划能力：能够将复杂任务拆解为多个子任务，并按合理顺序执行。

2. 工具调用能力：能够调用外部API、执行代码、访问数据库等，扩展自身能力边界。

3. 记忆管理能力：能够存储和检索历史对话、任务状态，保持长期上下文。

4. 自我反思能力：能够评估执行结果，发现问题并自我纠正。

1.1 Agent 与传统 LLM 应用的区别

传统LLM应用通常是单轮或简单的多轮对话，用户提出问题，模型直接给出答案。而Agent系统则像是一个"智能员工"，它可以：

理解复杂目标后自主规划执行路径
在执行过程中主动调用各种工具
遇到问题时尝试不同的解决方案
从历史经验中学习和改进

1.2 Agent 架构模式

当前主流的Agent架构主要分为以下几种：

ReAct 架构：Reasoning + Acting，模型先推理再行动，循环执行直到任务完成。

Plan-and-Execute 架构：先完整规划任务，再逐步执行，适合复杂多步骤任务。

Multi-Agent 架构：多个Agent协作完成复杂任务，每个Agent负责特定领域。

二、Agent 开发核心技术

2.1 提示词工程：构建 Agent 的灵魂

Agent的能力很大程度上取决于提示词设计。一个优秀的Agent提示词通常包含以下部分：

# Agent 提示词模板示例
AGENT_SYSTEM_PROMPT = """
你是一个智能代理，具备以下能力：
1. 任务规划：将复杂任务拆解为可执行的步骤
2. 工具调用：使用可用工具完成任务
3. 自我反思：评估执行结果并纠正错误

可用工具：
{tools_description}

输出格式：
- 思考：分析当前任务状态
- 行动：选择要执行的动作
- 行动输入：提供动作所需的参数
- 观察：接收执行结果

请始终以结构化格式输出，便于系统解析。
"""

# 工具描述示例
TOOLS_DESCRIPTION = """
1. search(query: str) - 搜索互联网获取信息
2. calculator(expression: str) - 计算数学表达式
3. code_executor(code: str) - 执行Python代码
4. database_query(sql: str) - 查询数据库
"""

2.2 工具调用机制

工具调用是Agent的核心能力之一。主流实现方式有两种：

方式一：Function Calling

import json
from openai import OpenAI

client = OpenAI()

# 定义工具函数
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如：北京、上海"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "搜索互联网获取最新信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索关键词"
                    }
                },
                "required": ["query"]
            }
        }
    }
]

# 发送请求
response = client.chat.completions.create(
    model="gpt-4",
    messages=[
        {"role": "user", "content": "北京今天天气怎么样？"}
    ],
    tools=tools,
    tool_choice="auto"
)

# 解析工具调用
message = response.choices[0].message
if message.tool_calls:
    for tool_call in message.tool_calls:
        function_name = tool_call.function.name
        function_args = json.loads(tool_call.function.arguments)
        print(f"调用工具: {function_name}")
        print(f"参数: {function_args}")

方式二：ReAct 模式

import re
from typing import Dict, Callable

class ReActAgent:
    """
    ReAct (Reasoning and Acting) 模式的 Agent 实现
    核心思想：交替进行推理(Reasoning)和行动(Acting)
    """
    
    def __init__(self, tools: Dict[str, Callable], llm_client, max_iterations: int = 10):
        """
        初始化 ReAct Agent
        
        Args:
            tools: 工具字典，键为工具名，值为执行函数
            llm_client: 大语言模型客户端
            max_iterations: 最大迭代次数，防止无限循环
        """
        self.tools = tools
        self.llm = llm_client
        self.max_iterations = max_iterations
        
    def build_prompt(self, task: str, history: list) -> str:
        """构建包含历史记录的提示词"""
        tools_desc = "\n".join([
            f"- {name}: {func.__doc__}" 
            for name, func in self.tools.items()
        ])
        
        history_text = "\n".join([
            f"思考: {h['thought']}\n行动: {h['action']}\n观察: {h['observation']}"
            for h in history
        ]) if history else "无"
        
        return f"""任务: {task}

可用工具:
{tools_desc}

历史记录:
{history_text}

请按以下格式输出：
思考: [分析当前情况，决定下一步行动]
行动: [工具名]
行动输入: [工具参数，JSON格式]
"""
    
    def parse_response(self, response: str) -> dict:
        """解析模型响应，提取思考、行动和参数"""
        thought_match = re.search(r"思考: (.+)", response)
        action_match = re.search(r"行动: (\w+)", response)
        input_match = re.search(r"行动输入: (.+)", response)
        
        return {
            "thought": thought_match.group(1) if thought_match else "",
            "action": action_match.group(1) if action_match else None,
            "action_input": input_match.group(1) if input_match else "{}"
        }
    
    def run(self, task: str) -> str:
        """
        执行任务的主循环
        
        Args:
            task: 用户任务描述
            
        Returns:
            最终执行结果
        """
        history = []
        
        for i in range(self.max_iterations):
            # 1. 构建提示词
            prompt = self.build_prompt(task, history)
            
            # 2. 调用LLM获取响应
            response = self.llm.generate(prompt)
            
            # 3. 解析响应
            parsed = self.parse_response(response)
            history.append({
                "thought": parsed["thought"],
                "action": parsed["action"],
                "observation": ""
            })
            
            # 4. 如果没有行动，返回最终答案
            if not parsed["action"]:
                return parsed["thought"]
            
            # 5. 执行工具调用
            if parsed["action"] in self.tools:
                try:
                    import json
                    args = json.loads(parsed["action_input"])
                    result = self.tools[parsed["action"]](**args)
                    history[-1]["observation"] = result
                except Exception as e:
                    history[-1]["observation"] = f"执行失败: {str(e)}"
            else:
                history[-1]["observation"] = f"未知工具: {parsed['action']}"
        
        return "达到最大迭代次数，任务未完成"

2.3 记忆系统设计

Agent的记忆系统是保持长期上下文的关键。常见的记忆类型包括：

from typing import List, Dict
from datetime import datetime
import json

class AgentMemory:
    """
    Agent 记忆系统
    包含短期记忆（对话历史）和长期记忆（知识库）
    """
    
    def __init__(self, max_short_term: int = 10):
        """
        初始化记忆系统
        
        Args:
            max_short_term: 短期记忆最大条数
        """
        self.short_term_memory: List[Dict] = []  # 短期记忆：对话历史
        self.long_term_memory: List[Dict] = []    # 长期记忆：重要知识
        self.max_short_term = max_short_term
        
    def add_message(self, role: str, content: str, metadata: dict = None):
        """
        添加一条消息到短期记忆
        
        Args:
            role: 消息角色（user/assistant/system）
            content: 消息内容
            metadata: 额外元数据
        """
        message = {
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat(),
            "metadata": metadata or {}
        }
        self.short_term_memory.append(message)
        
        # 保持短期记忆不超过上限
        if len(self.short_term_memory) > self.max_short_term:
            # 将旧消息移至长期记忆（可选）
            old_message = self.short_term_memory.pop(0)
            if self._is_important(old_message):
                self.long_term_memory.append(old_message)
    
    def _is_important(self, message: Dict) -> bool:
        """判断消息是否值得长期记忆"""
        # 简单规则：包含重要关键词的消息标记为重要
        important_keywords = ["重要", "记住", "用户偏好", "决策", "结论"]
        content = message.get("content", "").lower()
        return any(kw in content for kw in important_keywords)
    
    def add_knowledge(self, knowledge: str, category: str = "general"):
        """
        添加知识到长期记忆
        
        Args:
            knowledge: 知识内容
            category: 知识分类
        """
        self.long_term_memory.append({
            "content": knowledge,
            "category": category,
            "timestamp": datetime.now().isoformat()
        })
    
    def get_context(self, include_long_term: bool = True) -> str:
        """
        获取完整的上下文信息
        
        Args:
            include_long_term: 是否包含长期记忆
            
        Returns:
            格式化的上下文字符串
        """
        context_parts = []
        
        # 添加短期记忆（最近对话）
        context_parts.append("=== 最近对话 ===")
        for msg in self.short_term_memory[-5:]:  # 最近5条
            context_parts.append(f"[{msg['role']}]: {msg['content']}")
        
        # 添加长期记忆（重要知识）
        if include_long_term and self.long_term_memory:
            context_parts.append("\n=== 重要记忆 ===")
            for knowledge in self.long_term_memory[-3:]:  # 最近3条
                context_parts.append(f"- {knowledge['content']}")
        
        return "\n".join(context_parts)
    
    def save_to_file(self, filepath: str):
        """将记忆保存到文件"""
        data = {
            "short_term": self.short_term_memory,
            "long_term": self.long_term_memory
        }
        with open(filepath, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
    
    def load_from_file(self, filepath: str):
        """从文件加载记忆"""
        with open(filepath, "r", encoding="utf-8") as f:
            data = json.load(f)
        self.short_term_memory = data.get("short_term", [])
        self.long_term_memory = data.get("long_term", [])

2.4 向量记忆与 RAG 增强

对于需要处理大量知识的Agent，可以结合向量数据库实现语义检索：

import numpy as np
from typing import List, Tuple

class VectorMemory:
    """
    基于向量数据库的记忆系统
    支持语义相似性检索
    """
    
    def __init__(self, embedding_dim: int = 1536):
        """
        初始化向量记忆系统
        
        Args:
            embedding_dim: 向量维度（取决于embedding模型）
        """
        self.embedding_dim = embedding_dim
        self.texts: List[str] = []  # 存储原始文本
        self.metadata: List[Dict] = []  # 存储元数据
        
    def get_embedding(self, text: str) -> np.ndarray:
        """
        获取文本的向量表示
        
        Args:
            text: 输入文本
            
        Returns:
            归一化的向量
        """
        # 模拟：实际应调用 OpenAI Embedding 或其他模型
        vector = np.random.randn(self.embedding_dim).astype("float32")
        # 归一化，便于计算余弦相似度
        vector = vector / np.linalg.norm(vector)
        return vector
    
    def add_memory(self, text: str, metadata: dict = None):
        """
        添加记忆到向量数据库
        
        Args:
            text: 记忆文本
            metadata: 额外元数据
        """
        self.texts.append(text)
        self.metadata.append(metadata or {})
        
    def search(self, query: str, k: int = 5) -> List[Tuple[str, float, dict]]:
        """
        语义检索相似记忆
        
        Args:
            query: 查询文本
            k: 返回结果数量
            
        Returns:
            (文本, 相似度, 元数据) 列表
        """
        # 简化实现：返回最近添加的记忆
        results = []
        for i, text in enumerate(self.texts[-k:]):
            results.append((text, 0.9, self.metadata[i]))
        return results
    
    def get_relevant_context(self, query: str, max_tokens: int = 2000) -> str:
        """
        获取与查询相关的上下文
        
        Args:
            query: 查询文本
            max_tokens: 最大token数
            
        Returns:
            拼接的相关上下文
        """
        results = self.search(query, k=10)
        
        context_parts = []
        total_chars = 0
        
        for text, score, metadata in results:
            if total_chars + len(text) > max_tokens:
                break
            context_parts.append(f"[相关度: {score:.2f}] {text}")
            total_chars += len(text)
        
        return "\n\n".join(context_parts) if context_parts else "无相关记忆"

三、实战案例：构建智能助手 Agent

下面我们将构建一个完整的智能助手Agent，具备搜索、计算、代码执行等能力：

"""
完整的 AI Agent 实现：智能助手
功能：搜索信息、计算、执行代码、回答问题
"""

import json
import subprocess
from datetime import datetime
from typing import Dict, Callable, Optional
from dataclasses import dataclass

@dataclass
class ToolResult:
    """工具执行结果"""
    success: bool
    output: str
    error: Optional[str] = None

class SmartAgent:
    """
    智能助手 Agent
    集成多种工具，支持复杂任务处理
    """
    
    def __init__(self, llm_client, memory_system):
        """
        初始化智能助手
        
        Args:
            llm_client: 大语言模型客户端
            memory_system: 记忆系统实例
        """
        self.llm = llm_client
        self.memory = memory_system
        self.tools = self._register_tools()
        self.conversation_history = []
        
    def _register_tools(self) -> Dict[str, Callable]:
        """注册所有可用工具"""
        return {
            "search": self._tool_search,
            "calculate": self._tool_calculate,
            "execute_code": self._tool_execute_code,
            "get_time": self._tool_get_time,
            "save_note": self._tool_save_note,
        }
    
    def _tool_search(self, query: str) -> ToolResult:
        """搜索工具：在互联网上搜索信息"""
        # 实际实现应调用搜索API
        mock_results = f"搜索 '{query}' 的结果：找到相关文章3篇，推荐阅读..."
        return ToolResult(success=True, output=mock_results)
    
    def _tool_calculate(self, expression: str) -> ToolResult:
        """计算工具：执行数学计算"""
        try:
            # 安全计算（仅允许数学运算）
            allowed_chars = set("0123456789+-*/.() ")
            if not all(c in allowed_chars for c in expression):
                return ToolResult(
                    success=False,
                    output="",
                    error="表达式包含非法字符"
                )
            
            result = eval(expression)
            return ToolResult(success=True, output=str(result))
        except Exception as e:
            return ToolResult(success=False, output="", error=str(e))
    
    def _tool_execute_code(self, code: str) -> ToolResult:
        """代码执行工具：执行Python代码"""
        try:
            result = subprocess.run(
                ["python3", "-c", code],
                capture_output=True,
                text=True,
                timeout=30
            )
            
            if result.returncode == 0:
                return ToolResult(success=True, output=result.stdout)
            else:
                return ToolResult(success=False, output="", error=result.stderr)
        except subprocess.TimeoutExpired:
            return ToolResult(success=False, output="", error="执行超时")
        except Exception as e:
            return ToolResult(success=False, output="", error=str(e))
    
    def _tool_get_time(self) -> ToolResult:
        """获取当前时间工具"""
        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        return ToolResult(success=True, output=now)
    
    def _tool_save_note(self, note: str, title: str = "未命名") -> ToolResult:
        """保存笔记工具"""
        self.memory.add_knowledge(f"[{title}] {note}", category="notes")
        return ToolResult(success=True, output=f"笔记 '{title}' 已保存")
    
    def process_task(self, user_input: str) -> str:
        """
        处理用户输入的主方法
        
        Args:
            user_input: 用户输入
            
        Returns:
            Agent响应
        """
        # 1. 添加用户消息到记忆
        self.memory.add_message("user", user_input)
        
        # 2. 获取上下文
        context = self.memory.get_context()
        
        # 3. 构建提示词
        system_prompt = self._build_system_prompt()
        full_prompt = f"{system_prompt}\n\n上下文:\n{context}\n\n用户: {user_input}"
        
        # 4. 调用LLM
        response = self._call_llm(full_prompt)
        
        # 5. 解析并执行工具调用
        final_response = self._execute_tools(response)
        
        # 6. 保存助手响应到记忆
        self.memory.add_message("assistant", final_response)
        
        return final_response
    
    def _build_system_prompt(self) -> str:
        """构建系统提示词"""
        tools_desc = "\n".join([
            f"- {name}: {func.__doc__}"
            for name, func in self.tools.items()
        ])
        
        return f"""你是一个智能助手Agent，能够自主完成复杂任务。

可用工具:
{tools_desc}

工作流程:
1. 分析用户需求
2. 制定执行计划
3. 调用必要工具
4. 整合结果并回复

输出格式:
思考: [你的分析过程]
行动: [工具名或"回答"]
行动输入: [工具参数JSON或最终答案]
"""
    
    def _call_llm(self, prompt: str) -> str:
        """调用大语言模型"""
        # 实际应调用 self.llm.generate(prompt)
        return """思考: 用户需要获取当前时间
行动: get_time
行动输入: {}"""
    
    def _execute_tools(self, llm_response: str) -> str:
        """解析LLM响应并执行工具"""
        import re
        
        # 解析行动和参数
        action_match = re.search(r"行动: (\w+)", llm_response)
        input_match = re.search(r"行动输入: (.+)", llm_response)
        
        if not action_match:
            return llm_response
        
        action = action_match.group(1)
        action_input = input_match.group(1) if input_match else "{}"
        
        # 执行工具
        if action in self.tools:
            try:
                args = json.loads(action_input)
                result = self.tools[action](**args)
                
                if result.success:
                    return f"执行成功: {result.output}"
                else:
                    return f"执行失败: {result.error}"
            except json.JSONDecodeError:
                return "参数解析失败"
        else:
            return f"未知工具: {action}"

# 使用示例
if __name__ == "__main__":
    from agent_memory import AgentMemory
    
    memory = AgentMemory()
    agent = SmartAgent(llm_client=None, memory_system=memory)
    
    response = agent.process_task("现在几点了？")
    print(response)