分享一套 AI 大模型 Agent 开发实战教程,覆盖环境搭建、Agent 主循环设计、工具注册、插件机制与工作流编排。项目中会用一个可运行的“调研助手”串联全部知识点,代码可以直接复制改造,零基础可以跟着搭,有经验的开发者也可以用来做工程化参考。
1. 背景:为什么普通 API 调用不够用
先看一个常见的业务需求:用户输入“帮我查一下 Python 3.13 的最新特性,并整理成一篇简短的调研报告”。
如果只用单次大模型 API 调用,模型只能根据训练数据“回忆”,能查到的时间点可能明显滞后,也无法告诉你真正的官方发布说明。如果让大模型自主决定“先搜索网页→读取候选页面→提取关键信息→按报告模板输出”,整个任务的完成度就会高很多。这种“能调用工具、能规划步骤、能根据中间结果决定下一步”的程序结构,就是 AI Agent。
从技术角度来看,AI Agent 是一个以大型语言模型(简称大模型)为决策核心的软件实体。它不只是一个“文本生成器”,而是一个具备循环控制逻辑的执行器:
- 接收用户目标。
- 让大模型生成下一步行动(调用工具,或者直接回答)。
- 执行工具并拿到返回结果。
- 把结果反馈给大模型。
- 重复,直到完成目标或达到最大轮数。
与单次Prompt → Response相比,Agent 的核心差异在于引入了“循环”和“工具”。这也是为什么掌握 Agent 开发,不只是会调大模型接口,还需要熟悉流程编排、插件扩展和异常处理。
在具体应用上,Agent 适合这些场景:
- 信息搜集与报告生成:搜索、阅读、整理、输出。
- 数据库操作助手:把自然语言转换成 SQL。
- 代码辅助工具:读取仓库文件、执行测试、修复报错。
- 自动化运维:根据监控指标进行诊断并触发操作。
本文以一个“调研助手 Agent”为主线,完整演示 Python 驱动的 Agent 开发流程。项目中的设计思路不绑定具体大模型厂商,即使你换用其他模型,核心代码仍然可以复用。
2. 环境准备与版本说明
2.1 运行环境
本文示例代码以 Python 3.10 及以上版本编写。如果你本机还没有 Python,建议先安装 3.10 或 3.11 版本,并确认命令行可以执行python --version。
在项目开发中强烈推荐使用虚拟环境,避免依赖互相污染:
mkdir ai-agent-demo && cd ai-agent-demo python -m venv venv source venv/bin/activateWindows 环境激活命令是:
venv\Scripts\activate激活后,命令行的路径前缀会出现(venv),表示已经进入虚拟环境。
2.2 依赖清单
创建requirements.txt,内容如下:
openai==1.35.0 python-dotenv==1.0.1 PyYAML==6.0.1 requests==2.32.3 rich==13.7.1依赖说明:
openai:Python 官方 SDK。本文用它调用大模型的 Chat Completions 接口,并演示工具调用。其他模型服务如果兼容 OpenAI 协议,也可以复用这套代码。python-dotenv:读取.env文件中的环境变量,用来保存 API Key,避免硬编码.PyYAML:用于定义工作流配置文件。requests:用于搜索工具发起 HTTP 请求。rich:在终端中美化输出,方便观察 Agent 的执行过程。
安装依赖:
pip install -r requirements.txt2.3 API Key 准备
在项目根目录创建.env文件:
OPENAI_API_KEY=sk-你的key OPENAI_BASE_URL=https://api.openai.com/v1 OPENAI_MODEL=gpt-4o-mini MAX_AGENT_STEPS=6需要注意,不同模型服务的OPENAI_BASE_URL不同。例如,某些兼容 OpenAI 协议的国产模型服务,需要把地址换成官方文档提供的地址。代码层面不需要大改,只需要修改环境变量。
以下两种情况不会写入本文:
- 生产环境中,密钥应放在密钥管理服务中,而不是直接放在
.env并提交到 Git 仓库。 - 不要在你的博客或公开代码仓库中贴出真实 API Key。
3. Agent 核心概念拆解
3.1 感知、思考、行动循环
Agent 的基础结构可以概括成一个循环:
用户目标 -> 大模型生成决策 -> 执行工具 -> 返回结果 -> 大模型继续决策这个循环通常被称为 ReAct 模式。ReAct 强调把“推理”和“行动”交替进行。大模型在每一轮输出中说明自己发现了什么、下一步需要做什么;程序负责执行真正的外部动作,例如搜索网页、调用计算器、执行 Python 代码。
下面是一个极简的伪代码:
def run_agent(goal: str): messages = [{"role": "user", "content": goal}] for step in range(MAX_STEPS): reply = llm_chat(messages) action = parse_action(reply) if action is None: return reply result = execute_tool(action) messages.append({"role": "function", "content": result})理解这一节的关键点在于:
- 大模型不负责真正“做”事情,只负责“决策”。
- 工具执行结果必须重新塞回上下文,成为后续决策的依据。
- 必须有最大循环次数限制,否则 Agent 可能在错误的路径上无限循环。
3.2 工具调用(Function Calling)
为了让大模型能够调用函数,需要在请求中额外声明“工具列表”。每个工具需要描述:
- 函数名称。
- 函数的作用。
- 参数名和参数类型。
- 哪些参数是必填的。
以大模型 OpenAI 协议为例,请求中的tools参数是一个列表,每个元素包含type和function。模型看到工具描述后,如果认为需要调用,会返回一条tool_calls信息,里面包含函数名和参数 JSON。程序解析这个 JSON,然后执行本地函数。
这种设计的本质是:把大模型从“只能生成文本”扩展到“能操纵外部系统”。工具定义得越清晰,模型调用就越准确。反过来,如果工具描述含糊,模型就会频繁调用错误参数。
3.3 工作流与插件机制
工作流回答的问题是:Agent 的内部步骤如何组织。
简单任务不需要工作流,一个循环就能完成。但复杂任务需要阶段划分,例如:
- 意图理解。
- 信息收集。
- 结构化输出。
插件机制回答的问题是:如何在不修改主程序的情况下扩展 Agent 的能力。
在工程实践中,你不可能把每个工具都写进主函数。更好的做法是:
- 定义一个工具基类或协议。
- 每个插件实现统一的
name、description、run接口。 - 启动时扫描插件目录,自动注册工具。
这样,后续添加“查天气”“查数据库”“发邮件”等能力时,只需要新增一个插件文件,不需要改动 Agent 核心代码。
4. 完整实战:实现一个“调研助手 Agent”
接下来进入项目实战。我们将实现一个能自动搜索网络资料、总结要点、输出结构化报告的 Agent。
4.1 需求拆解
调研助手需要做到:
- 接收用户输入的研究主题。
- 通过搜索工具获取相关资料。
- 根据搜索结果生成文章摘要。
- 把摘要整理成 Markdown 报告。
- 如果首次检索内容不足,可以再次搜索。
用户使用方式:
python main.py "Python 3.13 新特性"预期输出是一份包含“核心信息、来源说明、总结建议”的 Markdown 报告。
4.2 项目结构
最终目录结构如下:
ai-agent-demo/ ├── .env ├── requirements.txt ├── agent/ │ ├── __init__.py │ ├── core.py │ ├── tools.py │ └── workflow.py ├── plugins/ │ └── search_plugin.py ├── output/ └── main.py4.3 编写 Agent 主循环
agent/core.py是全局核心,负责大模型调用、消息维护和循环控制。先看完整代码:
# agent/core.py import json import os from dotenv import load_dotenv from openai import OpenAI load_dotenv() client = OpenAI( api_key=os.getenv("OPENAI_API_KEY"), base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"), ) model_name = os.getenv("OPENAI_MODEL", "gpt-4o-mini") max_steps = int(os.getenv("MAX_AGENT_STEPS", "6")) def chat_with_tools(messages, tools): response = client.chat.completions.create( model=model_name, messages=messages, tools=tools, tool_choice="auto", ) return response.choices[0].message上面的chat_with_tools返回模型消息,消息里可能带有工具调用请求,也可能没有。下面是带循环的正式执行入口:
def run_agent(goal: str, tools_definition: list, tool_runner): messages = [ {"role": "system", "content": "你是一个严谨的调研助手。"}, {"role": "user", "content": goal}, ] for step in range(max_steps): print(f"[Agent] 第 {step + 1} 轮决策中...") message = chat_with_tools(messages, tools_definition) if not message.tool_calls: return message.content messages.append( { "role": "assistant", "content": message.content, "tool_calls": [ { "id": tc.id, "type": "function", "function": { "name": tc.function.name, "arguments": tc.function.arguments, }, } for tc in message.tool_calls ], } ) for tc in message.tool_calls: func_name = tc.function.name func_args = json.loads(tc.function.arguments) print(f"[Tool] 调用 {func_name},参数:{func_args}") result = tool_runner(func_name, func_args) messages.append( { "role": "tool", "tool_call_id": tc.id, "content": json.dumps(result, ensure_ascii=False), } ) return "已达到最大步骤数,任务结束。"关键逻辑说明:
messages中保留了完整的对话历史,工具调用结果也写入历史。tool_calls是列表,一次决策可能同时调用多个工具。- 每次工具执行结果通过
role: "tool"回传给模型,并带上对应的tool_call_id。 - 如果返回的消息不再包含
tool_calls,表示 Agent 完成了任务。
4.4 工具注册中心与插件开发
为了让工具定义与工具实现不散落各处,这里做一个注册中心。
agent/tools.py内容:
# agent/tools.py import inspect class ToolRegistry: def __init__(self): self._tools = {} def register(self, name, description, parameters, func): self._tools[name] = { "definition": { "type": "function", "function": { "name": name, "description": description, "parameters": parameters, }, }, "func": func, } def definitions(self): return [item["definition"] for item in self._tools.values()] def run(self, name, args): if name not in self._tools: return {"error": f"未知工具: {name}"} return self._tools[name]["func"](**args) registry = ToolRegistry()这里的ToolRegistry承担两件事:
- 向外提供工具的定义列表,供大模型读取。
- 根据模型返回的工具名称执行实际函数。
接下来做一个搜索插件。为了演示方便,这里使用 DuckDuckGo 的通用 JSON 接口作为搜索来源。生产项目中请替换为可用的搜索 API。
plugins/search_plugin.py:
# plugins/search_plugin.py import requests from agent.tools import registry def search_web(query: str, max_results: int = 5): url = "https://api.duckduckgo.com/" params = { "q": query, "format": "json", "no_html": 1, "skip_disambig": 1, } try: resp = requests.get(url, params=params, timeout=10) resp.raise_for_status() data = resp.json() results = [] for topic in data.get("RelatedTopics", [])[:max_results]: if "Text" in topic: results.append({"title": topic.get("Text", ""), "url": topic.get("FirstURL", "")}) return {"query": query, "results": results} except Exception as e: return {"query": query, "error": str(e), "results": []} def register_search_tool(): registry.register( name="search_web", description="搜索互联网,返回与查询相关的标题和链接列表。", parameters={ "type": "object", "properties": { "query": {"type": "string", "description": "搜索关键词"}, "max_results": {"type": "integer", "description": "返回结果数量,默认5"}, }, "required": ["query"], }, func=search_web, )定义一个加载插件的函数:
# plugins/__init__.py import importlib from pathlib import Path PLUGIN_DIR = Path(__file__).resolve().parent def load_plugins(): for py_file in PLUGIN_DIR.glob("*_plugin.py"): module_name = f"plugins.{py_file.stem}" module = importlib.import_module(module_name) register_func = getattr(module, "register", None) if register_func: register_func()插件文件统一实现register()函数,在程序启动时被加载。这样新增一个插件只需要在plugins目录下新增xxx_plugin.py文件,并在其中调用registry.register(...)即可。
4.5 工作流编排
调研任务可以按阶段组织。为了让流程更清晰,在agent/workflow.py中实现一个简单的阶段执行器。
# agent/workflow.py from agent.core import run_agent from agent.tools import registry from plugins import load_plugins def execute_research(goal: str) -> str: load_plugins() report = run_agent(goal, registry.definitions(), registry.run) return report如果想加入更复杂的阶段控制,例如“先搜索两次,再输出总结”,可以引入一个简单的步骤列表。下面是一个更工程化的版本:
STAGE_PROMPTS = { "search": "请先搜索与题目相关的资料,尽量找到 2~3 个不同的信息源。", "summary": "基于已获得的信息,写一份 500 字左右的总结报告。", } def execute_research_with_stages(goal: str) -> str: load_plugins() messages = [] for stage, prompt in STAGE_PROMPTS.items(): print(f"[Workflow] 当前阶段:{stage}") stage_goal = f"{goal}\n\n{prompt}" messages.append({"role": "user", "content": stage_goal}) result = run_agent(stage_goal, registry.definitions(), registry.run) messages.append({"role": "assistant", "content": result}) return messages[-1]["content"]实际项目的工作流往往会写进配置文件。比如使用 YAML 定义阶段顺序、最大轮数和是否启用某个插件。这样产品运营人员也能调整流程,而不需要改动代码。
4.6 主入口与运行
main.py:
# main.py import sys from agent.workflow import execute_research def main(): goal = sys.argv[1] if len(sys.argv) > 1 else "Python 3.13 新特性" output_dir = "output" from pathlib import Path Path(output_dir).mkdir(exist_ok=True) result = execute_research(goal) file_path = Path(output_dir) / "report.md" file_path.write_text(result, encoding="utf-8") print("报告已生成:", file_path) if __name__ == "__main__": main()运行:
python main.py "Python 3.13 新特性"运行过程中终端会显示 Agent 的决策轮次和工具调用情况:
[Agent] 第 1 轮决策中... [Tool] 调用 search_web,参数:{'query': 'Python 3.13 new features', 'max_results': 5} [Agent] 第 2 轮决策中... [Tool] 调用 search_web,参数:{'query': 'Python 3.13 release notes official', 'max_results': 5} [Agent] 第 3 轮决策中... 报告已生成: output/report.md生成的output/report.md是模型根据搜索结果整理的 Markdown 报告。由于不同模型配置差异,实际内容会有区别,这属于正常现象。如果你只看到“已达到最大步骤数”,说明当前搜索接口返回的内容不足以让模型收敛,可以适当增加MAX_AGENT_STEPS或者改善工具结果质量。
5. 常见问题与排查思路
5.1 API 调用报错
常见错误集中在环境变量、网络和模型权限方面。
| 问题现象 | 常见原因 | 解决思路 |
|---|---|---|
OpenAIError: The api_key client option must be set | 环境变量未加载或未配置 | 检查.env文件是否存在,确认load_dotenv()被调用 |
APIConnectionError | 网络无法访问目标 API 地址 | 检查网络的出口连通性,确认OPENAI_BASE_URL是否与企业网络策略冲突 |
ModelNotFoundError | 模型名称不存在或无权限 | 查看模型服务商文档,确认当前账号可以访问该模型 |
如果你是使用代理访问 OpenAI 服务的开发者,尤其是在企业内网环境中,请务必先确认你的代理配置和出口合规策略是否允许访问目标 API。本文不讨论任何代理技术的具体配置。总之,API 调用报错时先确认网络出口、密钥和模型名三个变量。
5.2 工具调用格式错误
模型返回的arguments可能不是合法 JSON。原因通常是:
- 模型输出被截断。
- 模型返回了 Markdown 代码块包裹的 JSON。
- 部分模型的工具调用接口并不严格遵循 OpenAI 格式。
建议在解析arguments时增加兜底逻辑:
import json def safe_json_loads(text): try: return json.loads(text) except json.JSONDecodeError: start = text.find("{") end = text.rfind("}") + 1 if start >= 0 and end > start: return json.loads(text[start:end]) return {}这段代码会尝试截取 JSON 子串,减少因模型输出多余内容导致的解析失败。
5.3 搜索工具返回空结果
DuckDuckGo 的示例接口返回字段有限,很多时候RelatedTopics是空的。这会造成 Agent 在空信息下继续编造内容。
解决方向有:
- 接入专业搜索 API,例如 Bing Web Search API 或其他商用搜索服务。
- 在插件返回
results=[]时,明确告诉模型“没有搜索到有效资料,请换关键词重试”。 - 增加搜索历史,避免重复调用相同关键词。
工具返回信息质量,直接决定 Agent 最终输出质量。这个结论在几乎所有 Agent 项目中都成立。
5.4 循环不收敛
如果 Agent 一直在调用同一个工具、参数也几乎不变,说明模型认为“当前信息仍然不够”。解决方式包括:
- 降低最大步骤数,让 Agent 更快给出结论。
- 在系统提示词中增加约束:“如果搜索结果没有新增信息,请基于现有信息输出报告。”
- 对工具结果做去重,反复出现的高相似度内容不再传给模型。
6. 最佳实践与工程建议
6.1 工具描述是 Agent 能力的关键
同样的模型,工具描述写得清晰,调用准确率明显更高。工具描述应该包含:
- 工具解决什么问题。
- 参数的单位、格式和可选范围。
- 典型使用示例。
例如,搜索工具的参数描述不要只写“搜索关键词”,而是写:
query: 搜索关键词,建议使用英文和中文两种语言分别搜索以获取更多结果6.2 安全边界与权限控制
Agent 能调用工具,就必须考虑权限边界:
- 不在工具函数中直接执行任意 Python 字符串。
- 涉及文件删除、数据库写入、发送消息等危险动作时,必须增加人工审批节点。
- 每个工具函数只暴露最小必要能力,不要在工具里给模型完整的 Shell 权限。
如果你的 Agent 会被多个用户使用,还需要考虑用户级别的权限隔离。例如用户 A 不能通过 Agent 查询用户 B 的数据。
6.3 日志与可观测性
Agent 的调试难度比普通接口高,因为它的行为依赖上下文和历史动作。生产环境中必须记录:
- 每一轮的完整消息列表。
- 工具名称、参数、返回结果。
- 每轮耗时和 token 消耗。
- 最终输出和终止原因。
推荐把日志写到结构化文件中,方便后续用日志平台检索分析。不要让团队靠终端输出排查线上问题。
6.4 成本控制
每一次工具调用都会把历史消息重新传给模型,token 消耗会随着步骤递增。控制成本的方法:
- 限制最大步骤数。
- 对大段工具返回内容做截断或摘要。
- 定期清理无效历史信息。
- 使用更便宜的模型处理简单动作,复杂总结才用强模型。
6.5 可维护性
不要把所有工具都堆在一个大文件里。建议约定:
- 每个插件文件对应一类能力。
- 插件内部自行完成参数校验。
- 插件返回统一结构:
{"ok": bool, "data": ..., "error": ...}。
这样即使后续引入几十个工具,主流程代码依然稳定。
7. 总结与学习路线
本文围绕“Python 驱动智能 Agent”这条主线,从概念讲到代码,完成了一个可运行的调研助手项目。
你现在应该已经掌握:
- Agent 与大模型 API 调用的区别。
- ReAct 循环的基本组成:决策、执行、反馈、循环。
- 工具注册与插件化扩展的方式。
- 工作流编排的基本思路。
- Agent 项目在安全、成本、可观测性方面的常见工程问题。
下一步建议按这个顺序继续深入学习:
- 给 Agent 增加长期记忆。用向量数据库保存历史对话,让 Agent 记住用户偏好。
- 引入多个 Agent 协作。例如“研究 Agent”负责查资料,“写作 Agent”负责润色,两个 Agent 通过消息队列通信。
- 完善工作流配置化。把阶段顺序、工具白名单、最大轮数全部下沉到 YAML 或 JSON 配置中。
- 加入评测集。用一批固定问题测试 Agent 输出质量,防止改动代码后效果回退。
如果你正在把 Agent 从 Demo 推向生产,建议在项目初期就把工具协议、日志结构和评测集提前定义好,这三个点后期返工的代价最高。现在可以打开终端,用本文的代码搭一个自己的 Agent 骨架,然后逐步替换成大模型服务和你自己的业务工具。