AI Agent 开发实战：从零构建一个完整的自主 Agent（含 Python 全栈实现）

from dataclasses import dataclass, field
from typing import Any, Callable
import json
import re
import subprocess
import os
from datetime import datetime


@dataclass
class ToolParameter:
    """工具参数定义"""
    name: str
    type: str  # "string", "number", "boolean", "array"
    description: str
    required: bool = True
    enum: list[str] | None = None


@dataclass
class Tool:
    """工具定义——Agent 可使用的能力单元"""
    name: str
    description: str
    parameters: list[ToolParameter]
    fn: Callable  # 实际执行函数
    category: str = "general"  # 工具分类

    def to_prompt_description(self) -> str:
        """生成 LLM 可读的工具描述"""
        params_desc = ", ".join(
            f"{p.name} ({p.type})" for p in self.parameters if p.required
        )
        return f"- {self.name}({params_desc}): {self.description}"

    def to_json_schema(self) -> dict:
        """生成 JSON Schema 格式（用于 function calling）"""
        properties = {}
        required = []
        for p in self.parameters:
            properties[p.name] = {
                "type": p.type,
                "description": p.description,
            }
            if p.enum:
                properties[p.name]["enum"] = p.enum
            if p.required:
                required.append(p.name)

        return {
            "name": self.name,
            "description": self.description,
            "parameters": {
                "type": "object",
                "properties": properties,
                "required": required,
            }
        }


# ===== 预定义工具集 =====

def search_web(query: str) -> str:
    """模拟网络搜索（实际可接入 DuckDuckGo API、Tavily 等）"""
    # 实际项目中替换为真实搜索 API
    return f"[搜索结果] 关于 '{query}' 的信息：
"            f"1. 这是关于 {query} 的参考资料...
"            f"2. 相关文档显示 {query} 在 2026 年有重要发展...
"            f"3. 社区讨论表明 {query} 的最佳实践是..."


def execute_python(code: str) -> str:
    """安全执行 Python 代码（带超时和输出捕获）"""
    try:
        # 生产环境应使用沙箱执行（如 Docker、E2B）
        result = subprocess.run(
            ["python3", "-c", code],
            capture_output=True, text=True, timeout=30
        )
        if result.returncode == 0:
            return f"执行成功
输出:
{result.stdout}"
        else:
            return f"执行失败
错误:
{result.stderr}"
    except subprocess.TimeoutExpired:
        return "执行超时（超过 30 秒）"
    except Exception as e:
        return f"执行异常: {str(e)}"


def read_file(filepath: str) -> str:
    """读取文件内容"""
    try:
        if not os.path.exists(filepath):
            return f"错误：文件 '{filepath}' 不存在"
        with open(filepath, 'r', encoding='utf-8') as f:
            content = f.read()
        return f"文件内容（{len(content)} 字符）:
{content[:2000]}"
    except Exception as e:
        return f"读取失败: {str(e)}"


def write_file(filepath: str, content: str) -> str:
    """写入文件"""
    try:
        os.makedirs(os.path.dirname(filepath) or '.', exist_ok=True)
        with open(filepath, 'w', encoding='utf-8') as f:
            f.write(content)
        return f"成功写入 {filepath}（{len(content)} 字符）"
    except Exception as e:
        return f"写入失败: {str(e)}"


# 注册工具列表
BUILTIN_TOOLS = [
    Tool(
        name="search_web",
        description="搜索互联网获取最新信息。适用于需要实时数据或知识库之外的信息。",
        parameters=[ToolParameter("query", "string", "搜索关键词", required=True)],
        fn=search_web,
        category="search",
    ),
    Tool(
        name="execute_python",
        description="执行 Python 代码。用于数据分析、计算、文本处理等编程任务。",
        parameters=[ToolParameter("code", "string", "要执行的 Python 代码", required=True)],
        fn=execute_python,
        category="code",
    ),
    Tool(
        name="read_file",
        description="读取本地文件内容。支持文本文件。",
        parameters=[ToolParameter("filepath", "string", "文件路径（绝对或相对路径）", required=True)],
        fn=read_file,
        category="file",
    ),
    Tool(
        name="write_file",
        description="写入内容到文件。可创建新文件或覆盖已有文件。",
        parameters=[
            ToolParameter("filepath", "string", "文件路径", required=True),
            ToolParameter("content", "string", "要写入的内容", required=True),
        ],
        fn=write_file,
        category="file",
    ),
]

import json
import os
from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime


@dataclass
class MemoryEntry:
    """单条记忆"""
    id: str
    summary: str        # 经验摘要
    keywords: list[str]  # 检索关键词
    tool_usage: str      # 使用的工具（JSON）
    outcome: str         # 结果（success/failed）
    timestamp: str
    importance: float = 1.0  # 重要度（影响检索权重）


class AgentMemory:
    """Agent 记忆系统：短期记忆 + 长期记忆"""

    def __init__(self, storage_path: str = ".agent_memory"):
        self.storage_path = storage_path
        self.short_term: list[dict] = []  # 短期记忆（对话历史）
        self.long_term: list[MemoryEntry] = []  # 长期记忆（经验）

        os.makedirs(storage_path, exist_ok=True)
        self._load_long_term()

    # === 短期记忆 ===

    def add_turn(self, role: str, content: str):
        """添加一轮对话到短期记忆"""
        self.short_term.append({
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat(),
        })
        # 限制短期记忆大小，避免超出 LLM 上下文窗口
        if len(self.short_term) > 20:
            self.short_term = self.short_term[-20:]

    def get_context(self) -> str:
        """获取当前对话上下文（格式化为文本）"""
        lines = []
        for turn in self.short_term:
            role_label = "User" if turn["role"] == "user" else "Assistant"
            lines.append(f"{role_label}: {turn['content']}")
        return "\n".join(lines)

    def clear_short_term(self):
        """清理短期记忆（新任务开始时调用）"""
        self.short_term = []

    # === 长期记忆 ===

    def store_experience(
        self,
        task_summary: str,
        tools_used: list[str],
        outcome: str,
        importance: float = 1.0,
    ):
        """存储经验到长期记忆"""
        entry = MemoryEntry(
            id=f"exp_{len(self.long_term) + 1:04d}",
            summary=task_summary,
            keywords=self._extract_keywords(task_summary),
            tool_usage=json.dumps(tools_used),
            outcome=outcome,
            timestamp=datetime.now().isoformat(),
            importance=importance,
        )
        self.long_term.append(entry)
        self._save_long_term()
        return entry

    def retrieve_relevant(self, query: str, top_k: int = 3) -> list[MemoryEntry]:
        """检索与查询相关的经验（基于关键词匹配）"""
        query_keywords = self._extract_keywords(query)
        scored = []
        for entry in self.long_term:
            # 计算关键词重叠分数
            overlap = len(set(query_keywords) & set(entry.keywords))
            score = overlap * entry.importance
            if score > 0:
                scored.append((score, entry))

        scored.sort(key=lambda x: x[0], reverse=True)
        return [entry for _, entry in scored[:top_k]]

    def get_summary_prompt(self, query: str) -> str:
        """格式化为 LLM 可读的经验摘要"""
        relevant = self.retrieve_relevant(query)
        if not relevant:
            return "（无相关历史经验）"

        lines = ["## 相关历史经验"]
        for exp in relevant:
            lines.append(
                f"- [{exp.outcome}] {exp.summary} "
                f"(工具: {exp.tool_usage}, 时间: {exp.timestamp[:10]})"
            )
        return "\n".join(lines)

    # === 内部方法 ===

    def _extract_keywords(self, text: str) -> list[str]:
        """从文本中提取关键词（简化版）"""
        # 移除停用词，提取有意义的词
        stop_words = {
            "的", "了", "在", "是", "我", "有", "和", "就", "不", "人",
            "都", "一", "一个", "上", "也", "很", "到", "说", "要",
            "a", "an", "the", "is", "in", "on", "at", "to", "for",
            "with", "by", "from", "or", "and", "of", "it", "this", "that",
        }
        words = re.findall(r'[\w\u4e00-\u9fff]+', text.lower())
        return [w for w in words if w not in stop_words and len(w) > 1]

    def _save_long_term(self):
        """持久化长期记忆到文件"""
        data = [
            {
                "id": e.id, "summary": e.summary,
                "keywords": e.keywords, "tool_usage": e.tool_usage,
                "outcome": e.outcome, "timestamp": e.timestamp,
                "importance": e.importance,
            }
            for e in self.long_term
        ]
        path = os.path.join(self.storage_path, "long_term.json")
        with open(path, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=2)

    def _load_long_term(self):
        """从文件加载长期记忆"""
        path = os.path.join(self.storage_path, "long_term.json")
        if os.path.exists(path):
            with open(path, 'r', encoding='utf-8') as f:
                data = json.load(f)
            self.long_term = [MemoryEntry(**item) for item in data]

import json
import re
import time
from typing import Optional

# 假设已导入 AgentMemory, Tool 等（见上文）

SYSTEM_PROMPT = """你是一个自主 AI Agent。你可以使用工具来完成任务。

## 可用工具
{tools}

## 历史经验
{experience}

## 输出格式
你必须严格按以下格式输出：

Thought: [你的思考，分析当前情况并决定下一步]
Action: [工具名称]
Action Input: [工具的输入参数，JSON 格式]

或者当你有最终答案时：

Thought: [你的思考]
Final Answer: [你的最终答案]

## 规则
1. 每次只调用一个工具
2. Action 必须是上述工具之一
3. Action Input 必须是有效的 JSON
4. 如果你已经有足够信息回答问题，使用 Final Answer
5. 最多进行 {max_steps} 步推理，超过则直接给出 Final Answer
"""


class SimpleAgent:
    """自主 AI Agent —— ReAct 推理循环实现"""

    def __init__(
        self,
        tools: list[Tool],
        memory: AgentMemory,
        llm_client,  # LLM API 客户端（支持 OpenAI / Anthropic 等）
        model: str = "gpt-4o",
        max_steps: int = 10,
    ):
        self.tools = {t.name: t for t in tools}
        self.memory = memory
        self.llm_client = llm_client
        self.model = model
        self.max_steps = max_steps

    def run(self, task: str) -> str:
        """运行 Agent——完成一个任务"""
        # 清理短期记忆（新任务）
        self.memory.clear_short_term()

        # 检索相关经验
        experience_prompt = self.memory.get_summary_prompt(task)

        # 构造系统提示
        tools_desc = "\n".join(t.to_prompt_description() for t in self.tools.values())
        system_prompt = SYSTEM_PROMPT.format(
            tools=tools_desc,
            experience=experience_prompt,
            max_steps=self.max_steps,
        )

        # 记录用户输入
        self.memory.add_turn("user", task)

        # 执行 ReAct 循环
        step = 0
        conversation = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": task},
        ]

        print(f"\n{'='*60}")
        print(f"🚀 开始执行任务: {task}")
        print(f"{'='*60}\n")

        while step < self.max_steps:
            step += 1
            print(f"--- Step {step}/{self.max_steps} ---")

            # 调用 LLM
            response = self._call_llm(conversation)
            print(f"🤔 Thought: {response.thought}")

            # 检查是否有最终答案
            if response.final_answer:
                print(f"\n✅ Final Answer: {response.final_answer}")
                self.memory.add_turn("assistant", response.final_answer)
                # 存储经验
                self.memory.store_experience(
                    task_summary=task[:100],
                    tools_used=[t for t in response.tools_used],
                    outcome="success",
                )
                return response.final_answer

            # 调用工具
            if response.action and response.action_input:
                tool_name = response.action
                tool_input = response.action_input

                if tool_name in self.tools:
                    print(f"🔧 Action: {tool_name}({json.dumps(tool_input, ensure_ascii=False)})")
                    result = self._execute_tool(tool_name, tool_input)
                    print(f"📋 Observation: {result[:200]}...")

                    # 将工具调用加入对话
                    conversation.append({
                        "role": "assistant",
                        "content": f"Thought: {response.thought}\nAction: {tool_name}\nAction Input: {json.dumps(tool_input)}"
                    })
                    conversation.append({
                        "role": "user",
                        "content": f"Observation: {result}"
                    })
                else:
                    error_msg = f"错误：未知工具 '{tool_name}'"
                    print(f"❌ {error_msg}")
                    conversation.append({
                        "role": "user",
                        "content": f"Observation: {error_msg}"
                    })

        # 超过最大步数
        fallback = "已达到最大推理步数限制。基于已有信息，我的最佳回答是：需要更多信息或更多工具来完成此任务。"
        print(f"\n⏰ 达到最大步数限制")
        print(f"📝 {fallback}")
        self.memory.add_turn("assistant", fallback)
        return fallback

    def _call_llm(self, conversation: list[dict]) -> 'AgentResponse':
        """调用 LLM 并解析响应"""
        messages = conversation.copy()

        # 调用 LLM API（这里用伪代码，实际接入 OpenAI/Anthropic API）
        # response = self.llm_client.chat.completions.create(
        #     model=self.model,
        #     messages=messages,
        #     temperature=0.1,
        #     max_tokens=1024,
        # )
        # raw_text = response.choices[0].message.content

        # 为演示目的，这里展示解析逻辑
        raw_text = self._mock_llm_call(messages)  # 替换为真实 API 调用
        return AgentResponse.parse(raw_text)

    def _execute_tool(self, name: str, input_dict: dict) -> str:
        """执行工具并返回结果"""
        tool = self.tools[name]
        try:
            result = tool.fn(**input_dict)
            return str(result)
        except Exception as e:
            return f"工具执行错误: {str(e)}"

    def _mock_llm_call(self, messages: list[dict]) -> str:
        """模拟 LLM 调用（实际项目中替换为真实 API）"""
        # 这只是一个示例，实际应调用 OpenAI / Anthropic / Claude API
        last_msg = messages[-1]["content"]
        if "数据集" in last_msg or "分析" in last_msg:
            return (
                "Thought: 用户需要分析数据，我需要先读取文件。\n"
                "Action: read_file\n"
                "Action Input: {\"filepath\": \"data.csv\"}"
            )
        return (
            "Thought: 我已经有了足够的信息来回答这个问题。\n"
            "Final Answer: 根据搜索结果，这个问题的答案是..."
        )


@dataclass
class AgentResponse:
    """Agent 的 LLM 输出解析结果"""
    thought: str
    action: Optional[str] = None
    action_input: Optional[dict] = None
    final_answer: Optional[str] = None
    tools_used: list[str] = field(default_factory=list)

    @classmethod
    def parse(cls, text: str) -> 'AgentResponse':
        """从 LLM 输出文本中解析 Thought/Action/Final Answer"""
        thought_match = re.search(r'Thought:\s*(.+?)(?=\nAction:|\nFinal Answer:|$)', text, re.DOTALL)
        action_match = re.search(r'Action:\s*(\w+)', text)
        action_input_match = re.search(r'Action Input:\s*({.*?})', text, re.DOTALL)
        final_match = re.search(r'Final Answer:\s*(.+)$', text, re.DOTALL)

        thought = thought_match.group(1).strip() if thought_match else ""
        final_answer = final_match.group(1).strip() if final_match else None

        action = action_match.group(1) if action_match else None
        action_input = None
        if action_input_match:
            try:
                action_input = json.loads(action_input_match.group(1))
            except json.JSONDecodeError:
                pass

        tools_used = [action] if action else []

        return cls(
            thought=thought,
            action=action,
            action_input=action_input,
            final_answer=final_answer,
            tools_used=tools_used,
        )


# ===== 使用示例 =====

if __name__ == "__main__":
    # 初始化组件
    memory = AgentMemory(storage_path="./my_agent_memory")
    tools = BUILTIN_TOOLS

    # 初始化 LLM 客户端（伪代码，实际接入真实 API）
    # from openai import OpenAI
    # llm_client = OpenAI(api_key="your-api-key")
    llm_client = None  # 替换为真实客户端

    # 创建 Agent
    agent = SimpleAgent(
        tools=tools,
        memory=memory,
        llm_client=llm_client,
        model="gpt-4o",
        max_steps=8,
    )

    # 运行任务
    result = agent.run("请读取 data.csv 文件并分析其中的数据趋势")
    print(f"\n最终结果: {result}")

class RobustAgent(SimpleAgent):
    """增强版 Agent——带错误恢复机制"""

    def __init__(self, *args, retry_count: int = 2, **kwargs):
        super().__init__(*args, **kwargs)
        self.retry_count = retry_count
        self.error_history: list[str] = []  # 记录遇到的错误

    def _execute_tool(self, name: str, input_dict: dict) -> str:
        """执行工具，带重试和错误记录"""
        for attempt in range(1, self.retry_count + 1):
            try:
                tool = self.tools[name]
                result = tool.fn(**input_dict)

                # 检查结果是否合理
                if self._is_error_result(str(result)):
                    error_info = f"工具 {name} 返回了错误信息: {result}"
                    self.error_history.append(error_info)

                    if attempt < self.retry_count:
                        print(f"  ⚠️  第 {attempt} 次尝试失败，重试中...")
                        time.sleep(1)
                        continue

                return str(result)

            except Exception as e:
                error_msg = f"工具 {name} 异常: {str(e)}"
                self.error_history.append(error_msg)
                if attempt < self.retry_count:
                    print(f"  ⚠️  第 {attempt} 次异常，重试中...")
                    time.sleep(1)
                    continue
                return f"❌ {error_msg}（已重试 {self.retry_count} 次）"

        return f"❌ 工具 {name} 所有重试均失败"

    def _is_error_result(self, result: str) -> bool:
        """判断工具返回结果是否表示错误"""
        error_indicators = ["错误", "失败", "Error", "Failed", "不存在", "Exception"]
        return any(indicator in result for indicator in error_indicators)

    def run(self, task: str) -> str:
        """重写 run 方法，加入错误恢复逻辑"""
        self.error_history = []
        result = super().run(task)

        # 任务结束后，分析错误历史并生成改进建议
        if self.error_history:
            print(f"\n📊 错误报告:")
            for err in self.error_history:
                print(f"  - {err}")

            # 将错误经验存储为"反面教材"
            self.memory.store_experience(
                task_summary=f"[失败经验] {task[:80]}",
                tools_used=["error_recovery"],
                outcome="failed",
                importance=0.8,
            )

        return result


# ===== 对比测试 =====

def compare_agents():
    """对比基础 Agent 和增强版 Agent"""
    memory = AgentMemory()
    tools = BUILTIN_TOOLS

    # 基础版
    basic = SimpleAgent(tools, memory, llm_client=None, max_steps=5)

    # 增强版
    robust = RobustAgent(
        tools, memory, llm_client=None,
        max_steps=5, retry_count=3
    )

    test_tasks = [
        "读取一个不存在的文件",
        "执行一段有语法错误的 Python 代码",
        "搜索 2026 年最新的 AI Agent 项目",
    ]

    print("\n" + "=" * 60)
    print("📋 Agent 错误恢复能力测试")
    print("=" * 60)

    for task in test_tasks:
        print(f"\n📌 任务: {task}")
        print("-" * 40)

        print("\n【基础版 Agent】")
        basic_result = basic.run(task)

        print("\n【增强版 Agent】")
        robust_result = robust.run(task)

    print("\n" + "=" * 60)
    print("测试完成！增强版 Agent 通过重试和错误恢复，")
    print("在面对工具失败时表现更稳定。")
    print("=" * 60)


if __name__ == "__main__":
    compare_agents()