Python + AI Agent 智能体：从原理到实战，构建自主决策的 AI 助手

AI Agent（智能体）是大模型落地应用的核心范式。与传统的"一问一答"不同，Agent 能够自主规划任务、调用外部工具、管理记忆上下文、甚至与其他 Agent 协作。本文将基于 Python 生态，从原理到实战，系统讲解如何构建一个生产级 AI Agent。

• 一、AI Agent 核心架构
  • 1.1 什么是 AI Agent？
  • 1.2 整体架构图
• 二、技术栈与生态
• 三、从零实现：最小可用 Agent
  • 3.1 ReAct 循环
  • 3.2 手写 ReAct Agent（不依赖框架）
• 四、记忆系统：短期记忆 + 长期记忆
  • 4.1 记忆架构
  • 4.2 记忆模块实现
• 五、Function Calling：结构化工具调用
  • 5.1 工具注册与调用流程
  • 5.2 基于 OpenAI Function Calling 的工具系统
• 六、多 Agent 协作
  • 6.1 多 Agent 协作架构
  • 6.2 多 Agent 实现
• 七、完整实战：带记忆和工具的智能助手
• 八、Agent 应用场景与性能对比
  • 关键设计考量
• 九、总结

一、AI Agent 核心架构

1.1 什么是 AI Agent？

AI Agent = LLM（大脑）+ Planning（规划）+ Tools（工具）+ Memory（记忆）

传统 LLM 调用: 用户 → Prompt → LLM → 文本回复 AI Agent 调用: 用户 → Agent → 规划 → [选工具 → 执行 → 观察] × N → 最终回复 

1.2 整体架构图

搜索

代码

数据库

文件

自定义

继续规划

任务完成

上下文注入

用户输入

任务规划器
ReAct / Plan-and-Execute

大语言模型
LLM

工具选择

Web Search API

Code Interpreter

SQL / API 调用

文件读写

自定义工具函数

执行结果观察

记忆模块

生成最终回复

二、技术栈与生态

35%20%12%10%10%8%5%2026 年 Python Agent 开发框架使用占比LangChain / LangGraphOpenAI Agents SDKCrewAIAutoGenDify (Python SDK)自研 / 裸调 API其他

三、从零实现：最小可用 Agent

3.1 ReAct 循环

Agent 的核心是 ReAct（Reasoning + Acting） 循环：

否

是

Thought: 思考下一步

Action: 选择并执行工具

Observation: 观察执行结果

任务是否完成?

Final Answer: 生成最终回复

3.2 手写 ReAct Agent（不依赖框架）

""" 最小 ReAct Agent 实现 依赖: pip install openai """import json import re from typing import Callable from openai import OpenAI client = OpenAI(base_url="http://localhost:8000/v1", api_key="not-needed")# ========== 工具定义 ==========defget_weather(city:str)->str:"""查询城市天气（模拟）。""" weather_db ={"北京":"晴，气温 18°C，空气质量良好","上海":"多云，气温 22°C，有轻微雾霾","深圳":"阵雨，气温 28°C，湿度 85%","成都":"阴，气温 16°C，预计下午转晴",}return weather_db.get(city,f"未找到 {city} 的天气数据")defsearch_web(query:str)->str:"""网络搜索（模拟）。"""returnf"搜索结果: 关于「{query}」，以下是最相关的信息...（模拟数据）"defcalculate(expression:str)->str:"""安全计算数学表达式。""" allowed =set("0123456789+-*/().% ")ifnotall(c in allowed for c in expression):return"错误: 表达式包含非法字符"try: result =eval(expression)# 仅允许数学字符returnf"计算结果: {result}"except Exception as e:returnf"计算错误: {e}"# ========== 工具注册表 ========== TOOLS ={"get_weather":{"func": get_weather,"description":"查询指定城市的天气情况","params":{"city":"城市名称，如：北京、上海"},},"search_web":{"func": search_web,"description":"在网络上搜索信息","params":{"query":"搜索关键词"},},"calculate":{"func": calculate,"description":"计算数学表达式","params":{"expression":"数学表达式，如：(25 + 37) * 2"},},} SYSTEM_PROMPT =f"""你是一个 AI Agent，可以使用工具来完成任务。 可用工具: {json.dumps({k:{"description": v["description"],"params": v["params"]}for k, v in TOOLS.items()}, ensure_ascii=False, indent=2)} 请严格按照以下格式回复: Thought: <分析当前情况，思考下一步> Action: <工具名称> Action Input: <JSON 格式参数> 或者当你认为任务已完成时: Thought: <总结分析> Final Answer: <最终答案> 注意: 每次只执行一个工具调用。"""# ========== Agent 主循环 ==========classReactAgent:def__init__(self, max_iterations:int=10): self.max_iterations = max_iterations defrun(self, user_query:str)->str: messages =[{"role":"system","content": SYSTEM_PROMPT},{"role":"user","content": user_query},]for i inrange(self.max_iterations): response = client.chat.completions.create( model="qwen-72b", messages=messages, temperature=0.3, max_tokens=1024,) reply = response.choices[0].message.content messages.append({"role":"assistant","content": reply})print(f"\n--- 第 {i +1} 轮 ---")print(reply)# 检查是否给出最终答案if"Final Answer:"in reply:return self._extract_final_answer(reply)# 解析工具调用 action, action_input = self._parse_action(reply)ifnot action:continue# 执行工具if action in TOOLS: tool_result = TOOLS[action]["func"](**action_input)else: tool_result =f"错误: 未知工具 '{action}'" observation =f"Observation: {tool_result}"print(observation) messages.append({"role":"user","content": observation})return"Agent 达到最大迭代次数，任务未完成。"def_parse_action(self, text:str)->tuple:"""从回复中解析 Action 和 Action Input。""" action_match = re.search(r"Action:\s*(\w+)", text) input_match = re.search(r"Action Input:\s*(\{.*?\})", text, re.DOTALL)ifnot action_match ornot input_match:returnNone,None action = action_match.group(1)try: action_input = json.loads(input_match.group(1))except json.JSONDecodeError: action_input ={}return action, action_input def_extract_final_answer(self, text:str)->str:match= re.search(r"Final Answer:\s*(.*)", text, re.DOTALL)returnmatch.group(1).strip()ifmatchelse text # ========== 运行 ==========if __name__ =="__main__": agent = ReactAgent(max_iterations=10) result = agent.run("帮我查一下北京和深圳的天气，然后算一下两个城市的温差是多少。")print(f"\n{'='*50}")print(f"最终结果: {result}")

运行示例输出：

--- 第 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 classShortTermMemory:"""短期记忆：维护最近 N 轮对话的滑动窗口。"""def__init__(self, max_rounds:int=20): self.max_rounds = max_rounds self.history:list[dict]=[]defadd(self, role:str, content:str): self.history.append({"role": role,"content": content,"timestamp": datetime.now().isoformat(),})# 超出窗口时保留系统提示 + 最近对话iflen(self.history)> self.max_rounds: self.history = self.history[-self.max_rounds:]defget_messages(self)->list[dict]:return[{"role": m["role"],"content": m["content"]}for m in self.history]defclear(self): self.history.clear()classLongTermMemory:"""长期记忆：基于向量数据库的语义检索。"""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()defstore(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()}],)defretrieve(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[]defstore_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(),},)# ========== 组合使用 ==========classAgentMemory:"""Agent 统一记忆管理器。"""def__init__(self): self.short_term = ShortTermMemory(max_rounds=20) self.long_term = LongTermMemory()defbuild_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 defsave_turn(self, role:str, content:str):"""保存一轮对话到短期记忆。""" self.short_term.add(role, content)defconsolidate(self):"""将短期记忆中的重要信息转入长期记忆。""" messages = self.short_term.get_messages()iflen(messages)>=10: self.long_term.store_conversation_summary(messages) self.short_term.clear()print("[Memory] 短期记忆已整合到长期记忆")

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

5.1 工具注册与调用流程

不需要

需要

用户请求

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")# ========== 工具注册 ==========classToolRegistry:"""工具注册中心，自动生成 OpenAI Function Calling schema。"""def__init__(self): self._tools:dict[str,dict]={} self._handlers:dict[str, Callable]={}defregister(self, name:str, description:str, parameters:dict):"""装饰器：注册一个工具函数。"""defdecorator(func: Callable): self._tools[name]={"type":"function","function":{"name": name,"description": description,"parameters": parameters,},} self._handlers[name]= func return func return decorator defget_schemas(self)->list[dict]:returnlist(self._tools.values())defexecute(self, name:str, arguments:dict)->str: handler = self._handlers.get(name)ifnot handler:return json.dumps({"error":f"未知工具: {name}"}, ensure_ascii=False)try: result = handler(**arguments)return json.dumps(result, ensure_ascii=False)ifnotisinstance(result,str)else result except Exception as e:return json.dumps({"error":str(e)}, ensure_ascii=False) registry = ToolRegistry()# ========== 注册具体工具 [email protected]( 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"],},)defquery_database(sql:str, database:str)->str:"""查询数据库（模拟）。"""# 生产环境应使用 SQLAlchemy 等安全执行ifnot 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"],},)defsend_email(to:str, subject:str, body:str)->str:"""发送邮件（模拟）。"""returnf"邮件已发送至 {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"],},)defread_file(path:str, encoding:str="utf-8")->str:"""读取文件内容。"""try:withopen(path,"r", encoding=encoding)as f: content = f.read(5000)# 限制读取长度return content except FileNotFoundError:returnf"错误: 文件不存在 - {path}"except Exception as e:returnf"错误: {e}"# ========== Agent 运行 ==========defrun_agent(user_query:str, max_rounds:int=5)->str:"""运行基于 Function Calling 的 Agent。""" messages =[{"role":"system","content":"你是一个智能助手，可以使用工具帮助用户完成任务。请使用中文回复。",},{"role":"user","content": user_query},]for round_num inrange(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 # 没有工具调用 → 直接回复ifnot 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")classAgentRole(Enum): ORCHESTRATOR ="orchestrator" RESEARCHER ="researcher" CODER ="coder" REVIEWER ="reviewer"@dataclassclassAgent: name:str role: AgentRole system_prompt:str model:str="qwen-72b" history:list[dict]= field(default_factory=list)defchat(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 ==========defcreate_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: 改进建议""",),}# ========== 协作流程 ==========defrun_multi_agent(task:str, max_rounds:int=12)->str: agents = create_agents() orchestrator = agents["orchestrator"] current_task = task task_history =[]for i inrange(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: context +=f"\n\n最新结果:\n{current_task}" decision_text = orchestrator.chat(context)print(f"[编排者] {decision_text[:300]}")# 解析决策try:# 从回复中提取 JSONimport re json_match = re.search(r'\{.*\}', decision_text, re.DOTALL)ifnot 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")classSmartAgent:"""集记忆、工具、多轮对话于一体的智能 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:returnf"搜索「{query}」的结果: ...（模拟数据）"def_tool_calculate(self, expression:str)->str: allowed =set("0123456789+-*/().% ")ifall(c in allowed for c in expression):returnstr(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"})returnf"已保存笔记（{len(content)} 字）"# --- 核心对话 ---defchat(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 应用场景与性能对比

22%18%15%12%10%10%8%5%AI Agent 典型应用场景分布代码开发辅助数据分析与报表客服与工单处理内容创作与 SEO自动化运维研究与知识管理工作流自动化其他

关键设计考量

九、总结

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

ReAct 循环
思考→行动→观察

工具系统
Function Calling

记忆系统
短期+长期

多 Agent
分工协作

生产就绪
安全+可观测

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