☰
AI Agent开发实战:从零搭建大模型驱动的调研助手
2026/10/20 14:49:04 网站建设 项目流程

分享一套 AI 大模型 Agent 开发实战教程,覆盖环境搭建、Agent 主循环设计、工具注册、插件机制与工作流编排。项目中会用一个可运行的“调研助手”串联全部知识点,代码可以直接复制改造,零基础可以跟着搭,有经验的开发者也可以用来做工程化参考。

1. 背景:为什么普通 API 调用不够用

先看一个常见的业务需求:用户输入“帮我查一下 Python 3.13 的最新特性,并整理成一篇简短的调研报告”。

如果只用单次大模型 API 调用,模型只能根据训练数据“回忆”,能查到的时间点可能明显滞后,也无法告诉你真正的官方发布说明。如果让大模型自主决定“先搜索网页→读取候选页面→提取关键信息→按报告模板输出”,整个任务的完成度就会高很多。这种“能调用工具、能规划步骤、能根据中间结果决定下一步”的程序结构,就是 AI Agent。

从技术角度来看,AI Agent 是一个以大型语言模型(简称大模型)为决策核心的软件实体。它不只是一个“文本生成器”,而是一个具备循环控制逻辑的执行器:

  1. 接收用户目标。
  2. 让大模型生成下一步行动(调用工具,或者直接回答)。
  3. 执行工具并拿到返回结果。
  4. 把结果反馈给大模型。
  5. 重复,直到完成目标或达到最大轮数。

与单次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/activate

Windows 环境激活命令是:

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.txt

2.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 的内部步骤如何组织。

简单任务不需要工作流,一个循环就能完成。但复杂任务需要阶段划分,例如:

  1. 意图理解。
  2. 信息收集。
  3. 结构化输出。

插件机制回答的问题是:如何在不修改主程序的情况下扩展 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.py

4.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 项目在安全、成本、可观测性方面的常见工程问题。

下一步建议按这个顺序继续深入学习:

  1. 给 Agent 增加长期记忆。用向量数据库保存历史对话,让 Agent 记住用户偏好。
  2. 引入多个 Agent 协作。例如“研究 Agent”负责查资料,“写作 Agent”负责润色,两个 Agent 通过消息队列通信。
  3. 完善工作流配置化。把阶段顺序、工具白名单、最大轮数全部下沉到 YAML 或 JSON 配置中。
  4. 加入评测集。用一批固定问题测试 Agent 输出质量,防止改动代码后效果回退。

如果你正在把 Agent 从 Demo 推向生产,建议在项目初期就把工具协议、日志结构和评测集提前定义好,这三个点后期返工的代价最高。现在可以打开终端,用本文的代码搭一个自己的 Agent 骨架,然后逐步替换成大模型服务和你自己的业务工具。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询