Python AI Agent 智能体构建指南：从原理到实战

运行示例输出：

--- 第 1 轮 ---
Thought: 用户需要查询两个城市的天气，我先查北京的。
Action: get_weather
Action Input: {"city": "北京"}
Observation: 晴，气温 18°C，空气质量良好
--- 第 2 轮 ---
Thought: 已获取北京天气，再查深圳。
Action: get_weather
Action Input: {"city": "深圳"}
Observation: 阵雨，气温 28°C，湿度 85%
--- 第 3 轮 ---
Thought: 北京 18°C，深圳 28°C，温差 = 28 - 18 = 10°C
Action: calculate
Action Input: {"expression": "28 - 18"}
Observation: 计算结果：10
--- 第 4 轮 ---
Thought: 已获取所有信息，可以回复了。
Final Answer: 北京：晴，18°C；深圳：阵雨，28°C。两城市温差为 10°C。
最终结果：北京：晴，18°C；深圳：阵雨，28°C。两城市温差为 10°C。

四、记忆系统：短期记忆 + 长期记忆

4.1 记忆架构

• 短期记忆：对话历史 / 滑动窗口
• 长期记忆：向量数据库 (ChromaDB)
• 流程：存储对话 → 提取关键事实 → 相似度检索 → 相关上下文注入

4.2 记忆模块实现

""" Agent 记忆系统：短期对话记忆 + 长期向量记忆 依赖：pip install chromadb sentence-transformers """
import hashlib
import json
from datetime import datetime
from typing import Optional
import chromadb
from sentence_transformers import SentenceTransformer

class ShortTermMemory:
    """短期记忆：维护最近 N 轮对话的滑动窗口。"""
    def __init__(self, max_rounds: int = 20):
        self.max_rounds = max_rounds
        self.history: list[dict] = []

    def add(self, role: str, content: str):
        self.history.append({"role": role, "content": content, "timestamp": datetime.now().isoformat()})
        # 超出窗口时保留系统提示 + 最近对话
        if len(self.history) > self.max_rounds:
            self.history = self.history[-self.max_rounds:]

    def get_messages(self) -> list[dict]:
        return [{"role": m["role"], "content": m["content"]} for m in self.history]

    def clear(self):
        self.history.clear()

class LongTermMemory:
    """长期记忆：基于向量数据库的语义检索。"""
    def __init__(self, collection_name: str = "agent_memory"):
        self.embedder = SentenceTransformer("BAAI/bge-small-zh-v1.5")
        self.client = chromadb.PersistentClient(path="./chroma_memory")
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            metadata={"hnsw:space": "cosine"},
        )

    def _embed(self, text: str) -> list[float]:
        return self.embedder.encode(text).tolist()

    def store(self, content: str, metadata: Optional[dict] = None):
        """存储一条记忆。"""
        doc_id = hashlib.md5(content.encode()).hexdigest()[:12]
        self.collection.upsert(
            ids=[doc_id],
            documents=[content],
            embeddings=[self._embed(content)],
            metadatas=[metadata or {"stored_at": datetime.now().isoformat()}],
        )

    def retrieve(self, query: str, top_k: int = 5) -> list[str]:
        """检索与查询最相关的记忆。"""
        results = self.collection.query(
            query_embeddings=[self._embed(query)],
            n_results=top_k,
        )
        return results["documents"][0] if results["documents"] else []

    def store_conversation_summary(self, messages: list[dict]):
        """将一轮完整对话提取为摘要后存储。"""
        conversation = json.dumps([{"role": m["role"], "content": m["content"][:200]} for m in messages], ensure_ascii=False)
        self.store(
            content=conversation,
            metadata={"type": "conversation", "rounds": len(messages), "stored_at": datetime.now().isoformat()},
        )

# ========== 组合使用 ==========
class AgentMemory:
    """Agent 统一记忆管理器。"""
    def __init__(self):
        self.short_term = ShortTermMemory(max_rounds=20)
        self.long_term = LongTermMemory()

    def build_context(self, user_input: str) -> list[dict]:
        """构建注入给 LLM 的完整上下文。"""
        # 从长期记忆中检索相关内容
        relevant_memories = self.long_term.retrieve(user_input, top_k=3)
        context = []
        # 注入长期记忆作为系统上下文
        if relevant_memories:
            memory_text = "\n---\n".join(relevant_memories)
            context.append({"role": "system", "content": f"以下是与当前对话相关的历史记忆:\n{memory_text}"})
        # 注入短期对话历史
        context.extend(self.short_term.get_messages())
        return context

    def save_turn(self, role: str, content: str):
        """保存一轮对话到短期记忆。"""
        self.short_term.add(role, content)

    def consolidate(self):
        """将短期记忆中的重要信息转入长期记忆。"""
        messages = self.short_term.get_messages()
        if len(messages) >= 10:
            self.long_term.store_conversation_summary(messages)
            self.short_term.clear()
            print("[Memory] 短期记忆已整合到长期记忆")

五、Function Calling：结构化工具调用

5.1 工具注册与调用流程

1. 用户请求
2. LLM 判断是否需要工具
   • 不需要：直接回复
   • 需要：生成 tool_calls JSON
3. 解析并执行工具
4. 将结果回传 LLM
5. LLM 生成最终回复

5.2 基于 OpenAI Function Calling 的工具系统

""" 结构化工具调用系统 使用 OpenAI Function Calling 协议，兼容 vLLM / Ollama / 任何兼容 API """
import json
from typing import Any, Callable
from openai import OpenAI

client = OpenAI(base_url="http://localhost:8000/v1", api_key="not-needed")

# ========== 工具注册 ==========
class ToolRegistry:
    """工具注册中心，自动生成 OpenAI Function Calling schema。"""
    def __init__(self):
        self._tools: dict[str, dict] = {}
        self._handlers: dict[str, Callable] = {}

    def register(self, name: str, description: str, parameters: dict):
        """装饰器：注册一个工具函数。"""
        def decorator(func: Callable):
            self._tools[name] = {
                "type": "function",
                "function": {
                    "name": name,
                    "description": description,
                    "parameters": parameters,
                },
            }
            self._handlers[name] = func
            return func
        return decorator

    def get_schemas(self) -> list[dict]:
        return list(self._tools.values())

    def execute(self, name: str, arguments: dict) -> str:
        handler = self._handlers.get(name)
        if not handler:
            return json.dumps({"error": f"未知工具：{name}"}, ensure_ascii=False)
        try:
            result = handler(**arguments)
            return json.dumps(result, ensure_ascii=False) if not isinstance(result, str) else result
        except Exception as e:
            return json.dumps({"error": str(e)}, ensure_ascii=False)

registry = ToolRegistry()

# ========== 注册具体工具 ==========
@registry.register(
    name="query_database",
    description="查询 SQL 数据库，返回查询结果",
    parameters={
        "type": "object",
        "properties": {
            "sql": {"type": "string", "description": "SQL 查询语句（仅支持 SELECT）"},
            "database": {"type": "string", "description": "数据库名称", "enum": ["orders", "users", "products"]},
        },
        "required": ["sql", "database"],
    },
)
def query_database(sql: str, database: str) -> str:
    """查询数据库（模拟）。"""
    if not sql.strip().upper().startswith("SELECT"):
        return "错误：仅允许 SELECT 查询"
    return json.dumps({
        "columns": ["id", "name", "amount"],
        "rows": [[1, "订单 A", 299.00], [2, "订单 B", 1599.00], [3, "订单 C", 89.50]],
        "total": 3,
    }, ensure_ascii=False)

@registry.register(
    name="send_email",
    description="发送电子邮件",
    parameters={
        "type": "object",
        "properties": {
            "to": {"type": "string", "description": "收件人邮箱"},
            "subject": {"type": "string", "description": "邮件主题"},
            "body": {"type": "string", "description": "邮件正文"},
        },
        "required": ["to", "subject", "body"],
    },
)
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件（模拟）。"""
    return f"邮件已发送至 {to}，主题：{subject}"

@registry.register(
    name="read_file",
    description="读取指定文件的内容",
    parameters={
        "type": "object",
        "properties": {
            "path": {"type": "string", "description": "文件路径"},
            "encoding": {"type": "string", "description": "文件编码", "default": "utf-8"},
        },
        "required": ["path"],
    },
)
def read_file(path: str, encoding: str = "utf-8") -> str:
    """读取文件内容。"""
    try:
        with open(path, "r", encoding=encoding) as f:
            content = f.read(5000)
            return content
    except FileNotFoundError:
        return f"错误：文件不存在 - {path}"
    except Exception as e:
        return f"错误：{e}"

# ========== Agent 运行 ==========
def run_agent(user_query: str, max_rounds: int = 5) -> str:
    """运行基于 Function Calling 的 Agent。"""
    messages = [
        {"role": "system", "content": "你是一个智能助手，可以使用工具帮助用户完成任务。请使用中文回复。"},
        {"role": "user", "content": user_query},
    ]
    for round_num in range(max_rounds):
        response = client.chat.completions.create(
            model="qwen-72b",
            messages=messages,
            tools=registry.get_schemas(),
            tool_choice="auto",
            temperature=0.3,
        )
        msg = response.choices[0].message
        # 没有工具调用 → 直接回复
        if not msg.tool_calls:
            return msg.content
        # 有工具调用 → 逐个执行
        messages.append(msg)
        for tool_call in msg.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)
            print(f" [调用工具] {func_name}({json.dumps(func_args, ensure_ascii=False)})")
            result = registry.execute(func_name, func_args)
            print(f" [执行结果] {result[:200]}")
            messages.append({"role": "tool", "tool_call_id": tool_call.id, "content": result})
    return "Agent 达到最大调用轮数。"

if __name__ == "__main__":
    result = run_agent("帮我查一下 orders 数据库中最近的订单数据，然后把结果发送到 [email protected]，邮件主题写'本周订单汇总'。")
    print(f"\n最终回复:\n{result}")

六、多 Agent 协作

6.1 多 Agent 协作架构

• 编排 Agent：任务拆解与分配
• 研究员 Agent：调研结果
• 程序员 Agent：代码产出
• 审核员 Agent：审核意见
• 流程：需要修改 → 通过审核 → 最终输出

6.2 多 Agent 实现

""" 多 Agent 协作系统：研究员 + 程序员 + 审核员 """
import json
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
from openai import OpenAI

client = OpenAI(base_url="http://localhost:8000/v1", api_key="not-needed")

class AgentRole(Enum):
    ORCHESTRATOR = "orchestrator"
    RESEARCHER = "researcher"
    CODER = "coder"
    REVIEWER = "reviewer"

@dataclass
class Agent:
    name: str
    role: AgentRole
    system_prompt: str
    model: str = "qwen-72b"
    history: list[dict] = field(default_factory=list)

    def chat(self, message: str) -> str:
        self.history.append({"role": "user", "content": message})
        messages = [{"role": "system", "content": self.system_prompt}] + self.history
        response = client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=0.3,
            max_tokens=2048,
        )
        reply = response.choices[0].message.content
        self.history.append({"role": "assistant", "content": reply})
        return reply

# ========== 创建 Agent ==========
def create_agents() -> dict[str, Agent]:
    return {
        "orchestrator": Agent(
            name="编排者",
            role=AgentRole.ORCHESTRATOR,
            system_prompt="""你是任务编排者。负责：
1. 将用户任务拆解为子任务
2. 分配给合适的 Agent 执行
3. 汇总结果并判断是否需要修改
4. 以 JSON 格式输出指令：{"action": "assign|finish|revise", "agent": "目标 agent", "task": "任务描述", "summary": "当前进度摘要"}
可用 Agent: researcher(调研), coder(编程), reviewer(审核)
任务完成后 action 设为 "finish" 并在 task 中给出最终结果。""",
        ),
        "researcher": Agent(
            name="研究员",
            role=AgentRole.RESEARCHER,
            system_prompt="你是一位技术研究员，负责对给定主题进行深入调研，给出技术方案、最佳实践和注意事项。回复要结构清晰、有理有据。",
        ),
        "coder": Agent(
            name="程序员",
            role=AgentRole.CODER,
            system_prompt="你是一位资深 Python 工程师。根据需求编写高质量代码，包含完整的类型注解、错误处理和文档字符串。只输出代码和必要的说明。",
        ),
        "reviewer": Agent(
            name="审核员",
            role=AgentRole.REVIEWER,
            system_prompt="""你是代码审核专家。审核代码时关注：
1. 正确性：逻辑是否正确
2. 安全性：是否存在安全漏洞
3. 性能：是否有明显性能问题
4. 可读性：命名、结构是否清晰
回复格式：
- status: "approved" 或 "needs_revision"
- issues: 问题列表（如有）
- suggestions: 改进建议""",
        ),
    }

# ========== 协作流程 ==========
def run_multi_agent(task: str, max_rounds: int = 12) -> str:
    agents = create_agents()
    orchestrator = agents["orchestrator"]
    current_task = task
    task_history = []
    for i in range(max_rounds):
        print(f"\n{'='*60}")
        print(f"第 {i + 1} 轮")
        print(f"{'='*60}")
        # 编排者决策
        context = f"原始任务：{task}\n\n执行历史:\n" + "\n".join(task_history) if i > 0 else f"原始任务：{task}"
        decision_text = orchestrator.chat(context)
        print(f"[编排者] {decision_text[:300]}")
        # 解析决策
        try:
            import re
            json_match = re.search(r'\{.*\}', decision_text, re.DOTALL)
            if not json_match:
                continue
            decision = json.loads(json_match.group())
        except (json.JSONDecodeError, AttributeError):
            continue
        action = decision.get("action", "")
        target_agent = decision.get("agent", "")
        sub_task = decision.get("task", "")
        if action == "finish":
            return sub_task
        if action in ("assign", "revise") and target_agent in agents:
            agent = agents[target_agent]
            result = agent.chat(sub_task)
            print(f"[{agent.name}] {result[:300]}")
            task_history.append(f"- {agent.name} 执行：{sub_task[:100]}... → 结果：{result[:200]}...")
            current_task = result
    return "多 Agent 协作达到最大轮数。"

# ========== 运行 ==========
if __name__ == "__main__":
    task = "开发一个 Python 爬虫，爬取某新闻网站的最新科技新闻标题和摘要，保存为 CSV 文件"
    result = run_multi_agent(task)
    print(f"\n{'='*60}")
    print(f"最终结果:\n{result}")

七、完整实战：带记忆和工具的智能助手

""" 完整 AI Agent：集成记忆 + 工具调用 + ReAct 循环 """
import json
import re
from datetime import datetime
from openai import OpenAI

client = OpenAI(base_url="http://localhost:8000/v1", api_key="not-needed")

class SmartAgent:
    """集记忆、工具、多轮对话于一体的智能 Agent。"""
    def __init__(self, name: str = "AI 助手"):
        self.name = name
        self.memory = AgentMemory()
        self.registry = self._setup_tools()

    def _setup_tools(self) -> dict:
        """注册可用工具。"""
        return {
            "search": {"handler": self._tool_search, "description": "搜索网络信息", "params": {"query": "str"}},
            "calculate": {"handler": self._tool_calculate, "description": "执行数学计算", "params": {"expression": "str"}},
            "get_time": {"handler": self._tool_get_time, "description": "获取当前时间", "params": {}},
            "save_note": {"handler": self._tool_save_note, "description": "保存笔记到记忆", "params": {"content": "str"}},
        }

    # --- 工具实现 ---
    def _tool_search(self, query: str) -> str:
        return f"搜索「{query}」的结果：...（模拟数据）"

    def _tool_calculate(self, expression: str) -> str:
        allowed = set("0123456789+-*/().% ")
        if all(c in allowed for c in expression):
            return str(eval(expression))
        return "错误：表达式不安全"

    def _tool_get_time(self) -> str:
        return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    def _tool_save_note(self, content: str) -> str:
        self.memory.long_term.store(content, metadata={"type": "note"})
        return f"已保存笔记（{len(content)} 字）"

    # --- 核心对话 ---
    def chat(self, user_input: str) -> str:
        # 构建上下文：长期记忆 + 短期记忆
        context = self.memory.build_context(user_input)
        context.append({"role": "user", "content": user_input})

        # 第一轮 LLM 调用（含工具定义）
        tools_schema = [
            {
                "type": "function",
                "function": {
                    "name": name,
                    "description": tool["description"],
                    "parameters": {
                        "type": "object",
                        "properties": {k: {"type": "string", "description": v} for k, v in tool["params"].items()},
                        "required": list(tool["params"].keys()),
                    },
                },
            }
            for name, tool in self.registry.items()
        ]
        response = client.chat.completions.create(
            model="qwen-72b",
            messages=context,
            tools=tools_schema,
            tool_choice="auto",
            temperature=0.5,
        )
        msg = response.choices[0].message

        # 处理工具调用
        if msg.tool_calls:
            context.append(msg)
            for tc in msg.tool_calls:
                tool_name = tc.function.name
                tool_args = json.loads(tc.function.arguments)
                tool_result = self.registry[tool_name]["handler"](**tool_args)
                context.append({"role": "tool", "tool_call_id": tc.id, "content": tool_result})

        # 第二轮调用，让 LLM 基于工具结果生成回复
        response = client.chat.completions.create(
            model="qwen-72b",
            messages=context,
            temperature=0.5,
        )
        msg = response.choices[0].message

        # 保存到记忆
        self.memory.save_turn("user", user_input)
        self.memory.save_turn("assistant", msg.content)
        return msg.content

if __name__ == "__main__":
    agent = SmartAgent(name="小智")
    # 多轮对话示例
    conversations = [
        "现在几点了？",
        "帮我搜索一下 Python 3.13 的新特性",
        "把刚才搜索到的内容保存为笔记",
        "我之前保存了什么笔记？",  # 会触发长期记忆检索
    ]
    for user_input in conversations:
        print(f"\n👤 用户：{user_input}")
        reply = agent.chat(user_input)
        print(f"🤖 {agent.name}: {reply}")

八、Agent 应用场景与性能对比

关键设计考量

九、总结

本文从零到一构建了一个完整的 AI Agent 系统，核心要点：

1. ReAct 是 Agent 的灵魂 — Think → Act → Observe 循环使 LLM 具备自主推理能力
2. 工具是 Agent 的双手 — Function Calling 提供了结构化、可靠的外部交互机制
3. 记忆是 Agent 的经验 — 短期记忆维持对话连贯，长期记忆实现知识积累
4. 多 Agent 是 Agent 的团队 — 角色分工、协作编排解决复杂任务
5. 安全与可观测是底线 — 生产环境必须做工具隔离、日志追踪和成本控制