AI Agent开发实战:从零搭建大模型驱动的调研助手

📅 发布时间:2026/9/9 19:49:29
AI Agent开发实战:从零搭建大模型驱动的调研助手 分享一套 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内容如下openai1.35.0 python-dotenv1.0.1 PyYAML6.0.1 requests2.32.3 rich13.7.1依赖说明openaiPython 官方 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_KEYsk-你的key OPENAI_BASE_URLhttps://api.openai.com/v1 OPENAI_MODELgpt-4o-mini MAX_AGENT_STEPS6需要注意不同模型服务的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_keyos.getenv(OPENAI_API_KEY), base_urlos.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( modelmodel_name, messagesmessages, toolstools, tool_choiceauto, ) 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_asciiFalse), } ) 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, paramsparams, timeout10) 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( namesearch_web, description搜索互联网返回与查询相关的标题和链接列表。, parameters{ type: object, properties: { query: {type: string, description: 搜索关键词}, max_results: {type: integer, description: 返回结果数量默认5}, }, required: [query], }, funcsearch_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 fplugins.{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_okTrue) result execute_research(goal) file_path Path(output_dir) / report.md file_path.write_text(result, encodingutf-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 骨架然后逐步替换成大模型服务和你自己的业务工具。