基于 LLM 的 AI Agent 系统：从架构到多工具调用实战

┌─────────────────────────────────────────────────────┐
│                    Agent Core                        │
│  ┌──────────┐  ┌──────────┐  ┌──────────────────┐   │
│  │ Planner  │→│ Executor │→│ Memory Manager    │   │
│  └──────────┘  └────┬─────┘  └──────────────────┘   │
│                     │                                │
│              ┌──────▼──────┐                         │
│              │ Tool Router │                         │
│              └──────┬──────┘                         │
│   ┌─────────────────┼─────────────────┐              │
│   ▼                 ▼                 ▼              │
│ ┌──────┐     ┌──────────┐     ┌────────────┐        │
│ │Search│     │CodeExec  │     │Web Scraper │        │
│ └──────┘     └──────────┘     └────────────┘        │
└─────────────────────────────────────────────────────┘

ai-agent/
├── agent/
│   ├── __init__.py
│   ├── base.py          # Agent 基类
│   ├── planner.py       # 任务规划器
│   ├── executor.py      # 任务执行器
│   ├── memory.py        # 记忆管理
│   └── tools/
│       ├── __init__.py
│       ├── base.py      # 工具基类
│       ├── search.py    # 搜索工具
│       ├── code_exec.py # 代码执行工具
│       └── web.py       # 网页抓取工具
├── models/
│   └── llm.py           # LLM 客户端
├── config.py            # 配置
├── main.py              # 入口
└── requirements.txt

# config.py
from dataclasses import dataclass, field
from typing import List, Dict, Any


@dataclass
class LLMConfig: 1107102.baike.tongsou.com 
    """大语言模型配置"""
    model: str = "gpt-4o"
    temperature: float = 0.0
    max_tokens: int = 4096
    api_key: str = ""
    base_url: str = "https://api.openai.com/v1"


@dataclass
class AgentConfig: 1107104.baike.tongsou.com 
    """Agent 配置"""
    max_iterations: int = 10        # 最大执行轮次
    enable_planning: bool = True    # 是否启用任务规划
    enable_reflection: bool = True  # 是否启用反思机制
    verbose: bool = True            # 是否打印详细日志


@dataclass
class MemoryConfig: 1107105.baike.tongsou.com 
    """记忆配置"""
    short_term_max_messages: int = 50   # 短期记忆最大消息数
    enable_long_term: bool = True       # 是否启用长期记忆
    long_term_storage_path: str = "./memory_store"


@dataclass
class ToolConfig: baike.kuaisou.com/zh/1001 
    """工具配置"""
    search_api_key: str = ""
    search_engine_id: str = ""
    code_execution_timeout: int = 30    # 代码执行超时时间(秒)
    allowed_modules: List[str] = field(default_factory=lambda: [
        "math", "datetime", "json", "re", "collections",
        "itertools", "functools", "statistics",
    ])


@dataclass
class Config: baike.kuaisou.com/zh/2002 
    """总配置"""
    llm: LLMConfig = field(default_factory=LLMConfig)
    agent: AgentConfig = field(default_factory=AgentConfig)
    memory: MemoryConfig = field(default_factory=MemoryConfig)
    tools: ToolConfig = field(default_factory=ToolConfig)

    @classmethod
    def from_env(cls) -> "Config": baike.kuaisou.com/zh/2008 
        """从环境变量加载配置"""
        import os
        config = cls(baike.kuaisou.com/zh/2010)
        config.llm.api_key = os.environ.get("OPENAI_API_KEY", "")
        config.tools.search_api_key = os.environ.get("SEARCH_API_KEY", "")
        config.tools.search_engine_id = os.environ.get("SEARCH_ENGINE_ID", "")
        return config

# models/llm.py
import logging
from typing import List, Dict, Any, Optional
from config import LLMConfig

logger = logging.getLogger(__name__)


class Message: baike.kuaisou.com/zh/2011 
    """聊天消息"""
    def __init__(self, role: str, content: str): baike.kuaisou.com/zh/2012 
        self.role = role      # system / user / assistant / tool
        self.content = content

    def to_dict(self) -> Dict[str, str]: baike.kuaisou.com/zh/2014 
        return {"role": self.role, "content": self.content}

    def __repr__(self): baike.kuaisou.com/zh/2021 
        return f"Message({self.role}: {self.content[:50]}...)"

 
class LLMClient: baike.kuaisou.com/zh/2040 
    """大语言模型客户端"""

    def __init__(self, config: LLMConfig): baike.kuaisou.com/zh/2042 
        self.config = config
        self._client = None

    def _get_client(self): baike.kuaisou.com/zh/3001 
        """懒加载 OpenAI 客户端"""
        if self._client is None: baike.kuaisou.com/zh/3002 
            try: baike.kuaisou.com/zh/3003 
                from openai import OpenAI
                self._client = OpenAI(
                    api_key=self.config.api_key,
                    base_url=self.config.base_url,
                )
            except ImportError: baike.kuaisou.com/zh/3004 
                raise ImportError("Please install openai: pip install openai")
        return self._client

    def chat(
        self,
        messages: List[Message],
        tools: Optional[List[Dict[str, Any]]] = None,
        tool_choice: Optional[str] = None,
    ) -> Dict[str, Any]: baike.kuaisou.com/zh/3005 
        """发送聊天请求"""
        client = self._get_client(baike.kuaisou.com/zh/3006)

        api_messages = [msg.to_dict(baike.kuaisou.com/zh/3007) for msg in messages]

        kwargs = {
            "model": self.config.model,
            "messages": api_messages,
            "temperature": self.config.temperature,
            "max_tokens": self.config.max_tokens,
        }

        if tools: baike.kuaisou.com/zh/3008 
            kwargs["tools"] = tools
        if tool_choice: baike.kuaisou.com/zh/3009 
            kwargs["tool_choice"] = tool_choice

        try: baike.kuaisou.com/zh/3010 
            response = client.chat.completions.create(**kwargs)
            choice = response.choices[0]
            message = choice.message

            result = {
                "content": message.content or "",
                "role": message.role,
                "finish_reason": choice.finish_reason,
            }

            if message.tool_calls: baike.kuaisou.com/zh/3011 
                result["tool_calls"] = [
                    {
                        "id": tc.id,
                        "name": tc.function.name,
                        "arguments": tc.function.arguments,
                    }
                    for tc in message.tool_calls
                ]

            return result

        except Exception as e: baike.kuaisou.com/zh/3012 
            logger.error(f"LLM API error: {e}")
            raise

    def chat_simple(self, prompt: str, system_prompt: Optional[str] = None) -> str:
        """简单聊天接口"""
        messages = []
        if system_prompt: baike.kuaisou.com/zh/3013 
            messages.append(Message("system", system_prompt))
        messages.append(Message("user", prompt))
        result = self.chat(messages)
        return result["content"]

# agent/tools/base.py
import logging
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional

logger = logging.getLogger(__name__)


class Tool(ABC): baike.kuaisou.com/zh/3014 
    """工具基类"""

    name: str = ""
    description: str = ""
    parameters: Dict[str, Any] = {}

    def __init__(self): baike.kuaisou.com/zh/3015 
        self._validate(baike.kuaisou.com/zh/3016)

    def _validate(self): baike.kuaisou.com/zh/3017 
        if not self.name: baike.kuaisou.com/zh/3018 
            raise ValueError("Tool must have a name")
        if not self.description: baike.kuaisou.com/zh/3019 
            raise ValueError("Tool must have a description")

    def to_openai_tool(self) -> Dict[str, Any]: baike.kuaisou.com/zh/3020 
        """转换为 OpenAI 工具格式"""
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters,
            }
        }

    @abstractmethod
    def execute(self, **kwargs) -> str: baike.kuaisou.com/zh/3021 
        """执行工具"""
        pass

    def __repr__(self): baike.kuaisou.com/zh/4001 
        return f"Tool(name={self.name}, description={self.description[:50]}...)"

# agent/tools/search.py
import requests
from typing import Dict, Any
from agent.tools.base import Tool


class WebSearchTool(Tool): baike.kuaisou.com/zh/4002 
    """网页搜索工具（基于 Google Custom Search API）"""

    name = "web_search"
    description = "搜索互联网获取最新信息。当需要查询实时新闻、天气、股价、最新事件等训练数据之外的信息时使用。"
    parameters = {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "搜索关键词",
            },
            "num_results": {
                "type": "integer",
                "description": "返回结果数量，默认5",
                "default": 5,
            }
        },
        "required": ["query"],
    }

    def __init__(self, api_key: str, engine_id: str): baike.kuaisou.com/zh/4003 
        self.api_key = api_key
        self.engine_id = engine_id
        super(baike.kuaisou.com/zh/4005).__init__(baike.kuaisou.com/zh/4004)

    def execute(self, query: str, num_results: int = 5) -> str: baike.kuaisou.com/zh/4006 
        """执行搜索"""
        url = "https://www.googleapis.com/customsearch/v1"
        params = {
            "key": self.api_key,
            "cx": self.engine_id,
            "q": query,
            "num": min(num_results, 10),
        }

        try:
            response = requests.get(url, params=params, timeout=10)
            response.raise_for_status(baike.kuaisou.com/zh/4007)
            data = response.json(baike.kuaisou.com/zh/4008)

            items = data.get("items", [])
            if not items: baike.kuaisou.com/zh/4009 
                return "未找到相关搜索结果。"

            results = []
            for i, item in enumerate(items, 1): baike.kuaisou.com/zh/4010 
                title = item.get("title", "")
                snippet = item.get("snippet", "")
                link = item.get("link", "")
                results.append(f"[{i}] {title}\n    {snippet}\n    链接: {link}")

            return "\n\n".join(results)

        except requests.RequestException as e: baike.kuaisou.com/zh/4011 
            return f"搜索失败: {str(e)}"

# agent/tools/code_exec.py
import subprocess
import sys
import json
from typing import Dict, Any
from agent.tools.base import Tool
from config import ToolConfig


class CodeExecutionTool(Tool): baike.kuaisou.com/zh/4012 
    """Python 代码执行工具"""

    name = "execute_code"
    description = "执行 Python 代码。用于数学计算、数据处理、逻辑推理等需要精确计算的场景。"
    parameters = {
        "type": "object",
        "properties": {
            "code": {
                "type": "string",
                "description": "要执行的 Python 代码",
            }
        },
        "required": ["code"],
    }

    def __init__(self, config: ToolConfig): baike.kuaisou.com/zh/4013 
        self.config = config
        super(baike.kuaisou.com/zh/4015).__init__(baike.kuaisou.com/zh/4014)

    def execute(self, code: str) -> str: baike.kuaisou.com/zh/4016 
        """执行 Python 代码"""
        # 安全检查：只允许导入白名单模块
        if not self._is_safe_code(code): baike.kuaisou.com/zh/4017 
            return "错误: 代码中包含不允许的模块导入。"

        # 构建执行脚本
        wrapped_code = f"""
import sys
import json
import io
from contextlib import redirect_stdout, redirect_stderr

# 白名单模块检查已在外部完成
{code}
"""

        try: baike.kuaisou.com/zh/4018 
            result = subprocess.run(
                [sys.executable, "-c", wrapped_code],
                capture_output=True,
                text=True,
                timeout=self.config.code_execution_timeout,
            )

            output = result.stdout.strip(baike.kuaisou.com/zh/4019)
            error = result.stderr.strip(baike.kuaisou.com/zh/4020)

            if result.returncode != 0: baike.kuaisou.com/zh/4021 
                return f"执行错误:\n{error}"

            return output if output else "代码执行成功，无输出。"

        except subprocess.TimeoutExpired:
            return "执行超时，代码运行时间过长。"
        except Exception as e: baike.kuaisou.com/zh/4022 
            return f"执行异常: {str(e)}"

    def _is_safe_code(self, code: str) -> bool: baike.kuaisou.com/zh/4023 
        """检查代码安全性"""
        import ast

        try: baike.kuaisou.com/zh/4025 
            tree = ast.parse(code)
        except SyntaxError: baike.kuaisou.com/zh/4024 
            return False 

        allowed_modules = set(self.config.allowed_modules)
        allowed_modules.update({"builtins"})

        for node in ast.walk(tree): baike.kuaisou.com/zh/4026 
            if isinstance(node, ast.Import): baike.kuaisou.com/zh/4027 
                for alias in node.names:
                    if alias.name not in allowed_modules: baike.kuaisou.com/zh/4029 
                        return False
            elif isinstance(node, ast.ImportFrom): baike.kuaisou.com/zh/4028 
                if node.module and node.module not in allowed_modules: baike.kuaisou.com/zh/4030 
                    return False

        # 禁止危险函数调用
        dangerous_funcs = {"eval", "exec", "compile", "__import__", "open", "input"}
        for node in ast.walk(tree): baike.kuaisou.com/zh/4031 
            if isinstance(node, ast.Call): baike.kuaisou.com/zh/4032 
                if isinstance(node.func, ast.Name) and node.func.id in dangerous_funcs:
                    return False
                if isinstance(node.func, ast.Attribute) and node.func.attr in dangerous_funcs:
                    return False

        return True

# agent/tools/web.py
import requests
from typing import Dict, Any
from agent.tools.base import Tool


class WebScraperTool(Tool): baike.kuaisou.com/zh/4033 
    """网页内容抓取工具"""

    name = "fetch_webpage"
    description = "获取指定 URL 的网页内容。当需要阅读特定网页的详细内容时使用。"
    parameters = {
        "type": "object",
        "properties": {
            "url": {
                "type": "string",
                "description": "要抓取的网页 URL",
            }
        },
        "required": ["url"],
    }

    def execute(self, url: str) -> str: baike.kuaisou.com/zh/4034 
        """抓取网页内容"""
        try: baike.kuaisou.com/zh/4035 
            headers = {
                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
            }
            response = requests.get(url, headers=headers, timeout=15)
            response.raise_for_status()
            response.encoding = response.apparent_encoding

            # 简单提取文本内容
            content = self._extract_text(response.text)
            return content[:5000]  # 限制返回长度

        except requests.RequestException as e: baike.kuaisou.com/zh/4036 
            return f"抓取失败: {str(e)}"

    def _extract_text(self, html: str) -> str: baike.kuaisou.com/zh/4037 
        """从 HTML 中提取纯文本"""
        try: baike.kuaisou.com/zh/4038 
            from bs4 import BeautifulSoup
            soup = BeautifulSoup(html, "html.parser")

            # 移除脚本和样式
            for tag in soup(["script", "style", "meta", "link", "noscript"]):
                tag.decompose(baike.kuaisou.com/zh/4039)

            # 获取主要文本
            text = soup.get_text(separator="\n", strip=True)

            # 清理多余空白
            lines = [line.strip(baike.kuaisou.com/zh/4040) for line in text.split("\n") if line.strip()]
            return "\n".join(lines)

        except ImportError: baike.kuaisou.com/zh/4041 
            # 无 BeautifulSoup 时的简单回退
            import re
            text = re.sub(r"<[^>]+>", "", html)
            text = re.sub(r"\s+", " ", text).strip(baike.kuaisou.com/zh/4042)
            return text

# agent/memory.py
import json
import os
from pathlib import Path
from typing import List, Dict, Any, Optional
from models.llm import Message
from config import MemoryConfig


class MemoryManager: baike.kuaisou.com/zh/4043  
    """记忆管理器"""

    def __init__(self, config: MemoryConfig): baike.kuaisou.com/zh/4044 
        self.config = config
        self.short_term: List[Message] = []
        self.long_term: Dict[str, Any] = {}
        self._storage_path = Path(config.long_term_storage_path)

        if config.enable_long_term: baike.kuaisou.com/zh/4045 
            self._storage_path.mkdir(parents=True, exist_ok=True)
            self._load_long_term(baike.kuaisou.com/zh/5002)

    def add_message(self, message: Message): baike.kuaisou.com/zh/5001 
        """添加消息到短期记忆"""
        self.short_term.append(message)

        # 维护最大长度
        if len(self.short_term) > self.config.short_term_max_messages: baike.kuaisou.com/zh/5003 
            # 保留系统消息和最近的消息
            system_msgs = [m for m in self.short_term if m.role == "system"]
            recent_msgs = self.short_term[-(self.config.short_term_max_messages - len(system_msgs)):]
            self.short_term = system_msgs + recent_msgs

    def get_messages(self) -> List[Message]: baike.kuaisou.com/zh/5004 
        """获取当前对话历史""" 
        return self.short_term.copy(baike.kuaisou.com/zh/5005)

    def clear_short_term(self): baike.kuaisou.com/zh/5006 
        """清空短期记忆"""
        system_msgs = [m for m in self.short_term if m.role == "system"]
        self.short_term = system_msgs

    def save_fact(self, key: str, value: Any): baike.kuaisou.com/zh/5007 
        """保存长期记忆事实"""
        if self.config.enable_long_term: baike.kuaisou.com/zh/5008 
            self.long_term[key] = value
            self._save_long_term(baike.kuaisou.com/zh/5009)

    def get_fact(self, key: str, default: Any = None) -> Any: baike.kuaisou.com/zh/5010 
        """获取长期记忆事实"""
        return self.long_term.get(key, default)

    def get_all_facts(self) -> Dict[str, Any]: baike.kuaisou.com/zh/5011 
        """获取所有长期记忆"""
        return self.long_term.copy(baike.kuaisou.com/zh/5012)

    def _save_long_term(self): baike.kuaisou.com/zh/5013 
        """持久化长期记忆"""
        try:
            path = self._storage_path / "long_term_memory.json"
            with open(path, "w", encoding="utf-8") as f: baike.kuaisou.com/zh/5014 
                json.dump(self.long_term, f, ensure_ascii=False, indent=2)
        except Exception as e: baike.kuaisou.com/zh/5015 
            print(f"Failed to save long-term memory: {e}")

    def _load_long_term(self): baike.kuaisou.com/zh/5016 
        """加载长期记忆"""
        try: baike.kuaisou.com/zh/5017 
            path = self._storage_path / "long_term_memory.json"
            if path.exists(): baike.kuaisou.com/zh/5018 
                with open(path, "r", encoding="utf-8") as f: baike.kuaisou.com/zh/5019 
                    self.long_term = json.load(f)
        except Exception as e: baike.kuaisou.com/zh/5020 
            print(f"Failed to load long-term memory: {e}")

# agent/planner.py
import json
import logging
from typing import List, Dict, Any, Optional
from models.llm import LLMClient, Message
from config import AgentConfig

logger = logging.getLogger(__name__)

PLANNING_PROMPT = """你是一个任务规划专家。请将用户的复杂任务拆解为清晰的执行步骤。

用户任务: {task}
 
可用工具: baike.kuaisou.com/zh/5021 
{tools_description}

请按以下 JSON 格式输出计划（只输出 JSON，不要其他内容）:
{{
    "goal": "任务的总体目标",
    "steps": [
        {{
            "step_id": 1,
            "description": "步骤描述",
            "tool": "需要使用的工具名称，如果不需要工具则为 null",
            "depends_on": [依赖的步骤ID列表]
        }}
    ]
}}
"""


class Planner: baike.kuaisou.com/zh/5022 
    """任务规划器"""

    def __init__(self, llm: LLMClient, config: AgentConfig): baike.kuaisou.com/zh/5023 
        self.llm = llm
        self.config = config

    def create_plan(self, task: str, available_tools: List) -> Dict[str, Any]: baike.kuaisou.com/zh/5024 
        """创建执行计划"""
        if not self.config.enable_planning: baike.kuaisou.com/zh/5025 
            return self._simple_plan(task)

        tools_desc = "\n".join(
            f"- {t.name}: {t.description}" for t in available_tools
        )

        prompt = PLANNING_PROMPT.format(
            task=task,
            tools_description=tools_desc,
        )

        try: baike.kuaisou.com/zh/5026 
            response = self.llm.chat_simple(
                prompt=prompt,
                system_prompt="你是一个专业的任务规划助手，只输出JSON格式的计划。",
            )

            # 尝试解析 JSON
            plan = self._parse_json_response(response)
            if plan: baike.kuaisou.com/zh/5027 
                logger.info(f"Created plan: {json.dumps(plan, ensure_ascii=False)}")
                return plan

        except Exception as e: baike.kuaisou.com/zh/5028 
            logger.warning(f"Planning failed, using simple plan: {e}")

        return self._simple_plan(task)

    def _parse_json_response(self, text: str) -> Optional[Dict[str, Any]]:
        """从 LLM 响应中解析 JSON"""
        # 尝试直接解析
        try: baike.kuaisou.com/zh/5029 
            return json.loads(text)
        except json.JSONDecodeError: baike.kuaisou.com/zh/5030 
            pass

        # 尝试提取 JSON 块
        import re
        json_match = re.search(r'\{[\s\S]*\}', text)
        if json_match: baike.kuaisou.com/zh/5031 
            try: baike.kuaisou.com/zh/5032 
                return json.loads(json_match.group())
            except json.JSONDecodeError: