Agent 开发这个方向最近被问到最多的往往不是“大模型怎么接入”而是“接入之后怎么让模型按照流程执行任务”以及“项目里的插件到底该怎么设计”。这篇文章围绕 Python 驱动 AI 大模型 Agent 的完整实战链路展开核心覆盖三块最小 Agent 怎么搭、工作流怎么编排、插件怎么扩展。如果你已经能调用大模型接口但不知道下一步该做什么或者正在搭建一个需要自动处理文档、生成报告、对接内部系统的项目这篇可以直接拿来当参考线。我建议不要一上来就接一堆 Agent 框架。先用 Python 写一个最小闭环模型收到任务判断需要哪些外部工具调用工具拿到结果再继续推理直到输出最终答案。跑通这个循环之后再往里面加多步骤工作流和插件系统思路会清晰很多。1. 先别急着写代码Agent、大模型、工作流、插件到底怎么分工1.1 普通模型调用和 Agent 的核心差异大部分刚接触大模型开发的人第一段代码通常是这样的把用户问题拼到 messages 里调用一次模型接口把返回内容打印出来。这是单轮对话模型没有外部信息也没有执行动作的能力。Agent 和普通模型调用的最大区别是多了“循环”和“工具”。在 Agent 模式里模型可以输出一个特殊请求比如“我需要查询某个数据库”“我需要读取某个文件”“我需要调用某个业务接口”。你的 Python 程序收到这个请求后去执行对应的函数把执行结果作为一条新消息返回给模型。模型看到结果后决定是继续调用工具还是直接给出最终答案。这个过程拆开看有三层大模型负责理解任务、拆解计划、生成自然语言结果。程序负责执行工具函数、控制循环、管理上下文。工具函数负责真正拿到外部数据或产生外部效果。理解了这个结构后面所有代码都是围绕它展开的。1.2 代码驱动、低代码平台、混合开发的适用边界现在做 Agent 项目主要有三种形态。第一种是代码驱动。用 Python 直接调模型接口自己控制循环、工具调用、工作流编排。它的优点是可测试、可版本管理、可精确控制逻辑适合生产环境、本地批处理、私有数据接入。缺点是开发门槛高需要自己处理很多边界情况。第二种是低代码平台。比如 Dify、Coze、n8n 这类可视化工作流平台拖拽节点就能搭出一个 Agent 流程。适合快速验证想法、给业务同学做原型、或处理比较标准的在线流程。缺点也很明显复杂分支、私有部署、本地资源访问、异常处理都比较受限。第三种是混合开发。用低代码平台搭主流程用自定义 Python 插件处理特殊输入输出或者反过来用 Python 做核心逻辑用平台来做可视化配置。到底选哪种取决于任务是否固定、是否需要本地文件处理、是否需要和内部系统深度绑定。如果只是学习我建议直接从代码驱动的最小 Agent 开始。后面所有概念你在代码里亲手实现一遍比看十个平台教程都有用。2. Python 环境与项目骨架版本、依赖、密钥、目录一次理清2.1 环境准备清单先确认基础环境。Python 版本建议用 3.10 以上如果你机器上有多个 Python 版本务必用虚拟环境隔离不要直接往系统环境里装包。创建虚拟环境并激活python -m venv .venv # Windows .venv\Scripts\activate # macOS / Linux source .venv/bin/activate安装依赖时先装最小集。很多 Agent 教程会直接让你装 LangChain 全家桶但实际从学习角度我更建议先装这几个pip install openai python-dotenv requestsopenai 这个包是兼容接口的常用 SDKpython-dotenv 用来读取 .env 文件requests 用于后续写插件时调用业务接口。如果你需要做数据处理后面按需再装 pandas。原始项目没有给出明确依赖清单实际落地时以你使用的模型服务商文档为准。环境变量建议统一放 .env 文件格式像这样LLM_API_KEY你的密钥 LLM_BASE_URL你的接口地址 LLM_MODEL你的模型名然后在代码里用 python-dotenv 加载import os from dotenv import load_dotenv load_dotenv() api_key os.getenv(LLM_API_KEY) base_url os.getenv(LLM_BASE_URL) model_name os.getenv(LLM_MODEL)密钥不要硬编码在代码里更不要提交到 Git 仓库。.env 文件一定要加进 .gitignore。2.2 项目目录结构设计项目一开始目录结构就要分清楚。不然插件写多了、任务种类多了之后代码很快会变成一大坨。一个适合中小型 Agent 项目的目录大概是这样的agent_project/ ├── .venv/ ├── .env ├── .gitignore ├── config.py ├── core/ │ ├── __init__.py │ ├── llm.py │ ├── agent.py │ ├── registry.py │ └── pipeline.py ├── plugins/ │ ├── __init__.py │ ├── file_tool.py │ ├── search_tool.py │ └── report_tool.py ├── data/ │ ├── input/ │ └── output/ └── scripts/ └── run_demo.pycore 里放模型调用、Agent 循环、工作流引擎。plugins 里放各类插件也就是可被模型调用的工具函数。data/input 放待处理文件data/output 放结果文件。scripts 放入口脚本。这样分层的好处是以后新增一个插件不用改 Agent 主逻辑新增一个工作流不用动插件文件排查问题的时候日志、输入、输出都有固定位置。2.3 为什么先跑通最小调用不要急着接复杂框架很多人卡在环境问题上一整天其实问题很可能不是模型接口而是依赖版本冲突、密钥没读到、或者目录不对。所以我建议所有 Agent 项目第一步都先做一次“最小模型调用测试”。代码不复杂就是读取环境变量发一条简单消息from openai import OpenAI client OpenAI(api_keyapi_key, base_urlbase_url) resp client.chat.completions.create( modelmodel_name, messages[{role: user, content: 你好请回复连接正常}], ) print(resp.choices[0].message.content)能打印出内容说明密钥、地址、模型名、网络连接都没问题。这一步跑通了再往里面加 Agent 循环就不会频繁怀疑“是不是模型接口坏了”。实际项目中我见过很多次因为模型名填错、接口地址少了个斜杠、密钥文件没被加载导致后面所有 Agent 逻辑看起来都像坏了。最小调用测试就是帮你把变量切到最小范围。3. 搭建最小可用 Agent让大模型学会调用工具函数3.1 函数调用机制模型不直接执行而是请求工具结果Agent 里最核心的机制是“函数调用”。你需要理解模型本身不会去读文件、不会查数据库、不会发 HTTP 请求。它只是在生成文本的时候多了一种输出类型叫“工具调用请求”。举个例子你告诉模型有一个工具叫 read_file描述是“读取指定路径的文本文件”输入参数是 path。模型处理任务时发现需要读取文件内容它就会输出一个结构化结构内容大概是“我要调用 read_file参数是 xxx”。你的程序收到这个请求后去执行真正的 Python 函数把函数返回的字符串作为工具结果消息追加到对话里。模型再基于这个结果继续推理。很多 Agent 报错“execution terminated due to error”或者卡住不动往往不是模型问题而是这个循环里没有正确把工具结果回传给模型。所以后面写循环时要特别注意消息格式。3.2 最小示例工具注册表 循环调用我不想在没有任何项目源码的情况下生造一个复杂框架。这里给一个通用的最小实现思路你可以按照自己的模型服务商文档适配。第一步做一个简单的工具注册表。用字典来维护工具名、描述、执行函数import json TOOL_REGISTRY {} def tool(name, description): def decorator(func): TOOL_REGISTRY[name] { fn: func, description: description, } return func return decorator tool(get_current_time, 获取当前系统时间当用户询问时间时使用) def get_current_time(): import datetime return datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S)第二步把注册表里的工具转换成模型接口需要的 tools 参数格式。不同服务商格式有差异大部分兼容接口的结构类似下面这样TOOL_SCHEMAS [] for name, meta in TOOL_REGISTRY.items(): TOOL_SCHEMAS.append({ type: function, function: { name: name, description: meta[description], parameters: { type: object, properties: {}, }, }, })如果你的工具需要参数可以在 properties 里定义字段。刚开始不要设计得太复杂参数越简单模型越容易正确生成。第三步写一个 Agent 循环。大致逻辑如下MAX_ROUNDS 5 messages [ {role: system, content: 你是一个能调用工具的助手根据用户问题选择合适工具。}, {role: user, content: 现在是几点}, ] for _ in range(MAX_ROUNDS): resp client.chat.completions.create( modelmodel_name, messagesmessages, toolsTOOL_SCHEMAS, ) msg resp.choices[0].message if msg.tool_calls: # 先把助手消息追加进去 messages.append(msg) for tc in msg.tool_calls: tool_name tc.function.name arguments json.loads(tc.function.arguments or {}) try: result TOOL_REGISTRY[tool_name][fn](**arguments) except Exception as e: result f工具执行出错: {e} messages.append({ role: tool, tool_call_id: tc.id, content: json.dumps(result, ensure_asciiFalse), }) else: # 模型不再要求调用工具输出最终答案 final_answer msg.content break print(final_answer)这段代码的核心判断是每轮都检查模型返回的 msg.tool_calls 是否为空。不为空就执行工具并回传结果为空就认为 Agent 完成了任务。一定要注意不同 SDK 返回结构会略有差异。比如某些接口 tc.id 的字段名可能不同模型消息对象的追加方式也可能不同。写代码时把返回对象先打印出来看一眼再决定字段名。不要照抄网上代码就以为一定跑通。3.3 参数与边界温度、max_tokens、循环轮数和上下文控制Agent 循环跑通之后不要急着加功能先把几个关键参数理解清楚。temperature 控制随机性。抽取类任务比如从文本中提取字段、做结构化输出建议设为 0 或接近 0。创意类任务比如写文案、头脑风暴可以调到 0.7 到 0.9。不要所有任务都用同一个值。max_tokens 控制单次输出长度。如果模型经常输出一半就断掉要考虑调大。但如果任务本身是短输出比如抽取关键字段设一个较小的值可以避免模型输出多余的废话。MAX_ROUNDS 是循环轮数上限。我建议学习阶段设 5 到 10不要设成无限循环。如果你发现同一个工具被连续调用三轮以上大概率说明工具结果没有让模型得到新信息或者任务描述不清楚。上下文控制是最容易被忽略的点。每次调用工具、回传结果messages 都在变长。几十轮之后长文档工具结果很容易撑爆上下文限制。解决办法之一是在工具函数内部做截断比如只返回前 2000 个字符另一个办法是定期压缩历史消息保留系统指令、最初任务、最近几轮关键结果。这里补充一个判断标准如果你的 Agent 只是做单文档抽取上下文一般不会太大如果要做批量文档处理上下文管理就必须提前设计否则跑几条任务之后就会报错。4. 工作流搭建从单次 Agent 调用变成可追踪的任务流水线4.1 工作流的第一步是定义步骤不是写代码很多人的工作流是从“把多个模型调用串在一起”开始的结果经常是流程越加越长出了问题根本不知道是哪一步导致的。真正的工作流设计第一步应该是定义步骤。我以“每周自动读一批文档并生成分析报告”为例。可以先拆成这些步骤扫描输入目录拿到待处理文件列表。读取每个文件内容。对每个文件调用抽取 Agent得到结构化字段。校验抽取结果缺失字段做标记。把全部抽取结果汇总。调用汇总 Agent生成 Markdown 报告。保存报告到输出目录。这一步做完你就能看到哪些步骤是和模型交互哪些步骤是纯 Python 处理哪些步骤可能出错需要重试。4.2 一个简单的 Pipeline 骨架工作流引擎不一定要用现成框架。小项目里一个简单的 Pipeline 类就够用。我习惯把每一步定义成一个函数输入一个字典输出一个字典。这样每一步之间只通过 dict 传递数据不共享全局变量追查问题非常方便。class Pipeline: def __init__(self, run_id): self.steps [] self.run_id run_id def add_step(self, name, handler): self.steps.append({name: name, handler: handler}) return self def run(self, payload): current payload for step in self.steps: print(f[{self.run_id}] start {step[name]}) current step[handler](current) print(f[{self.run_id}] finish {step[name]}) return current然后按顺序注册pipeline Pipeline(run_id20250101) pipeline.add_step(list_files, list_input_files) pipeline.add_step(read_files, read_files) pipeline.add_step(extract_fields, extract_fields_with_agent) pipeline.add_step(validate_result, validate_result) pipeline.add_step(merge_summary, merge_summary) pipeline.add_step(save_report, save_report) result pipeline.run({input_dir: data/input, output_dir: data/output})每一步的 handler 里建议把关键信息打印出来尤其是输入文件数量、抽取成功数量、校验失败数量。这样跑批处理任务时你能随时知道进度。4.3 日志、状态、重试与断点恢复工作流跑单条任务很容易跑批量任务才是真正的考验。批量任务最容易出的问题是第 5 个文件出错了你修完代码是从第 1 个文件重新跑还是从第 6 个文件继续跑如果每次都全量重跑时间成本会非常高。解决思路是给每个任务分配一个唯一 ID并记录每个步骤的状态。最简单的方式是每完成一个文件就把中间结果写成 JSON 存到临时目录。下次运行时如果检测到某个文件已经完成抽取就直接读取结果跳过模型调用。这个“断点恢复”机制在 Agent 项目里非常重要。因为模型调用是最耗时、最花钱的环节能跳过就一定要跳过。重试也要有策略。模型接口偶尔超时是正常现象但不要无限重试。我一般建议最多重试 3 次重试间隔 1 到 2 秒。如果重试后仍然失败把失败任务写入单独的错误目录不要影响整个工作流继续跑。日志建议用标准 logging 模块而不是到处 print。因为批量任务会一直运行print 的内容容易丢失且无法按级别过滤。把日志同时输出到控制台和文件排查问题会容易很多。4.4 低代码平台和代码工作流怎么选你在实际搜索时可能经常看到 Dify、Coze、n8n 这类平台也会看到 Flowable 这类传统流程引擎。这里我补充一个边界判断。Flowable 是传统 BPM 工作流引擎核心是审批流、状态机、人工任务和 AI Agent 的“大模型动态决策流程”不是一回事。如果你的项目是人事审批、工单流转应该用 BPM 引擎如果是要让大模型根据任务动态调用工具才属于 Agent 工作流。Dify、Coze、n8n 这类平台适合快速搭可视化流程。它们的好处是节点拖拽、调试方便适合原型验证。但是如果你需要本地文件批处理、私有数据接入、复杂分支判断或者需要把流程纳入 Git 版本管理代码 Pipeline 明显更可控。我的习惯是先想清楚最终落地场景。如果只是做一个演示 Demo低代码平台半小时能搞定如果要做成稳定跑每周任务、每天任务的生产服务代码实现虽然慢一点但后面维护省心很多。5. 插件开发用统一接口给 Agent 扩展真实能力5.1 插件在 Agent 项目里到底是什么插件开发这个词在不同领域含义不一样。IDE 插件是扩展编辑器功能浏览器插件是增强网页能力这里说的插件指的是 Agent 的工具函数也就是我在前面注册表里写的那些tool装饰器包裹的函数。从模型视角看一个插件其实就是三样东西名字用于模型输出调用请求。描述告诉模型这个工具什么时候该用。参数说明告诉模型调用时需要填哪些字段。从程序视角看插件是一个普通函数接收参数返回一个字符串或 JSON。返回的内容最终会被拼到 messages 里回传给模型。所以插件开发的核心不是写一个大类而是设计一个可靠的函数边界。5.2 一个通用插件接口示例下面是一个读取本地文件的插件示例。注意我这里只是演示设计思路实际路径规则、文件大小限制要按你的项目要求补。import os import json from core.registry import tool tool(read_text_file, 读取指定文本文件前 N 个字符适用于读取 Markdown、TXT 等内容文件) def read_text_file(path: str, max_chars: int 2000): try: with open(path, r, encodingutf-8) as f: content f.read() if len(content) max_chars: content content[:max_chars] \n...[已截断] return json.dumps({path: path, content: content}, ensure_asciiFalse) except Exception as e: return json.dumps({error: f读取文件失败: {e}}, ensure_asciiFalse)这里有两个设计点值得注意第一函数内部捕获了异常并把错误以字符串形式返回。这样模型能看到错误信息并可能调整策略而不是让整个程序崩溃。第二限制了返回长度。模型上下文是有限的工具返回内容越短Agent 越稳定。插件写完之后要记得让模型能感知到这个插件的存在。在 Agent 循环构造 tools 参数时需要把插件的名字、描述、参数结构传进去。这就是前面 TOOL_SCHEMAS 那段代码做的事。5.3 插件开发检查清单与常见问题我整理一份插件开发时比较实用的检查清单插件名是否唯一是否见名知意。description 是否写清楚了“什么时候该用”而不是只写“一个工具”。参数结构是否简单尽量用一级字段。函数是否有错误捕获是否返回结构化错误。返回值是否太长是否已经做截断或摘要。插件内是否硬编码了路径、密钥、账号。插件是否包含敏感操作如果包含是否做了权限校验或二次确认。常见问题里最典型的是“模型死活不调用某个插件”。这种情况先去检查 description。比如你写“文件处理工具”模型根本不知道什么时候该用。改成“当用户需要读取服务器上的本地 Markdown 文件时使用参数 path 为绝对路径”触发率会明显提升。另一种情况是插件调用时报错但模型没有感知到。原因是程序把异常吞掉了返回给模型的是空字符串或一个普通对象。正确做法是把错误信息写清楚回传格式保持稳定。模型读到错误后通常能重新调整输入参数或换一种解法。还有一种情况是参数解析失败。模型偶尔生成的 JSON 不规范比如多了一个逗号、用了单引号。解析失败时不要直接崩溃可以先把原始参数原样返回给模型提示它重新生成。这本质上是一种容错机制。6. 全项目实战自动读文档、抽字段、生成 Markdown 报告6.1 需求拆解前面概念讲了不少这一节我用一个完整的小项目把 Agent、工作流、插件串起来。需求读取 data/input 目录下的多个 Markdown 文档自动提取每篇文档的标题、简介、核心步骤最后生成一份汇总报告 report.md。这个需求可以覆盖三个核心点文档读取插件、抽取 Agent、报告生成工作流。先拆步骤列出 input 目录下所有 .md 文件。对每个文件调用 read_text_file 插件读取内容。调用模型让模型输出一篇文档的结构化信息。校验结构化信息是否包含标题、简介、核心步骤。汇总所有文档的结构化信息。调用模型生成 Markdown 汇总报告。保存 report.md 到 output 目录。6.2 分步实现先写一个简单的提示词模板。抽取 Agent 的系统提示词可以这样设计你是一个文档分析助手。用户会给你一篇 Markdown 文档内容。 请提取以下字段 - title: 文档标题字符串 - summary: 50 字以内简介 - steps: 由字符串组成的列表表示文档中的核心步骤 只输出 JSON不要输出其他文字。把文档内容拼到 user 消息里。这里的关键是让模型“只输出 JSON”并且字段结构要清晰。实际开发中模型偶尔还是会输出额外解释文字所以解析 JSON 时最好做容错。解析函数可以这样写import json def parse_model_json(raw): try: return json.loads(raw) except json.JSONDecodeError: start raw.find({) end raw.rfind(}) if start ! -1 and end ! -1: return json.loads(raw[start:end1]) raise ValueError(模型输出不包含合法 JSON)汇总 Agent 的系统提示词你是一个报告生成助手。下面是一批文档的结构化信息。 请生成一份 Markdown 格式的汇总报告包含 - 文档数量 - 每个文档的标题和简介 - 按主题合并后的核心步骤清单 报告需要有标题、目录、分节。工作流入口脚本大致是from plugins.file_tool import read_text_file from core.pipeline import Pipeline from core.llm import extract_from_documents result Pipeline(run_idgenerate_run_id()) \ .add_step(list_files, list_input_files) \ .add_step(read_docs, read_all_docs) \ .add_step(extract, extract_from_documents) \ .add_step(validate, validate_extracted) \ .add_step(generate_report, generate_markdown_report) \ .add_step(save, save_report) \ .run({input_dir: data/input, output_dir: data/output}) print(报告已生成:, result[report_path])这个例子里的具体函数你完全可以根据自己的项目补充。重要的是整体流程输入目录、读取、抽取、校验、汇总、输出。6.3 验证标准和资源观察工作流写完不要只看“跑没跑通”还要看质量和资源消耗。先准备三个结构差异比较大的 Markdown 文档最好包含标题、分节、列表。然后运行脚本重点观察每篇文档的 title 是否提取正确。summary 是否太长或太短。steps 是否准确反映了文档里的核心步骤。报告里的分节、目录是否清晰。失败文件有没有被跳过并记录。如果某个文档字段频繁缺失先看原文档是不是结构太乱或者提示词里字段定义不够清楚。此时建议先打印该文档的输入内容和模型原始输出定位问题发生在“模型理解”还是“解析逻辑”。资源占用也很重要。使用云端模型接口时要关注单文档消耗的 token 数和总耗时。可以在每个步骤里记录开始时间和 token 数。比如单文档抽取耗时为几秒到十几秒在常见网络环境下是正常的具体以你的模型服务商和文档大小为准。如果单条任务耗时超过几十秒要考虑是不是文档太长、上下文裁剪不够或者出现多次无效工具调用。如果使用本地模型还要额外关注显存。低配机器能跑通不代表能跑大批量任务。批量处理时建议把并发数降下来或者按顺序处理不要并发拉满。7. 排查链路启动失败、循环调用、输出异常按什么顺序查7.1 先看现象再动配置Agent 项目出问题时最容易犯的错是一上来就改参数、换模型、重装依赖。这种操作很可能让问题更隐蔽。我建议按照这样一个顺序排查先看现象再看输入再看环境再看参数最后看工具本身。现象是什么是程序直接抛异常还是模型返回空内容还是 Agent 一直循环不结束还是输出格式不对不同现象对应的排查路径完全不同。如果是启动就报错优先看第一行异常堆栈。ModuleNotFoundError 通常是依赖没装或虚拟环境没激活KeyError 通常是 .env 没加载或字段名拼错ConnectionError 通常是接口地址不可达。如果是模型接口调用报错要区分状态码401 类错误密钥不对检查 .env 是否存在、key 有没有多余空格。404 类错误接口地址或模型名不对。429 类错误触发限流降低并发或增加重试等待时间。7.2 常见报错与对应处理下面这张表是我在 Agent 项目中遇到频率比较高的问题按排查优先级排列。现象优先排查方向常见原因模型返回为空打印模型原始响应max_tokens 太小、内容被过滤、返回结构取错字段Agent 不调用工具检查 tools 参数是否传入工具描述不清晰、schema 格式不对、模型不支持函数调用Agent 反复调用同一工具检查工具结果是否回传消息格式错误、工具返回内容模型没收到、任务描述导致模型误判输出 JSON 解析失败打印模型原始输出提示词没有约束“只输出 JSON”、模型额外生成解释、上下文里示例不足上下文超长检查 messages 长度插件返回过长、历史轮数过多、文档未截断中文乱码检查文件编码和终端文件非 UTF-8 编码、Windows 终端编码未调这里我想重点说一个容易被忽略的问题工具调用循环卡死。很多时候模型会连续输出同一个工具请求比如每次都调用 read_text_file但传入的参数是一样的返回的结果也一样。程序看起来像一个死循环。这种情况下不要只加 MAX_ROUNDS还要考虑工具结果是否给模型提供了新信息。比如你可以缓存某路径文件的读取结果第二次请求同一个路径时直接返回“该文件已读取结果同上”节省上下文也打破循环。7.3 容易被忽略的边界条件最后补充几个和 Agent 本身无关、但经常导致项目跑不起来的边界条件。路径分隔符。Windows 是反斜杠Linux 和 macOS 是正斜杠。写插件时尽量不要拼死路径用 os.path.join 或 pathlib。当前工作目录。很多脚本从项目根目录执行没问题但改成从别的目录启动时相对路径全都失效。建议在入口脚本最顶部显式切换到项目根目录。文件编码。读取用户上传的文件时不要默认 UTF-8。很多 Windows 生成的文件是 GBK 编码建议先尝试 UTF-8失败后再尝试其他编码。日志没有落盘。批量任务跑几个小时如果日志只输出到控制台一旦终端关闭或内存缓冲区刷新中间信息就丢了。建议 logging 同时写到文件和控制台。模型调用重试的副作用。调用模型本身没有副作用但 Agent 循环里的工具函数可能有。比如一个发送消息的插件如果接口超时后自动重试接收方可能收到两条消息。处理方法是给每次工具调用生成唯一请求 ID在插件内部做去重。这些小细节看起来都不复杂但往往就是它们决定了项目能不能从“本机跑通”变成“稳定运行”。Agent 开发没有太多玄学。模型负责理解和生成你负责把任务拆清楚、把流程管住、把错误接住。先跑通一个最小闭环再逐步加工作流、加插件。等项目跑过几轮真实任务、看过了日志和失败记录你自然会知道下一步该优化哪里。