这次我们来看一套完整的“AI 大模型 Agent 开发实战”内容。它的主线非常明确用 Python 把大模型接入可执行代码、把 Agent 做成能自主规划任务的工作流、再通过插件机制延伸能力边界。不是只讲概念而是按“环境准备 - 开发框架 - 工作流搭建 - 插件开发 - API 集成 - 批量任务”的顺序走一遍项目实战。内容适合三类人正在做大模型应用开发的 Python 工程师、想在企业内部落地 AI 工作流的业务系统开发者以及准备系统学习 Agent 开发的初学者。这套实战里有一个非常务实的观点Agent 不是玄学它本质上就是“大模型 工具调用 任务状态管理”的组合。只要把这三件事拆开你完全可以用 Python 自己写出一个可运行的 Agent而不是只能依赖别人的 SaaS 平台。本文会把整套流程拆开讲清楚包括 Agent 的四大核心组件、三种不同复杂度的工作流实现方式、插件注册机制、API 暴露方式以及批量任务如何用异步队列接收和调度。全文中的代码示例都可以直接复制到你的项目里做骨架再按业务替换提示词和工具函数即可。要提前说明的一点是本文不会绑定某一个大模型厂商的专属 SDK。无论是 OpenAI 兼容接口、通义、文心还是本地部署的 Ollama、vLLM 服务只要暴露的是标准的 OpenAI Chat Completions 格式下面的 Agent 骨架都能直接对接。这也是当前大模型 Agent 开发最省事的做法先定接口协议再选模型供应商技术栈始终掌握在自己手里。1. 核心能力速览能力项说明项目类型Agent 开发实战教程覆盖大模型接入、工作流搭建、插件机制、API 与批量任务核心技术栈Python 3.10、OpenAI 兼容接口、LangChain 概念、异步任务队列大模型接入方式OpenAI 兼容接口、Ollama 本地模型、云端 API 均可是否支持本地部署支持但需根据所选模型7B/14B/72B确认本机显存与内存推荐硬件仅用云端 API 时无特殊要求本地推理建议 NVIDIA 显卡显存 12GB 起步主要功能对话 Agent、工具调用、工作流编排、插件注册、API 服务、批量任务是否支持 API支持通过 FastAPI 暴露 Agent 调用接口是否支持批量任务支持使用异步队列接收批量请求避免阻塞安装启动方式pip 安装依赖 Python 脚本启动适合场景企业知识库问答、自动化数据分析、内容批量生成、系统集成助手这里要区分一个常见误区Agent 开发不是非得在 LangChain 里做。你可以用 LangChain 这类框架也可以只用 Python 标准库加一个模型请求模块手写。本文为了讲清底层原理以手写骨架为主同时也会给出 LangChain 的对应思路。这样无论你最后选择框架还是自研都能理解 Agent 内部在做什么。2. 适用场景与使用边界2.1 适合谁用Agent 开发最适合的落地场景有三类对话式业务系统比如企业内部的工单助手、HR 问答机器人、销售培训助手。用户自然语言提问Agent 判断意图调用对应 API 拉取数据或执行操作。知识库问答把公司文档切分、向量化、存入检索库Agent 在回答用户问题前先检索资料再结合大模型生成答案避免模型“凭空编造”。自动化数据处理Excel 导入、日志分析、周报汇总、多源数据归并。Agent 将用户口语化指令翻译成 Python 函数调用把重复劳动自动化。2.2 不适合什么场景Agent 不是万能的。下面这几种场景建议先做技术验证再决定是否上 Agent强实时响应Agent 规划耗时通常要 1 到 5 秒多工具调用时更久。支付、登录、导航这类毫秒级交互不适合。强确定性逻辑库存扣减、订单状态流转这类业务必须用代码控制状态机不能让大模型自由发挥。低容错场景医疗诊断、法律结论、财务数字。如果 Agent 生成错误且没有人工复核风险很高。2.3 合规与安全边界使用大模型 Agent 时必须注意以下几点用户输入可能包含隐私数据。数据发送到大模型 API 前要确认供应商的数据协议并做好脱敏处理。Agent 调用内部 API 时要使用服务账号的最小权限不要使用个人管理员凭证。生成内容要保留人工审核入口尤其是对外发布的文案、客服回复、代码补丁。如果涉及图片、语音、人像等素材必须确认素材来源和授权禁止在未经授权的情况下处理他人肖像或声音。3. 技术选型与整体架构设计3.1 四类组件缺一不可一个可用的 Agent 系统至少包含四层模型层大模型负责意图理解、任务拆解、结果生成。可以选云端 API也能选本地模型。Agent 调度层接收用户目标维护任务列表决定下一步调用什么工具。工具层Agent 能操作的函数集合比如搜索、查数据库、发请求、执行 Python 脚本。服务层对外提供 HTTP API承接 Web 页面、IM 机器人、批量任务队列的请求。3.2 框架选择与对比方案定位适合项目上手难度手写 Python Agent 骨架最轻量无强绑定学习原理、中小型项目中LangChain / LangGraph生产级框架状态图清晰复杂 Agent、多分支流程高AutoGen多 Agent 协作研究型、多角色对话高Dify / Coze可视化工作流低代码搭建、运营人员低从学习和代码掌控的角度建议先实现一个手写骨架跑通后再决定要不要引入 LangChain。原因很简单手写一遍能让你理解“系统提示词、工具描述、函数执行、结果回填”这四个 Agent 核心动作遇到问题也知道去哪个环节排查。3.3 一个通用架构下面是不依赖任何可视化框架的推荐架构用户请求 └── HTTP API / 命令行 / 批量队列 └── Agent 调度器维护任务状态 ├── 大模型理解目标输出 JSON 动作 ├── 工具注册表路由到具体 Python 函数 └── 记忆模块会话历史、任务上下文 └── 执行结果回填给模型直到任务完成这个架构图用文字描述就是一个循环先给模型一个目标模型返回计划或工具调用程序执行工具把结果交回模型模型决定继续调用还是给出最终回答。实际代码里这个循环也就是一个几十行的run_agent()函数。4. 环境准备与前置条件4.1 软件环境建议使用以下环境组合操作系统Windows 10/11、macOS、Ubuntu 20.04 及以上Python3.10 或 3.11包管理pip venv 或 conda模型接口OpenAI 兼容服务或云端 APIWindows 用户尤其注意创建项目时建议单独建一个虚拟环境避免把依赖装进全局 Python 环境。4.2 依赖安装以下是本教程所需的核心依赖可以直接写入requirements.txtopenai1.30.0 fastapi0.110.0 uvicorn0.29.0 requests2.31.0 pydantic2.6.0 python-dotenv1.0.0安装命令pip install -r requirements.txt如果后续要接入向量检索、Excel 处理、定时调度再按需追加依赖pip install chromadb pandas celery4.3 模型服务准备文章里的代码示例均调用 OpenAI 兼容接口。以本地部署 Ollama 为例先确保服务已启动ollama pull qwen2.5:7b ollama serve如果使用云端 API配置好环境变量即可export OPENAI_API_BASEhttps://your-api-endpoint/v1 export OPENAI_API_KEYyour-api-key注意这里的OPENAI_API_BASE不一定只指 OpenAI 官方所有兼容该协议的供应商都可以。5. Agent 核心开发框架理解5.1 Agent 的本质一句话概括Agent 模型 能调用的函数 调度循环。传统程序是“代码决定流程”什么时候调什么函数是程序员写死的。Agent 则是“模型决定流程”用户输入目标后模型在候选工具列表里选择合适的函数并给出参数程序负责执行并回填结果。工具越多Agent 能做的事情越多同时出错的风险也越高。5.2 核心组件拆解在代码层面手写 Agent 需要四个组件系统提示词构建器给模型交代身份、可用工具、输出格式。工具注册表字典结构key 是工具名value 是函数和描述。模型客户端封装统一处理请求与响应解析。执行循环器根据模型返回的动作执行工具循环直到模型输出最终答案。5.3 提示词设计关键点给 Agent 的工具描述一定要写清楚以下信息工具做什么、什么时候用、参数含义、输出格式。因为大模型不是通过阅读你的源代码理解工具的它只能看到你写在提示词里的描述。一个推荐的工具描述模板工具名: search_knowledge_base 用途: 在内部知识库中检索与问题相关的文档片段 参数: query: str检索关键词或问题原文 输出: 匹配片段的文本列表这段描述会随系统提示词一起发给模型模型据此决定是否调用该工具。因此工具描述质量基本决定了 Agent 的决策准确程度。6. 工作流搭建实操工作流这块从简到繁给出三种实现方案。实际项目中根据任务复杂度选择即可。6.1 阶段一Prompt 串联工作流最简单的工作流是“多步 Prompt 串联”。比如生成周报先让模型阅读数据再让它整理结论最后生成固定模板。这种方式代码量最少适合文本处理类任务。def generate_report(raw_data: str) - str: step1_prompt f请阅读以下销售数据提取本周销售概况\\n{raw_data} step1_result call_llm(step1_prompt) step2_prompt f根据概况生成周报包含问题和下周计划\\n{step1_result} return call_llm(step2_prompt)这种方式没有 Agent 的自主性每一步是固定执行的适合流程确定的业务。6.2 阶段二函数调用工作流当流程中出现分支时就可以引入“意图识别 函数分发”。比如客服助手先判断用户是询问订单、退款还是物流再调用不同接口。def dispatch(intent: str, params: dict): if intent order_status: return query_order(params[order_id]) elif intent refund: return create_refund(params[order_id]) elif intent logistics: return query_logistics(params[order_id]) else: return 暂不支持该操作这种工作流本质上就是传统的意图识别 槽位填充在大模型时代依然适用而且稳定性很高。6.3 阶段三LLM 自主决策工作流完整 Agent 的核心是让模型自己决定调用哪个工具。这要求模型输出结构化动作比如 JSON{ action: search_knowledge_base, action_input: {query: 如何申请年假} }程序解析该 JSON执行函数拿到结果再返回给模型循环直到模型输出{action: final_answer}。这是当前 AI Agent 开发的通用模式无论是 LangChain 还是 AutoGen底层逻辑都类似。7. 插件开发与工具调用7.1 工具注册机制插件的本质就是“让 Agent 认识新函数”。用一个装饰器把函数注册到工具表中是最直接的实现方式。tools {} def register_tool(name: str, description: str): def decorator(func): tools[name] {func: func, description: description} return func return decorator register_tool(calculate, 计算两个数字的四则运算) def calculate(expression: str) - str: return str(eval(expression))7.2 把工具描述传给模型调用模型时把注册的工具逐个拼进系统提示词让模型知道可用工具。def build_system_prompt(): lines [你是智能助手可以根据需求调用工具。, 可用工具] for name, meta in tools.items(): lines.append(f- {name}: {meta[description]}) return \\n.join(lines)7.3 插件化开发建议随着工具数量增多建议按插件目录拆分plugins/ ├── __init__.py ├── excel_utils.py ├── http_tools.py └── db_tools.py每个插件模块负责注册自己的工具函数主程序启动时扫描并加载所有register_tool装饰器即可。这样可以保证 Agent 的工具集可插拔、可扩展。7.4 工具调用结果处理工具返回的内容不要太长。推荐将长文本截断保留前 1000 字避免占用大模型上下文窗口。如果工具失败也要返回包含错误信息的文本让模型决定是重试还是放弃。8. 实战项目Python 驱动的多工具 Agent下面给出一套可直接运行的 Agent 项目骨架。它包含三个工具获取当前时间、查询知识库模拟数据、执行简单计算。用户可以输入任意任务Agent 自主决定工具调用顺序。8.1 项目目录结构agent_demo/ ├── .env ├── requirements.txt ├── main.py ├── agent.py ├── tools.py └── api.py8.2 工具层tools.pyfrom datetime import datetime def get_current_time() - str: return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def search_knowledge_base(query: str) - str: mock_data { 年假: 员工每年享有 10 天年假需要提前 3 天申请。, 社保: 社保由公司统一缴纳比例为当地政策标准。, 加班: 加班需要提前在 OA 系统提交申请。, } for key in mock_data: if key in query: return mock_data[key] return 未找到相关资料。 def calculate(expression: str) - str: return str(eval(expression))8.3 Agent 核心agent.pyimport json import re from tools import get_current_time, search_knowledge_base, calculate TOOL_DESCRIPTIONS [ {name: get_current_time, description: 获取当前系统时间无参数, func: get_current_time}, {name: search_knowledge_base, description: 在知识库中搜索 query 对应内容, func: search_knowledge_base}, {name: calculate, description: 对数学表达式进行运算参数为 expression 字符串, func: calculate}, ] def call_llm(messages): # 这里替换为实际的模型服务调用 raise NotImplementedError(请接入 OpenAI 兼容接口) def run_agent(user_input: str, max_rounds: int 5) - str: system_prompt build_system_prompt() messages [{role: system, content: system_prompt}, {role: user, content: user_input}] for _ in range(max_rounds): response call_llm(messages) action_text response[content] action parse_action(action_text) if action is None: messages.append({role: assistant, content: action_text}) continue if action[name] final_answer: return action[args].get(answer, action_text) tool find_tool(action[name]) if tool is None: result f工具 {action[name]} 不存在 else: try: result tool[func](**action[args]) except Exception as e: result f工具执行失败{str(e)} messages.append({role: assistant, content: f调用 {action[name]}结果为{result}}) return 多次尝试后未完成任务 def parse_action(text: str): match re.search(r\\{.*?\\}, text, re.S) if not match: return None try: return json.loads(match.group()) except json.JSONDecodeError: return None def build_system_prompt(): lines [你是智能助手根据用户需求调用工具。, 输出格式为 JSON{\name\: \工具名\, \args\: {...}}, 工具列表] for t in TOOL_DESCRIPTIONS: lines.append(f- {t[name]}: {t[description]}) return \\n.join(lines) def find_tool(name): for t in TOOL_DESCRIPTIONS: if t[name] name: return t return None8.4 主入口main.pyfrom agent import run_agent if __name__ __main__: while True: user_input input( ) if user_input.lower() in (exit, quit): break result run_agent(user_input) print(Agent:, result)8.5 预期效果用户输入“今天是星期几顺便计算 23 * 45”Agent 会先调用get_current_time再调用calculate最后通过final_answer返回整合结果。整个过程不需要程序员提前写死调用顺序完全由模型根据用户目标动态决定。9. 接口 API 与批量任务9.1 暴露 HTTP API将上面的 Agent 包装成 FastAPI 服务即可接入 Web 页面或第三方系统。from fastapi import FastAPI from pydantic import BaseModel from agent import run_agent app FastAPI() class ChatRequest(BaseModel): message: str app.post(/api/chat) def chat(req: ChatRequest): result run_agent(req.message) return {reply: result}启动命令uvicorn api:app --host 127.0.0.1 --port 8000启动后发送请求验证curl -X POST http://127.0.0.1:8000/api/chat \\ -H Content-Type: application/json \\ -d {message: 今天几号}9.2 批量任务设计Agent 的单次调用有延时批量任务不能一个一个等。推荐方案用 Redis Celery 或简单的状态任务表把请求丢进队列Worker 异步执行再通过任务 ID 查询结果。# 伪代码示例 from celery import Celery app Celery(tasks, brokerredis://localhost:6379/0) app.task def run_agent_batch(task_id, message): result run_agent(message) save_result(task_id, result)调用方只需要一次提交多个任务然后轮询结果即可。批量任务中一定要加入日志和失败重试机制单个任务报错不要影响整个队列。10. 资源占用与性能观察10.1 本地模型 vs 云端 API本地部署大模型时显存占用是首要观察指标。以 7B 量级模型为例FP16 推理通常需要 14GB 左右显存4bit 量化后约 6GB14B 模型量化后约需 10GB 以上。实际数字因模型版本和推理框架不同会有浮动建议用nvidia-smi实测确认。watch -n 1 nvidia-smi10.2 如何降低显存与延迟使用量化模型加载例如q4_k_m、q8_0格式的 GGUF 文件能明显降低显存占用代价是生成质量略降。系统提示词和工具描述尽量精简减少每条请求往返的 token。工具结果截断避免长文本撑爆上下文。批量任务使用并发 Worker并限制同时运行的数量防止显存溢出。10.3 接口调用成本观察云端 API 模式下成本主要由 token 数决定。每次工具调用都会消耗输入 token因为工具描述和对话历史都会重新发送。优化方式是保存精简历史只保留最近 10 条消息避免上下文无限增长。11. 常见问题与排查方法问题现象可能原因排查方式解决方案依赖安装失败Python 版本过低或与包冲突检查python --version使用 Python 3.10 创建新虚拟环境模型请求超时本地模型推理慢或网络波动查看模型服务日志增加timeout参数使用异步调用Agent 调用工具格式错误模型输出 JSON 不规范打印原始输出文本在提示词中给示例使用宽容解析函数工具不存在工具名称拼写不一致检查注册表和提示词工具描述与代码函数名保持完全一致显存不足模型过大或并发太高观察nvidia-smi换量化模型或减少并发数端口被占用8000 端口已有服务lsof -i:8000换端口启动API 调用失败API Key 或 Base URL 配置错误打印环境变量核对.env文件批量任务卡住队列 Worker 未启动查看队列日志确认 Celery Worker 正常运行12. 最佳实践与使用建议12.1 项目工程化建议Agent 项目不要只写一个单文件脚本。建议按应用层、Agent 层、工具层拆分。工具层只做纯函数式实现Agent 层负责状态管理和决策循环应用层负责 HTTP 接口、命令行走入等外部交互。这样模型升级或者加工具时副作用最小。12.2 降低试错成本第一次搭建 Agent先把模型接口、工具调用、结果返回整条链路跑通再扩展业务工具。每次只加一个新工具验证无误后再加下一个。不要一次性注册 20 个工具模型决策混乱后很难排查是哪个描述写得不好。12.3 数据与权限安全Agent 能调工具也就意味着它有“手”。在真实业务中必须为 Agent 分配独立的、最小化的凭证不能让它有修改核心数据或绕过后台校验的能力。所有涉及用户隐私的操作要写入操作日志保留审批和回滚入口。12.4 输出质量复核无论是文本生成还是工具执行建议在正式输出前做一轮“答案校验”。可以写一个简单的校验函数检查必填字段是否齐全、数字格式是否正常、引用来源是否真实。这一步能拦截大量低级错误也方便后期迭代审核规则。13. 总结与下一步这篇实战内容最值得尝试的点是把 Agent 开发从“框架调包”降维成了三步走先理解工具注册机制再用循环调度器连接模型与工具最后用 FastAPI 把能力透出成标准接口。整套代码不依赖特定厂商掌握了换什么模型都能用。建议拿到代码后先做三件事第一把call_llm函数接上你的大模型接口验证最基本的对话能通第二用两个最熟悉的工具跑通工具调用循环第三给 Agent 增加“退出条件”和“最大轮数限制”防止死循环。这三步走通Agent 开发的完整链路就掌握了。最容易踩的坑有三个工具描述与函数实现不一致、模型输出 JSON 解析失败、没有限制最大循环轮数。这三个问题在真实项目和选型评估里几乎都会遇到提前在代码里加好容错能省掉大量排查时间。后续可以扩展的方向很多把工具调用中搜索到的资料片段加入向量缓存降低重复查询成本给批量任务加上优先级队列用 Graph 状态机替代单轮线性循环或者把当前单轮工具结果压缩功能升级成完整的记忆模块让 Agent 能跨会话记住用户偏好。想深入的话也可以选择 LangGraph、AutoGen 这类框架改写当前骨架逐步向生产级能力靠拢。如果你正准备在公司里试水 Agent 项目建议先把这套 Python 骨架跑通做成公司内部评估 Demo。它能帮你在一周内验证模型选型、工具调用、API 封装和批量任务是否满足业务预期。这套流程走完之后再决定自研还是引入框架选择就会清晰很多。
Python实战：从零构建AI大模型Agent工作流与插件系统
这次我们来看一套完整的“AI 大模型 Agent 开发实战”内容。它的主线非常明确用 Python 把大模型接入可执行代码、把 Agent 做成能自主规划任务的工作流、再通过插件机制延伸能力边界。不是只讲概念而是按“环境准备 - 开发框架 - 工作流搭建 - 插件开发 - API 集成 - 批量任务”的顺序走一遍项目实战。内容适合三类人正在做大模型应用开发的 Python 工程师、想在企业内部落地 AI 工作流的业务系统开发者以及准备系统学习 Agent 开发的初学者。这套实战里有一个非常务实的观点Agent 不是玄学它本质上就是“大模型 工具调用 任务状态管理”的组合。只要把这三件事拆开你完全可以用 Python 自己写出一个可运行的 Agent而不是只能依赖别人的 SaaS 平台。本文会把整套流程拆开讲清楚包括 Agent 的四大核心组件、三种不同复杂度的工作流实现方式、插件注册机制、API 暴露方式以及批量任务如何用异步队列接收和调度。全文中的代码示例都可以直接复制到你的项目里做骨架再按业务替换提示词和工具函数即可。要提前说明的一点是本文不会绑定某一个大模型厂商的专属 SDK。无论是 OpenAI 兼容接口、通义、文心还是本地部署的 Ollama、vLLM 服务只要暴露的是标准的 OpenAI Chat Completions 格式下面的 Agent 骨架都能直接对接。这也是当前大模型 Agent 开发最省事的做法先定接口协议再选模型供应商技术栈始终掌握在自己手里。1. 核心能力速览能力项说明项目类型Agent 开发实战教程覆盖大模型接入、工作流搭建、插件机制、API 与批量任务核心技术栈Python 3.10、OpenAI 兼容接口、LangChain 概念、异步任务队列大模型接入方式OpenAI 兼容接口、Ollama 本地模型、云端 API 均可是否支持本地部署支持但需根据所选模型7B/14B/72B确认本机显存与内存推荐硬件仅用云端 API 时无特殊要求本地推理建议 NVIDIA 显卡显存 12GB 起步主要功能对话 Agent、工具调用、工作流编排、插件注册、API 服务、批量任务是否支持 API支持通过 FastAPI 暴露 Agent 调用接口是否支持批量任务支持使用异步队列接收批量请求避免阻塞安装启动方式pip 安装依赖 Python 脚本启动适合场景企业知识库问答、自动化数据分析、内容批量生成、系统集成助手这里要区分一个常见误区Agent 开发不是非得在 LangChain 里做。你可以用 LangChain 这类框架也可以只用 Python 标准库加一个模型请求模块手写。本文为了讲清底层原理以手写骨架为主同时也会给出 LangChain 的对应思路。这样无论你最后选择框架还是自研都能理解 Agent 内部在做什么。2. 适用场景与使用边界2.1 适合谁用Agent 开发最适合的落地场景有三类对话式业务系统比如企业内部的工单助手、HR 问答机器人、销售培训助手。用户自然语言提问Agent 判断意图调用对应 API 拉取数据或执行操作。知识库问答把公司文档切分、向量化、存入检索库Agent 在回答用户问题前先检索资料再结合大模型生成答案避免模型“凭空编造”。自动化数据处理Excel 导入、日志分析、周报汇总、多源数据归并。Agent 将用户口语化指令翻译成 Python 函数调用把重复劳动自动化。2.2 不适合什么场景Agent 不是万能的。下面这几种场景建议先做技术验证再决定是否上 Agent强实时响应Agent 规划耗时通常要 1 到 5 秒多工具调用时更久。支付、登录、导航这类毫秒级交互不适合。强确定性逻辑库存扣减、订单状态流转这类业务必须用代码控制状态机不能让大模型自由发挥。低容错场景医疗诊断、法律结论、财务数字。如果 Agent 生成错误且没有人工复核风险很高。2.3 合规与安全边界使用大模型 Agent 时必须注意以下几点用户输入可能包含隐私数据。数据发送到大模型 API 前要确认供应商的数据协议并做好脱敏处理。Agent 调用内部 API 时要使用服务账号的最小权限不要使用个人管理员凭证。生成内容要保留人工审核入口尤其是对外发布的文案、客服回复、代码补丁。如果涉及图片、语音、人像等素材必须确认素材来源和授权禁止在未经授权的情况下处理他人肖像或声音。3. 技术选型与整体架构设计3.1 四类组件缺一不可一个可用的 Agent 系统至少包含四层模型层大模型负责意图理解、任务拆解、结果生成。可以选云端 API也能选本地模型。Agent 调度层接收用户目标维护任务列表决定下一步调用什么工具。工具层Agent 能操作的函数集合比如搜索、查数据库、发请求、执行 Python 脚本。服务层对外提供 HTTP API承接 Web 页面、IM 机器人、批量任务队列的请求。3.2 框架选择与对比方案定位适合项目上手难度手写 Python Agent 骨架最轻量无强绑定学习原理、中小型项目中LangChain / LangGraph生产级框架状态图清晰复杂 Agent、多分支流程高AutoGen多 Agent 协作研究型、多角色对话高Dify / Coze可视化工作流低代码搭建、运营人员低从学习和代码掌控的角度建议先实现一个手写骨架跑通后再决定要不要引入 LangChain。原因很简单手写一遍能让你理解“系统提示词、工具描述、函数执行、结果回填”这四个 Agent 核心动作遇到问题也知道去哪个环节排查。3.3 一个通用架构下面是不依赖任何可视化框架的推荐架构用户请求 └── HTTP API / 命令行 / 批量队列 └── Agent 调度器维护任务状态 ├── 大模型理解目标输出 JSON 动作 ├── 工具注册表路由到具体 Python 函数 └── 记忆模块会话历史、任务上下文 └── 执行结果回填给模型直到任务完成这个架构图用文字描述就是一个循环先给模型一个目标模型返回计划或工具调用程序执行工具把结果交回模型模型决定继续调用还是给出最终回答。实际代码里这个循环也就是一个几十行的run_agent()函数。4. 环境准备与前置条件4.1 软件环境建议使用以下环境组合操作系统Windows 10/11、macOS、Ubuntu 20.04 及以上Python3.10 或 3.11包管理pip venv 或 conda模型接口OpenAI 兼容服务或云端 APIWindows 用户尤其注意创建项目时建议单独建一个虚拟环境避免把依赖装进全局 Python 环境。4.2 依赖安装以下是本教程所需的核心依赖可以直接写入requirements.txtopenai1.30.0 fastapi0.110.0 uvicorn0.29.0 requests2.31.0 pydantic2.6.0 python-dotenv1.0.0安装命令pip install -r requirements.txt如果后续要接入向量检索、Excel 处理、定时调度再按需追加依赖pip install chromadb pandas celery4.3 模型服务准备文章里的代码示例均调用 OpenAI 兼容接口。以本地部署 Ollama 为例先确保服务已启动ollama pull qwen2.5:7b ollama serve如果使用云端 API配置好环境变量即可export OPENAI_API_BASEhttps://your-api-endpoint/v1 export OPENAI_API_KEYyour-api-key注意这里的OPENAI_API_BASE不一定只指 OpenAI 官方所有兼容该协议的供应商都可以。5. Agent 核心开发框架理解5.1 Agent 的本质一句话概括Agent 模型 能调用的函数 调度循环。传统程序是“代码决定流程”什么时候调什么函数是程序员写死的。Agent 则是“模型决定流程”用户输入目标后模型在候选工具列表里选择合适的函数并给出参数程序负责执行并回填结果。工具越多Agent 能做的事情越多同时出错的风险也越高。5.2 核心组件拆解在代码层面手写 Agent 需要四个组件系统提示词构建器给模型交代身份、可用工具、输出格式。工具注册表字典结构key 是工具名value 是函数和描述。模型客户端封装统一处理请求与响应解析。执行循环器根据模型返回的动作执行工具循环直到模型输出最终答案。5.3 提示词设计关键点给 Agent 的工具描述一定要写清楚以下信息工具做什么、什么时候用、参数含义、输出格式。因为大模型不是通过阅读你的源代码理解工具的它只能看到你写在提示词里的描述。一个推荐的工具描述模板工具名: search_knowledge_base 用途: 在内部知识库中检索与问题相关的文档片段 参数: query: str检索关键词或问题原文 输出: 匹配片段的文本列表这段描述会随系统提示词一起发给模型模型据此决定是否调用该工具。因此工具描述质量基本决定了 Agent 的决策准确程度。6. 工作流搭建实操工作流这块从简到繁给出三种实现方案。实际项目中根据任务复杂度选择即可。6.1 阶段一Prompt 串联工作流最简单的工作流是“多步 Prompt 串联”。比如生成周报先让模型阅读数据再让它整理结论最后生成固定模板。这种方式代码量最少适合文本处理类任务。def generate_report(raw_data: str) - str: step1_prompt f请阅读以下销售数据提取本周销售概况\\n{raw_data} step1_result call_llm(step1_prompt) step2_prompt f根据概况生成周报包含问题和下周计划\\n{step1_result} return call_llm(step2_prompt)这种方式没有 Agent 的自主性每一步是固定执行的适合流程确定的业务。6.2 阶段二函数调用工作流当流程中出现分支时就可以引入“意图识别 函数分发”。比如客服助手先判断用户是询问订单、退款还是物流再调用不同接口。def dispatch(intent: str, params: dict): if intent order_status: return query_order(params[order_id]) elif intent refund: return create_refund(params[order_id]) elif intent logistics: return query_logistics(params[order_id]) else: return 暂不支持该操作这种工作流本质上就是传统的意图识别 槽位填充在大模型时代依然适用而且稳定性很高。6.3 阶段三LLM 自主决策工作流完整 Agent 的核心是让模型自己决定调用哪个工具。这要求模型输出结构化动作比如 JSON{ action: search_knowledge_base, action_input: {query: 如何申请年假} }程序解析该 JSON执行函数拿到结果再返回给模型循环直到模型输出{action: final_answer}。这是当前 AI Agent 开发的通用模式无论是 LangChain 还是 AutoGen底层逻辑都类似。7. 插件开发与工具调用7.1 工具注册机制插件的本质就是“让 Agent 认识新函数”。用一个装饰器把函数注册到工具表中是最直接的实现方式。tools {} def register_tool(name: str, description: str): def decorator(func): tools[name] {func: func, description: description} return func return decorator register_tool(calculate, 计算两个数字的四则运算) def calculate(expression: str) - str: return str(eval(expression))7.2 把工具描述传给模型调用模型时把注册的工具逐个拼进系统提示词让模型知道可用工具。def build_system_prompt(): lines [你是智能助手可以根据需求调用工具。, 可用工具] for name, meta in tools.items(): lines.append(f- {name}: {meta[description]}) return \\n.join(lines)7.3 插件化开发建议随着工具数量增多建议按插件目录拆分plugins/ ├── __init__.py ├── excel_utils.py ├── http_tools.py └── db_tools.py每个插件模块负责注册自己的工具函数主程序启动时扫描并加载所有register_tool装饰器即可。这样可以保证 Agent 的工具集可插拔、可扩展。7.4 工具调用结果处理工具返回的内容不要太长。推荐将长文本截断保留前 1000 字避免占用大模型上下文窗口。如果工具失败也要返回包含错误信息的文本让模型决定是重试还是放弃。8. 实战项目Python 驱动的多工具 Agent下面给出一套可直接运行的 Agent 项目骨架。它包含三个工具获取当前时间、查询知识库模拟数据、执行简单计算。用户可以输入任意任务Agent 自主决定工具调用顺序。8.1 项目目录结构agent_demo/ ├── .env ├── requirements.txt ├── main.py ├── agent.py ├── tools.py └── api.py8.2 工具层tools.pyfrom datetime import datetime def get_current_time() - str: return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def search_knowledge_base(query: str) - str: mock_data { 年假: 员工每年享有 10 天年假需要提前 3 天申请。, 社保: 社保由公司统一缴纳比例为当地政策标准。, 加班: 加班需要提前在 OA 系统提交申请。, } for key in mock_data: if key in query: return mock_data[key] return 未找到相关资料。 def calculate(expression: str) - str: return str(eval(expression))8.3 Agent 核心agent.pyimport json import re from tools import get_current_time, search_knowledge_base, calculate TOOL_DESCRIPTIONS [ {name: get_current_time, description: 获取当前系统时间无参数, func: get_current_time}, {name: search_knowledge_base, description: 在知识库中搜索 query 对应内容, func: search_knowledge_base}, {name: calculate, description: 对数学表达式进行运算参数为 expression 字符串, func: calculate}, ] def call_llm(messages): # 这里替换为实际的模型服务调用 raise NotImplementedError(请接入 OpenAI 兼容接口) def run_agent(user_input: str, max_rounds: int 5) - str: system_prompt build_system_prompt() messages [{role: system, content: system_prompt}, {role: user, content: user_input}] for _ in range(max_rounds): response call_llm(messages) action_text response[content] action parse_action(action_text) if action is None: messages.append({role: assistant, content: action_text}) continue if action[name] final_answer: return action[args].get(answer, action_text) tool find_tool(action[name]) if tool is None: result f工具 {action[name]} 不存在 else: try: result tool[func](**action[args]) except Exception as e: result f工具执行失败{str(e)} messages.append({role: assistant, content: f调用 {action[name]}结果为{result}}) return 多次尝试后未完成任务 def parse_action(text: str): match re.search(r\\{.*?\\}, text, re.S) if not match: return None try: return json.loads(match.group()) except json.JSONDecodeError: return None def build_system_prompt(): lines [你是智能助手根据用户需求调用工具。, 输出格式为 JSON{\name\: \工具名\, \args\: {...}}, 工具列表] for t in TOOL_DESCRIPTIONS: lines.append(f- {t[name]}: {t[description]}) return \\n.join(lines) def find_tool(name): for t in TOOL_DESCRIPTIONS: if t[name] name: return t return None8.4 主入口main.pyfrom agent import run_agent if __name__ __main__: while True: user_input input( ) if user_input.lower() in (exit, quit): break result run_agent(user_input) print(Agent:, result)8.5 预期效果用户输入“今天是星期几顺便计算 23 * 45”Agent 会先调用get_current_time再调用calculate最后通过final_answer返回整合结果。整个过程不需要程序员提前写死调用顺序完全由模型根据用户目标动态决定。9. 接口 API 与批量任务9.1 暴露 HTTP API将上面的 Agent 包装成 FastAPI 服务即可接入 Web 页面或第三方系统。from fastapi import FastAPI from pydantic import BaseModel from agent import run_agent app FastAPI() class ChatRequest(BaseModel): message: str app.post(/api/chat) def chat(req: ChatRequest): result run_agent(req.message) return {reply: result}启动命令uvicorn api:app --host 127.0.0.1 --port 8000启动后发送请求验证curl -X POST http://127.0.0.1:8000/api/chat \\ -H Content-Type: application/json \\ -d {message: 今天几号}9.2 批量任务设计Agent 的单次调用有延时批量任务不能一个一个等。推荐方案用 Redis Celery 或简单的状态任务表把请求丢进队列Worker 异步执行再通过任务 ID 查询结果。# 伪代码示例 from celery import Celery app Celery(tasks, brokerredis://localhost:6379/0) app.task def run_agent_batch(task_id, message): result run_agent(message) save_result(task_id, result)调用方只需要一次提交多个任务然后轮询结果即可。批量任务中一定要加入日志和失败重试机制单个任务报错不要影响整个队列。10. 资源占用与性能观察10.1 本地模型 vs 云端 API本地部署大模型时显存占用是首要观察指标。以 7B 量级模型为例FP16 推理通常需要 14GB 左右显存4bit 量化后约 6GB14B 模型量化后约需 10GB 以上。实际数字因模型版本和推理框架不同会有浮动建议用nvidia-smi实测确认。watch -n 1 nvidia-smi10.2 如何降低显存与延迟使用量化模型加载例如q4_k_m、q8_0格式的 GGUF 文件能明显降低显存占用代价是生成质量略降。系统提示词和工具描述尽量精简减少每条请求往返的 token。工具结果截断避免长文本撑爆上下文。批量任务使用并发 Worker并限制同时运行的数量防止显存溢出。10.3 接口调用成本观察云端 API 模式下成本主要由 token 数决定。每次工具调用都会消耗输入 token因为工具描述和对话历史都会重新发送。优化方式是保存精简历史只保留最近 10 条消息避免上下文无限增长。11. 常见问题与排查方法问题现象可能原因排查方式解决方案依赖安装失败Python 版本过低或与包冲突检查python --version使用 Python 3.10 创建新虚拟环境模型请求超时本地模型推理慢或网络波动查看模型服务日志增加timeout参数使用异步调用Agent 调用工具格式错误模型输出 JSON 不规范打印原始输出文本在提示词中给示例使用宽容解析函数工具不存在工具名称拼写不一致检查注册表和提示词工具描述与代码函数名保持完全一致显存不足模型过大或并发太高观察nvidia-smi换量化模型或减少并发数端口被占用8000 端口已有服务lsof -i:8000换端口启动API 调用失败API Key 或 Base URL 配置错误打印环境变量核对.env文件批量任务卡住队列 Worker 未启动查看队列日志确认 Celery Worker 正常运行12. 最佳实践与使用建议12.1 项目工程化建议Agent 项目不要只写一个单文件脚本。建议按应用层、Agent 层、工具层拆分。工具层只做纯函数式实现Agent 层负责状态管理和决策循环应用层负责 HTTP 接口、命令行走入等外部交互。这样模型升级或者加工具时副作用最小。12.2 降低试错成本第一次搭建 Agent先把模型接口、工具调用、结果返回整条链路跑通再扩展业务工具。每次只加一个新工具验证无误后再加下一个。不要一次性注册 20 个工具模型决策混乱后很难排查是哪个描述写得不好。12.3 数据与权限安全Agent 能调工具也就意味着它有“手”。在真实业务中必须为 Agent 分配独立的、最小化的凭证不能让它有修改核心数据或绕过后台校验的能力。所有涉及用户隐私的操作要写入操作日志保留审批和回滚入口。12.4 输出质量复核无论是文本生成还是工具执行建议在正式输出前做一轮“答案校验”。可以写一个简单的校验函数检查必填字段是否齐全、数字格式是否正常、引用来源是否真实。这一步能拦截大量低级错误也方便后期迭代审核规则。13. 总结与下一步这篇实战内容最值得尝试的点是把 Agent 开发从“框架调包”降维成了三步走先理解工具注册机制再用循环调度器连接模型与工具最后用 FastAPI 把能力透出成标准接口。整套代码不依赖特定厂商掌握了换什么模型都能用。建议拿到代码后先做三件事第一把call_llm函数接上你的大模型接口验证最基本的对话能通第二用两个最熟悉的工具跑通工具调用循环第三给 Agent 增加“退出条件”和“最大轮数限制”防止死循环。这三步走通Agent 开发的完整链路就掌握了。最容易踩的坑有三个工具描述与函数实现不一致、模型输出 JSON 解析失败、没有限制最大循环轮数。这三个问题在真实项目和选型评估里几乎都会遇到提前在代码里加好容错能省掉大量排查时间。后续可以扩展的方向很多把工具调用中搜索到的资料片段加入向量缓存降低重复查询成本给批量任务加上优先级队列用 Graph 状态机替代单轮线性循环或者把当前单轮工具结果压缩功能升级成完整的记忆模块让 Agent 能跨会话记住用户偏好。想深入的话也可以选择 LangGraph、AutoGen 这类框架改写当前骨架逐步向生产级能力靠拢。如果你正准备在公司里试水 Agent 项目建议先把这套 Python 骨架跑通做成公司内部评估 Demo。它能帮你在一周内验证模型选型、工具调用、API 封装和批量任务是否满足业务预期。这套流程走完之后再决定自研还是引入框架选择就会清晰很多。
