AI Agent 技术：从原理到工程实现的完整解析

pythonimport json
from typing import Callable

class Tool:
    def __init__(self, name: str, description: str, func: Callable):
        self.name = name
        self.description = description
        self.func = func

class ReActAgent:
    def __init__(self, llm, tools: list[Tool], max_steps: int = 10):
        self.llm = llm
        self.tools = {t.name: t for t in tools}
        self.max_steps = max_steps
        self.memory = []

    def _build_prompt(self, task: str) -> str:
        tool_desc = "\n".join(
            f"- {t.name}: {t.description}" for t in self.tools.values()
        )
        return f"""你是一个智能体，请通过推理和工具调用完成任务。

可用工具：
{tool_desc}

任务：{task}

请按以下格式输出：
思考：你的推理过程
行动：工具名称
参数：JSON格式参数
"""

    def _parse_action(self, response: str):
        # 解析 LLM 输出中的行动和参数
        lines = response.strip().split("\n")
        action, params = None, {}
        for i, line in enumerate(lines):
            if line.startswith("行动："):
                action = line.replace("行动：", "").strip()
            if line.startswith("参数："):
                params = json.loads(line.replace("参数：", "").strip())
        return action, params

    def run(self, task: str) -> str:
        self.memory.append(f"任务：{task}")
        for step in range(self.max_steps):
            prompt = self._build_prompt(task)
            response = self.llm.generate(prompt + "\n".join(self.memory))
            self.memory.append(response)

            action, params = self._parse_action(response)
            if action == "完成":
                return params.get("结果", "")

            if action in self.tools:
                try:
                    result = self.tools[action].func(**params)
                    self.memory.append(f"观察：{result}")
                except Exception as e:
                    self.memory.append(f"观察：工具执行失败 - {str(e)}")
            else:
                self.memory.append(f"观察：未知工具 {action}")

        return "达到最大步数，任务未完成"

def search_web(query: str, top_k: int = 3) -> str:
    """搜索网络信息"""
    if not query or not isinstance(query, str):
        raise ValueError("查询内容不能为空")
    # 实际实现调用搜索 API
    results = [f"结果{i+1}: 关于'{query}'的信息" for i in range(top_k)]
    return "\n".join(results)

def read_file(path: str) -> str:
    """读取文件内容"""
    import os
    if not os.path.exists(path):
        raise FileNotFoundError(f"文件不存在: {path}")
    with open(path, "r", encoding="utf-8") as f:
        return f.read()

tools = [
    Tool("search_web", "搜索网络信息，参数: query(字符串), top_k(整数)", search_web),
    Tool("read_file", "读取本地文件，参数: path(字符串)", read_file),
]

class MemoryStore:
    def __init__(self, vector_db):
        self.vector_db = vector_db

    def add(self, content: str, metadata: dict = None):
        embedding = self._embed(content)
        self.vector_db.insert(embedding, content, metadata or {})

    def retrieve(self, query: str, top_k: int = 5) -> list[str]:
        embedding = self._embed(query)
        return self.vector_db.search(embedding, top_k)

    def _embed(self, text: str) -> list[float]:
        # 调用嵌入模型
        return [0.0] * 768