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

Manus 这类通用 AI Agent 产品走红之后，很多开发者的第一反应是“这不就是调大模型吗”，但真正动手复现一个最小版本时，才会发现事情没有那么简单。一个能自主规划、调用工具、读取结果、继续执行的 Agent，核心不是某一次 Prompt 写得好，而是一套完整的工程回路：任务拆解、工具注册、上下文维护、结果验证、失败重试和终止条件。这篇博客会从概念开始，讲清楚 Agent 工作流的基本结构，然后搭建一个 Python 最小项目，实现一个类 Manus 的轻量 Agent 循环，最后给出运行验证、常见问题排查和生产化建议。

1. Manus 走红的本质：Agent 不是“多轮聊天”，而是“计划-执行-验证”回路

1.1 传统聊天与 Agent 工作流的区别

传统聊天机器人通常只做一件事：把用户问题丢给大模型，拿到文字回复后直接展示。这个模式适合问答、写作、翻译，但它不具备“做事”的能力。

Agent 工作流则不同。Agent 会把目标当成一个需要完成的任务，先拆解成多个步骤，然后逐步执行。每一步可能需要调用外部工具，比如搜索、计算、操作浏览器、读文件、写数据库；执行完之后，模型还要根据工具返回的结果决定下一步做什么，直到任务完成或达到安全边界。

Manus 之所以让人印象深刻，是因为它把“看到结果后继续行动”这个过程做成了产品体验。从工程角度看，这种体验背后就是一个循环：模型产出动作，系统执行动作，观察结果，再把结果交回模型，模型继续产出下一个动作。

1.2 为什么 Agent 需要工具、记忆和循环控制

一个完整的 Agent，至少要包含三样东西。

第一是工具。没有工具的大模型只能输出文字，无法改变外部世界。工具可以是函数、API、数据库接口或浏览器操作。工具把模型和业务系统连接起来，Agent 才具备实际完成任务的能力。

第二是记忆。Agent 在多次工具调用之间，必须知道“我之前已经查到了什么”“哪个步骤已经完成”。在目前的大模型架构下，最直接的记忆载体就是对话历史消息列表。每一步模型生成的内容、工具返回的结果都追加到消息里，模型在下一轮就能看到上下文。

第三是循环控制。Agent 不能无限制地调用工具。如果模型陷入重复调用，或者工具一直失败，系统必须能够在指定的最大迭代次数内停止，并返回当前进度和失败信息。缺少循环控制，Agent 在生产环境里会变成“费用黑洞”和“死循环机器”。

1.3 一个最小 Agent 回路的五个阶段

任何 Agent 都可以抽象成下面这条链路：

任务输入 -> 模型规划 -> 产生工具调用 -> 执行工具 -> 返回观察结果 -> 模型再规划 -> ... -> 终止

具体到代码实现，这条链路可以拆成五个阶段：

1. 组装系统提示词和任务输入。
2. 调用大模型，获得结构化动作。
3. 解析动作。如果是工具调用，执行对应函数。
4. 把工具返回值追加到消息列表。
5. 判断是否终止，如果未终止则回到第 2 步。

只要把这条链路实现一遍，你就掌握了 Agent 最核心的骨架。Manus 比这个复杂的地方在于它加入了更多工具、更长的规划、文件系统和浏览器能力，但底层循环是一致的。

2. 搭建一个类 Manus 的轻量 Agent 环境

2.1 技术选型和运行环境

这个项目不追求复刻 Manus 的完整能力，只实现最小 Agent 闭环。技术选型上，我选择 Python 和openaiSDK，因为当前大多数模型服务都提供 OpenAI 兼容接口，便于替换模型供应商。

运行环境建议如下：

如果你的项目会用到浏览器自动化，可以把playwright作为后续扩展。本文第一阶段只做基础工具调用，不引入浏览器，降低环境复杂度。

2.2 准备 Python 项目和依赖

在本地创建项目目录，并初始化虚拟环境。

mkdir mini-agent cd mini-agent python -m venv .venv source .venv/bin/activate

然后安装依赖。

pip install openai python-dotenv

openai用于调用大模型，python-dotenv用于从.env文件加载 API Key。建议把.env文件加入.gitignore，避免密钥被提交到代码仓库。

如果你的模型服务不是 OpenAI 官方服务，而是兼容接口，可以设置base_url。示例代码会预留这个参数。

2.3 配置文件和环境变量

在项目根目录创建.env文件：

LLM_API_KEY=你的密钥 LLM_BASE_URL=https://api.openai.com/v1 LLM_MODEL=gpt-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

各文件职责如下：

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_key=os.getenv("LLM_API_KEY"), base_url=os.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": f"unknown tool: {function_name}"}) func = TOOL_FUNCTIONS[function_name] return func(**arguments)

这里的关键点是把arguments由 JSON 字符串解析成 Python 字典，再按参数名展开。如果模型生成的参数缺少必填项，工具函数会抛出TypeError，主循环需要把异常转换成工具返回信息，避免整个程序崩溃。

3.4 实现主循环

主循环是整个 Agent 的心脏。它要做四件事：

1. 调用模型，携带完整历史消息和工具描述。
2. 判断模型是否要求调用工具。
3. 如果调用工具，执行工具并追加结果消息。
4. 如果不调用工具，说明模型认为任务已完成，输出最终回答。

# 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( model=os.getenv("LLM_MODEL"), messages=messages, tools=TOOLS, ) 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.py

4. 关键机制拆解：为什么这样设计

4.1 工具描述如何驱动模型生成参数

在传统函数调用里，参数由调用方写死；在 Agent 里，参数是由大模型根据用户任务和工具描述动态生成的。这意味着工具描述本身必须像一份“接口文档”一样精确。

以calculator工具为例，模型收到“计算 (1 + 2) * 4”这个任务时，会从参数 schema 中知道需要提供expression字符串，并且这个字符串应该是合法数学表达式。如果description写得太短，比如只写“计算”，模型可能生成expression: "(1+2)*4"，也可能生成expression: "计算(1+2)*4"，后者就会导致工具执行失败。

实际项目里，给工具写描述时，建议包含：

• 这个工具做什么。
• 什么场景下应该调用。
• 每个参数如何填写。
• 有没有格式限制。

4.2 多轮上下文如何形成短期记忆

messages列表就是 Agent 的短期记忆。每一轮对话都包含四类消息：

模型每生成一次回复，都会基于完整的消息列表。这相当于让它“记得”自己已经执行过哪些工具，以及工具返回结果是什么。只要消息列表没有被截断，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 学习环境与生产环境的差异

当前代码适合在本地学习，但直接搬到生产环境会有明显问题。

5.4 参数调优速查表

6. 常见问题排查链路

6.1 模型总是返回空内容，不调用工具

现象：日志里只有[Agent Final]，但内容为空。

可能原因：

• 系统提示词让模型认为可以直接回答。
• 模型本身不支持工具调用。
• tools参数没有传对，或格式不符合模型要求。

检查顺序：

1. 确认tools参数传入的是列表。
2. 确认模型型号支持tool_calls。
3. 在系统提示词中说明“如果需要工具，请先调用工具”。
4. 输出message.tool_calls日志，确认模型是否返回了调用。

6.2 模型不调用工具，直接把答案编出来

现象：任务需要查询知识库，但模型只输出文字，没有调用query_knowledge。

可能原因：

• 工具描述没有说明“必须调用才能获取答案”。
• 模型认为自己的内建知识足够回答。
• 工具名称和任务描述不匹配。

处理方式：

• 在工具描述里加“该问题必须调用本工具才能回答”。
• 在系统提示词中强调：不要编造工具结果，工具结果必须来自真实调用。

6.3 同一个工具被反复调用

现象：calculator被连续调用多次，参数几乎一样。

可能原因：

• 工具返回结果包含模型不理解的格式。
• 模型没有在下一轮看到工具结果。
• tool_call_id或消息顺序错误。

检查方式：

1. 打印每轮messages[-2:]，确认 tool 结果是否已经被追加。
2. 确认 tool 消息的content是字符串而非对象。
3. 在 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 能否从演示走向真正的生产环境。