ARTICLE DETAIL

建站实战干货

来自一线的建站与推广经验沉淀,每一条都经过真实交付验证。

AI Agent工作流核心原理与Python最小实现

2026/8/30 19:35:42 拓冰建站 浏览量
AI Agent工作流核心原理与Python最小实现 Manus 这类通用 AI Agent 产品走红之后很多开发者的第一反应是“这不就是调大模型吗”但真正动手复现一个最小版本时才会发现事情没有那么简单。一个能自主规划、调用工具、读取结果、继续执行的 Agent核心不是某一次 Prompt 写得好而是一套完整的工程回路任务拆解、工具注册、上下文维护、结果验证、失败重试和终止条件。这篇博客会从概念开始讲清楚 Agent 工作流的基本结构然后搭建一个 Python 最小项目实现一个类 Manus 的轻量 Agent 循环最后给出运行验证、常见问题排查和生产化建议。1. Manus 走红的本质Agent 不是“多轮聊天”而是“计划-执行-验证”回路1.1 传统聊天与 Agent 工作流的区别传统聊天机器人通常只做一件事把用户问题丢给大模型拿到文字回复后直接展示。这个模式适合问答、写作、翻译但它不具备“做事”的能力。Agent 工作流则不同。Agent 会把目标当成一个需要完成的任务先拆解成多个步骤然后逐步执行。每一步可能需要调用外部工具比如搜索、计算、操作浏览器、读文件、写数据库执行完之后模型还要根据工具返回的结果决定下一步做什么直到任务完成或达到安全边界。对比维度传统聊天Agent 工作流输入一轮用户问题一个目标可能包含多个隐含步骤输出一段文字一串动作和最终结果是否调用工具通常不调用按需调用工具结果参与决策状态维护只维护对话历史维护任务计划、中间结果、工具记录容错方式答错就重新问工具失败后重试、改路径、终止典型场景客服问答、内容生成数据分析、自动填表、执行多步操作Manus 之所以让人印象深刻是因为它把“看到结果后继续行动”这个过程做成了产品体验。从工程角度看这种体验背后就是一个循环模型产出动作系统执行动作观察结果再把结果交回模型模型继续产出下一个动作。1.2 为什么 Agent 需要工具、记忆和循环控制一个完整的 Agent至少要包含三样东西。第一是工具。没有工具的大模型只能输出文字无法改变外部世界。工具可以是函数、API、数据库接口或浏览器操作。工具把模型和业务系统连接起来Agent 才具备实际完成任务的能力。第二是记忆。Agent 在多次工具调用之间必须知道“我之前已经查到了什么”“哪个步骤已经完成”。在目前的大模型架构下最直接的记忆载体就是对话历史消息列表。每一步模型生成的内容、工具返回的结果都追加到消息里模型在下一轮就能看到上下文。第三是循环控制。Agent 不能无限制地调用工具。如果模型陷入重复调用或者工具一直失败系统必须能够在指定的最大迭代次数内停止并返回当前进度和失败信息。缺少循环控制Agent 在生产环境里会变成“费用黑洞”和“死循环机器”。1.3 一个最小 Agent 回路的五个阶段任何 Agent 都可以抽象成下面这条链路任务输入 - 模型规划 - 产生工具调用 - 执行工具 - 返回观察结果 - 模型再规划 - ... - 终止具体到代码实现这条链路可以拆成五个阶段组装系统提示词和任务输入。调用大模型获得结构化动作。解析动作。如果是工具调用执行对应函数。把工具返回值追加到消息列表。判断是否终止如果未终止则回到第 2 步。只要把这条链路实现一遍你就掌握了 Agent 最核心的骨架。Manus 比这个复杂的地方在于它加入了更多工具、更长的规划、文件系统和浏览器能力但底层循环是一致的。2. 搭建一个类 Manus 的轻量 Agent 环境2.1 技术选型和运行环境这个项目不追求复刻 Manus 的完整能力只实现最小 Agent 闭环。技术选型上我选择 Python 和openaiSDK因为当前大多数模型服务都提供 OpenAI 兼容接口便于替换模型供应商。运行环境建议如下项目建议配置Python3.10 或更高版本包管理pip 或 poetry模型服务OpenAI 兼容的 Chat Completions 接口开发调试本地命令行即可暂不需要 Web 服务可选依赖python-dotenv 用于加载环境变量如果你的项目会用到浏览器自动化可以把playwright作为后续扩展。本文第一阶段只做基础工具调用不引入浏览器降低环境复杂度。2.2 准备 Python 项目和依赖在本地创建项目目录并初始化虚拟环境。mkdir mini-agent cd mini-agent python -m venv .venv source .venv/bin/activate然后安装依赖。pip install openai python-dotenvopenai用于调用大模型python-dotenv用于从.env文件加载 API Key。建议把.env文件加入.gitignore避免密钥被提交到代码仓库。如果你的模型服务不是 OpenAI 官方服务而是兼容接口可以设置base_url。示例代码会预留这个参数。2.3 配置文件和环境变量在项目根目录创建.env文件LLM_API_KEY你的密钥 LLM_BASE_URLhttps://api.openai.com/v1 LLM_MODELgpt-4o-mini这里有几个注意点。LLM_API_KEY是访问模型服务必需的凭证。不要把它硬编码在代码里也不要写进测试用例的公开输出。LLM_BASE_URL用于切换模型供应商如果使用 OpenAI 官方服务可以保留默认值。LLM_MODEL选择支持工具调用能力的模型gpt-4o-mini是示例实际项目要以你的模型服务实际支持的型号为准。学习环境里直接读取环境变量就能跑通但生产环境通常还会使用密钥管理服务或 K8s Secret。这个差异后面会单独说明。2.4 项目目录结构最小项目可以保持扁平便于理解。mini-agent/ ├── .env ├── .gitignore ├── requirements.txt ├── agent.py ├── tools.py └── main.py各文件职责如下文件职责agent.pyAgent 主循环、消息组装、终止判断tools.py工具定义和工具执行器main.py入口读取任务并启动 Agentrequirements.txtPython 依赖列表.env模型服务和密钥配置3. 写一个最小可运行的 Agent 循环3.1 定义消息结构和工具描述在tools.py中定义工具的执行逻辑。为了演示我实现两个工具一个是四则运算计算器一个是模拟知识查询。真实项目中工具可以替换成搜索接口、数据库查询或内部 API。# tools.py import json def calculator(expression: str) - str: 执行一个简单的四则运算表达式只允许数字、、-、*、/、括号和空格。 allowed set(0123456789-*/(). ) if not set(expression).issubset(allowed): return json.dumps({error: expression contains invalid characters}) try: result eval(expression, {__builtins__: {}}, {}) return json.dumps({result: result}) except Exception as exc: return json.dumps({error: str(exc)}) def query_knowledge(question: str) - str: 模拟一次知识库查询实际项目可替换为搜索 API 或内部文档检索。 data { manus: Manus 是一款通用 AI Agent 产品核心能力是自主规划、调用工具、多步执行任务。, agent: Agent 是能够感知环境并采取行动达成目标的程序通常结合大模型和工具完成复杂任务。, } for key, value in data.items(): if key in question.lower(): return json.dumps({answer: value}) return json.dumps({answer: 未找到相关知识请尝试其他问题。})eval在真实项目中直接使用有安全风险这里只是最小演示。生产环境请使用ast或专门的计算库并严格控制输入范围。接下来定义工具描述也就是给模型看的 JSON Schema。模型会根据这些描述决定调用哪个工具以及传入什么参数。# tools.py TOOL_CALCULATOR { type: function, function: { name: calculator, description: 计算四则运算表达式。, parameters: { type: object, properties: { expression: { type: string, description: 要计算的数学表达式例如 1 2 * 3, } }, required: [expression], }, }, } TOOL_QUERY_KNOWLEDGE { type: function, function: { name: query_knowledge, description: 查询内建知识库用于解释名词或背景。, parameters: { type: object, properties: { question: { type: string, description: 要查询的问题, } }, required: [question], }, }, } TOOLS [TOOL_CALCULATOR, TOOL_QUERY_KNOWLEDGE]工具描述里的字段不是随便写的。description会直接影响模型是否能正确选择工具写得太含糊模型会不知道该调用谁参数定义不清晰模型就会生成错误参数。3.2 实现 LLM 调用层agent.py负责调用模型。先加载环境变量再创建客户端。# agent.py import os from openai import OpenAI from tools import TOOLS def create_client() - OpenAI: return OpenAI( api_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL) or None, )base_url在.env中已有值因此使用os.getenv(LLM_BASE_URL)。如果某些本地服务不需要 API Key也可以把api_key设为占位字符串。3.3 实现工具执行器Agent 拿到模型返回的工具调用后需要根据function.name找到对应函数并执行。为了避免直接使用全局函数名我建立一个名字到函数的映射。# agent.py import json from tools import calculator, query_knowledge TOOL_FUNCTIONS { calculator: calculator, query_knowledge: query_knowledge, } def execute_tool_call(tool_call) - str: function_name tool_call.function.name arguments json.loads(tool_call.function.arguments or {}) if function_name not in TOOL_FUNCTIONS: return json.dumps({error: funknown tool: {function_name}}) func TOOL_FUNCTIONS[function_name] return func(**arguments)这里的关键点是把arguments由 JSON 字符串解析成 Python 字典再按参数名展开。如果模型生成的参数缺少必填项工具函数会抛出TypeError主循环需要把异常转换成工具返回信息避免整个程序崩溃。3.4 实现主循环主循环是整个 Agent 的心脏。它要做四件事调用模型携带完整历史消息和工具描述。判断模型是否要求调用工具。如果调用工具执行工具并追加结果消息。如果不调用工具说明模型认为任务已完成输出最终回答。# agent.py SYSTEM_PROMPT 你是一个通用 AI Agent。 你的任务是根据用户给出的目标逐步完成操作。 你可以调用工具获取计算结果或查询知识。 当工具结果不满足要求时可以尝试修改参数重新调用。 当任务已经完成时直接输出最终答案不要再调用工具。 def run_agent(task: str, max_iterations: int 5): client create_client() messages [ {role: system, content: SYSTEM_PROMPT}, {role: user, content: task}, ] for step in range(1, max_iterations 1): print(f\n[Step {step}] calling model...) response client.chat.completions.create( modelos.getenv(LLM_MODEL), messagesmessages, toolsTOOLS, ) message response.choices[0].message messages.append( { role: assistant, content: message.content or , tool_calls: [ { id: tc.id, type: function, function: { name: tc.function.name, arguments: tc.function.arguments, }, } for tc in (message.tool_calls or []) ] if message.tool_calls else None, } ) if not message.tool_calls: print([Agent Final], message.content) return message.content for tool_call in message.tool_calls: print(f[Tool] {tool_call.function.name}({tool_call.function.arguments})) result execute_tool_call(tool_call) print(f[Tool Result] {result}) messages.append( { role: tool, tool_call_id: tool_call.id, content: result, } ) print([Agent] max iterations reached, stopping.) return 已达到最大迭代次数任务未能完整执行。 if __name__ __main__: run_agent(帮我计算 (1 2) * 4 的结果然后用知识库工具查询一下 Agent 是什么意思。)上面代码有两个容易被忽略的细节。第一追加 assistant 消息时必须保留tool_calls字段。模型和工具结果之间的关联靠tool_call_id完成如果丢掉tool_calls后续请求会触发校验错误。第二追加 tool 消息时content必须是字符串。如果工具返回的是列表或字典必须先转换成 JSON 字符串。3.5 完整入口文件main.py只需要读取任务并启动。# main.py from agent import run_agent if __name__ __main__: task input(请输入任务).strip() if not task: task 帮我计算 3 * 7然后查询 Manus 是什么。 run_agent(task)运行方式python main.py4. 关键机制拆解为什么这样设计4.1 工具描述如何驱动模型生成参数在传统函数调用里参数由调用方写死在 Agent 里参数是由大模型根据用户任务和工具描述动态生成的。这意味着工具描述本身必须像一份“接口文档”一样精确。以calculator工具为例模型收到“计算 (1 2) * 4”这个任务时会从参数 schema 中知道需要提供expression字符串并且这个字符串应该是合法数学表达式。如果description写得太短比如只写“计算”模型可能生成expression: (12)*4也可能生成expression: 计算(12)*4后者就会导致工具执行失败。实际项目里给工具写描述时建议包含这个工具做什么。什么场景下应该调用。每个参数如何填写。有没有格式限制。4.2 多轮上下文如何形成短期记忆messages列表就是 Agent 的短期记忆。每一轮对话都包含四类消息角色作用system设定 Agent 行为规范user用户原始任务assistant模型上一轮的思考结果或工具调用意图tool工具执行后的观察结果模型每生成一次回复都会基于完整的消息列表。这相当于让它“记得”自己已经执行过哪些工具以及工具返回结果是什么。只要消息列表没有被截断Agent 就能继续推进任务。4.3 最大迭代次数和终止条件max_iterations是这个最小项目中最重要的安全阀。一般来说Agent 会一直循环到模型不再发起工具调用为止但模型可能因为任务复杂而持续调用工具也可能因为 Prompt 写得不好而陷入重复调用。如果去掉最大迭代次数可能遇到的情况包括模型反复调用同一个失败工具。模型在多个工具之间来回尝试。任务本身没有明确终态。一次运行产生大量 token费用不可控。生产环境中除了最大迭代次数还应该加入运行超时、单次工具执行超时和每日费用上限。4.4 错误分支工具失败如何回传工具执行结果不一定都是成功。在calculator中如果表达式非法函数返回包含error的 JSON 字符串。这个错误信息会被追加到消息列表里模型会在下一轮看到从而决定是修改参数重试还是停止调用工具。这种“错误也作为观察结果回传”的设计是 Agent 具备自我修复能力的基础。代码中不要把工具异常直接抛出到整个进程而应该捕获后转成结构化返回。否则模型无法感知错误也无法自行调整策略。5. 运行验证与结果分析5.1 运行后的预期输出执行python main.py后输入帮我计算 (1 2) * 4 的结果然后用知识库工具查询一下 Agent 是什么意思。如果一切正常你应该看到类似下面的流程[Step 1] calling model... [Tool] calculator({expression:(1 2) * 4}) [Tool Result] {result: 12} [Step 2] calling model... [Tool] query_knowledge({question:Agent 是什么意思}) [Tool Result] {answer: Agent 是能够感知环境并采取行动达成目标的程序通常结合大模型和工具完成复杂任务。} [Step 3] calling model... [Agent Final] 计算结果为 12。Agent 是指能够感知环境并采取行动达成目标的程序通常结合大模型和工具完成复杂任务。这段输出说明 Agent 完成了三步计算、查询、汇总。如果 Step 3 变成了再次调用calculator说明系统提示词和工具描述对终止条件的约束还不够清晰。5.2 如何验证 Agent 真的在“按计划执行”验证一个 Agent 不能只看最终回答是否正确还要看执行路径是否合理。建议关注以下几点。工具调用顺序是否符合直觉。每次工具参数是否能被工具正常解析。工具结果是否被模型正确引用而不是答非所问。是否及时终止没有多余循环。工具失败后模型是否尝试修正参数。可以把[Step]、[Tool]、[Tool Result]的日志输出保存到文件便于复盘。更专业的做法是引入 trace把消息列表、工具调用时间、耗时和 token 消耗都记录下来。5.3 学习环境与生产环境的差异当前代码适合在本地学习但直接搬到生产环境会有明显问题。关注项学习环境生产环境密钥管理.env文件密钥管理服务或 K8s Secret错误处理打印日志结构化日志、告警、追踪工具安全最小演示白名单、权限校验、参数校验并发控制单任务串行队列、限流、超时成本控制手动看日志token 统计、费用预算、配额状态存储内存消息列表数据库或 Redis 持久化模型版本固定模型灰度、回滚、多模型切换5.4 参数调优速查表参数作用调小的风险调大的风险max_iterations控制最大循环次数复杂任务被截断死循环带来高费用temperature控制输出随机性回答过于保守工具参数不稳定max_tokens限制单次生成长度输出被截断无效等待变长工具description长度帮助模型理解工具模型选错工具占用上下文 token6. 常见问题排查链路6.1 模型总是返回空内容不调用工具现象日志里只有[Agent Final]但内容为空。可能原因系统提示词让模型认为可以直接回答。模型本身不支持工具调用。tools参数没有传对或格式不符合模型要求。检查顺序确认tools参数传入的是列表。确认模型型号支持tool_calls。在系统提示词中说明“如果需要工具请先调用工具”。输出message.tool_calls日志确认模型是否返回了调用。6.2 模型不调用工具直接把答案编出来现象任务需要查询知识库但模型只输出文字没有调用query_knowledge。可能原因工具描述没有说明“必须调用才能获取答案”。模型认为自己的内建知识足够回答。工具名称和任务描述不匹配。处理方式在工具描述里加“该问题必须调用本工具才能回答”。在系统提示词中强调不要编造工具结果工具结果必须来自真实调用。6.3 同一个工具被反复调用现象calculator被连续调用多次参数几乎一样。可能原因工具返回结果包含模型不理解的格式。模型没有在下一轮看到工具结果。tool_call_id或消息顺序错误。检查方式打印每轮messages[-2:]确认 tool 结果是否已经被追加。确认 tool 消息的content是字符串而非对象。在 message 中检查 assistant 消息是否保留了tool_calls。这类问题在 OpenAI 兼容接口中尤其常见只要tool_call_id对不上服务端就会报错如果服务端校验不严格模型可能表现得像“失忆”一样。6.4 上下文越来越长费用快速上涨现象Agent 运行 5 步后请求体变得很大。可能原因每次循环都把完整消息列表发送给模型。工具返回大块文本并一直保留。max_iterations设置过高。处理方式对工具结果做摘要只保留关键字段。对历史消息做截断或压缩。对大任务拆成多个子 Agent而不是让一个 Agent 无限循环。6.5 API 超时或限流现象请求抛出Timeout或RateLimitError。可能原因模型服务负载高。请求 token 过多响应时间超过客户端超时时间。并发任务过多。处理方式增加客户端超时时间。增加重试策略但要带指数退避。控制并发数引入队列。调整模型或降低单次max_tokens。注意不要在生产环境直接对所有异常无限重试。超时、限流和参数错误需要的处理方式不同重试前要区分错误类型。7. 生产级 Agent 的最佳实践与扩展方向7.1 上线前检查清单从 demo 到线上建议先过一遍下面的清单。工具是否只暴露必要能力是否做了权限校验。工具参数是否经过严格校验避免注入类风险。Agent 是否设置了最大迭代次数、超时和预算上限。每个步骤是否都有结构化日志。模型调用失败时是否有降级方案。工具结果是否可能包含敏感数据是否做了脱敏。消息列表是否设置了长度上限和摘要策略。是否评估过单次任务的平均 token 成本和最坏成本。7.2 从 demo 到生产的六个改造点第一个改造点是工具化。把工具从普通函数升级为带有超时、鉴权、审计和幂等控制的服务。工具返回值需要统一为 JSON 结构至少要包含success、data、error三个字段。第二个改造点是状态持久化。当前消息列表只存在内存中进程重启就丢失。生产环境可以把消息、任务状态和工具调用记录存到数据库这样既支持断点续跑也便于排查问题。第三个改造点是任务规划。当前最小版本是完全靠模型自由发挥。生产环境可以引入“规划器”在任务开始时先让模型产出阶段计划再逐步执行每一个阶段都有独立的验证条件。第四个改造点是结果验证。不要默认工具结果一定正确。可以在工具执行后增加校验器比如检查计算结果的类型、检查数据库查询是否返回空值。校验失败可以触发重试。第五个改造点是成本控制。用 token 计数器和费用预算限制单次任务的上限超过阈值立即终止。还可以对工具调用次数设置独立上限避免某个工具被高频调用。第六个改造点是可观测性。除了打印日志还要记录 trace_id、耗时、模型名、token 消耗、工具调用顺序。如果使用 OpenTelemetry可以把 Agent 的每一步都封装成 span方便在链路追踪系统里查看。7.3 可以继续学习的方向如果完成了本文的最小 Agent下一步可以从这几个方向深入。一是多 Agent 协作。一个复杂的分析任务可以由规划 Agent、工具 Agent、审查 Agent 共同完成每个 Agent 职责单一通过消息队列或共享状态协调。二是工具协议的标准化。调研 Function Calling、MCP 等协议理解工具描述如何跨平台复用。MCP 这类标准可以让 Agent 以统一方式发现和调用外部工具。三是长期记忆与知识管理。当前消息列表只是短期记忆生产 Agent 还需要把重要结论存入向量库或关系库跨会话复用。四是人机协同。很多任务并不适合完全无人值守可以设计“Agent 执行 人工审批”的模式。比如 Agent 生成操作建议人工确认后再执行关键动作。8. 小结回到 Agent 的起点先把闭环做扎实Manus 让人看到通用 Agent 的可能性但工程的起点永远是那条最简单的循环模型规划、工具执行、观察结果、再次规划。把这个闭环跑通再逐步加入权限、持久化、监控、成本控制和多 Agent 协作才是比较稳妥的路径。对于新手不需要一开始就复刻复杂产品。先写一个能调用两个工具的 Agent手动打印每一步日志观察模型如何生成参数、工具结果如何影响下一轮决策比直接搭一个庞大的框架更有价值。对于已经在做 Agent 应用的开发者建议把精力放在工具设计的稳定性、状态可恢复性和成本可视性上这三项决定了 Agent 能否从演示走向真正的生产环境。