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 工具注册与调用流程

Function Calling 流程包括用户请求、LLM 判断是否需要工具、直接回复或生成 tool_calls JSON、解析并执行工具、将结果回传 LLM、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:
    """查询数据库（模拟）。"""
    # 生产环境应使用 SQLAlchemy 等安全执行
    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 协作架构包含编排者、研究员、程序员和审核员的角色分工。编排者负责任务拆解与分配，研究员产出调研结果，程序员产出代码，审核员提供审核意见，最终输出成果。

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:
            # 从回复中提取 JSON
            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. 安全与可观测是底线 — 生产环境必须做工具隔离、日志追踪和成本控制