十年匠心定制 · 商业建站与技术教学双线并行 咨询热线:400-886-1026 service@lmnt.cn
ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

AI Agent智能体入门:从工作流、反射机制到MCP与多智能体实战

AI Agent智能体入门:从工作流、反射机制到MCP与多智能体实战 最近 Agent 相关的讨论热度非常高从“AI Agent 开发”到“MCP 多智能体”再到各种“工作流”和“反射”机制几乎每天都能看到新文章、新工具冒出来。很多朋友在社区里问同一个问题Agent 到底该怎么学传统的大模型应用和 Agent 应用之间边界在哪里为了把这块内容啃下来我专门去看了吴恩达老师的 Agent 智能体官方教程结合自己的工程实践做了一套笔记和示例代码。这篇文章就把这些内容整理出来从智能体工作流、反射机制、工具调用、MCP 到多智能体协作按一条完整的路线讲清楚。如果你是刚接触 Agent 的新手可以把它当成一份带代码的入门教程如果你已经在做 LLM 应用开发可以跳过部分基础直接看 MCP 和多智能体协作的实战片段里面有不少值得直接复用的设计思路。1. 背景与核心概念1.1 什么是 Agent 智能体Agent 智能体通俗地说是一个“能自己做事的 AI 程序”。它不只是回答你的问题还能根据你给的目标自己决定下一步要调用什么工具、查询什么数据然后循环执行直到完成任务。专业一点的描述是Agent 是一个以大语言模型LLM为核心控制器结合规划能力、工具调用能力和记忆能力在动态环境中自主完成任务的系统。它和传统大模型应用的核心区别可以从下面这个简单对比里看出来对比维度传统 Prompt 应用Agent 智能体应用工作模式一次性输入一次性输出多次推理、多次调用工具循环推进是否调用外部工具通常不调用经常调用代码解释器、搜索、数据库、API状态管理无状态或简单上下文有记忆、有中间状态、有反思错误处理输出错误就结束可以自我纠正、重试、换策略典型场景文本摘要、翻译、分类自动写代码并执行、数据查询、多步骤任务吴恩达老师在教程里反复强调一个核心观点大模型本身只是一个“推理引擎”真正让模型“动起来”去完成任务的是围绕模型搭建的智能体框架。这个思路对我后来的开发方向影响很大下面会逐步展开。1.2 Agentic AI 与四个核心设计模式Agentic AI 指的就是以智能体为核心的人工智能应用形态。吴恩达老师的教程把这类应用拆成了几种常见设计模式分别是反思Reflection工具使用Tool Use规划Planning多智能体协作Multi-Agent Collaboration这四种模式并不是互斥的实际项目里往往会把它们组合起来用。比如一个复杂的数据分析 Agent既要用工具调用去执行代码也要通过反思机制来修正自己的中间结果可能还会拆分成一个“写代码 Agent”和一个“检查代码 Agent”。我建议你在学习时不要一上来就学框架而是先用最简单的 Python 代码理解这四种模式各自解决什么问题。等理解透彻了再用 LangGraph、AutoGen 之类框架去落地会顺利得多。2. 为什么 Agent 值得系统学习2.1 业务价值从“能用”到“好用”很多开发者都有这种体验直接用 Prompt 调用大模型写出来的应用很“飘”。你问它一个开放问题它说得头头是道但一旦需要它做实事——比如帮你查一下数据库里的用户订单、把一段 Markdown 转成 Word、抓取一个网页并提炼重点——它就容易出错。Agent 的价值正好体现在这里它把“大模型生成文本”升级成了“大模型驱动完成任务”。对于后端开发者来说这意味着很多重复的“数据搬运 逻辑判断 格式化输出”工作可以逐步交给带工具调用的 Agent 去完成。从技术发展角度来说MCP 协议的推出也让 Agent 接入外部数据源的方式变得更统一。以前每个工具都要单独写一套接入代码现在通过 MCP Server 暴露接口多个 Agent 和多个应用都能共用一套工具协议这也是最近“MCP 多智能体”相关的搜索量持续走高的原因。2.2 学习路线先原理后框架再实战我见过很多朋友一上来就学 AutoGen、LangGraph结果被各种抽象概念绕晕。这里分享一下我推荐的顺序先学 Prompt Engineering理解模型输入输出的基本规律。再学 Function Calling理解模型怎么决定调用工具。然后用原生 Python 自己写一个最小 Agent 循环不用任何框架。接着理解四种设计模式Reflection、Tool Use、Planning、Multi-Agent。再接触 MCP 协议学习如何用统一方式接入外部工具。最后学习 LangGraph 或 AutoGen 等框架用于生产级落地。这篇文章后面就是按照这条路径来组织的。每一部分都会给出可运行的代码片段你可以在本地环境直接跟着敲一遍。3. 环境准备与版本说明为了便于演示本文示例使用以下环境。版本需要根据你的项目实际情况调整重点是理解思路。操作系统Windows 10/11、macOS、Ubuntu 均可Python3.10 或更高版本大模型 APIOpenAI 兼容接口或者国内大模型厂商的兼容接口pip 包openai、langchain、langgraph、fastmcp 等建议先创建虚拟环境避免污染全局 Python 环境python -m venv agent_env source agent_env/bin/activate # Windows 下使用 agent_env\Scripts\activate安装基础依赖pip install openai langchain langgraph fastmcp python-dotenv如果你只是想跑通最基础的 Agent 循环只需要 openai 和 python-dotenv 就够了。后面涉及 MCP 和 LangGraph 的部分再补装其他依赖。在项目根目录创建.env文件用来存放 API KeyOPENAI_API_KEY你的_API_Key注意不要把.env文件提交到 Git 仓库。生产环境建议使用密钥管理服务。4. 从零实现一个最小 Agent 工作流4.1 什么是 Agent 工作流Agent 工作流Agent Workflow指的是把一次任务拆成多个步骤让大模型在每一步中做决策并根据决策结果执行相应动作直到任务完成。一个最小的工作流通常包含下面几个节点接收用户指令将指令注入 Prompt调用大模型得到结果判断结果是否包含工具调用如果有工具调用执行工具并返回结果给模型如果没有工具调用结束任务并返回最终答案这个循环看起来简单但它是所有复杂 Agent 系统的地基。你不理解这个循环后面学 LangGraph 的节点和边、多智能体的通信机制都会觉得云里雾里。4.2 实现最小循环下面用原生 Python 实现一个最简单的 Agent 工作流。这里使用 OpenAI 兼容接口你可以把base_url换成任意兼容接口的地址。# 文件路径agent_basic.py import json import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() client OpenAI( api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL, https://api.openai.com/v1), ) # 模拟一个简单的工具函数 def get_weather(city: str) - str: 根据城市名返回模拟天气 weather_data { 北京: 晴25°C, 上海: 多云28°C, 广州: 小雨30°C, } return weather_data.get(city, 暂无该城市天气数据) # 工具定义用于 Function Calling tools [ { type: function, function: { name: get_weather, description: 查询指定城市的当前天气, parameters: { type: object, properties: { city: { type: string, description: 城市名称如北京、上海 } }, required: [city] } } } ] def run_agent(user_input: str, max_steps: int 5): messages [ {role: system, content: 你是一个有用的助手可以调用工具获取天气信息。}, {role: user, content: user_input}, ] for step in range(max_steps): print(f\n--- 第 {step 1} 次调用模型 ---) response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolstools, tool_choiceauto, ) assistant_message response.choices[0].message # 如果模型没有请求调用工具直接返回最终答案 if not assistant_message.tool_calls: print(f最终回答{assistant_message.content}) return assistant_message.content # 如果有工具调用把 assistant 消息加入上下文 messages.append(assistant_message) # 执行每一个工具调用 for tool_call in assistant_message.tool_calls: if tool_call.function.name get_weather: args json.loads(tool_call.function.arguments) city args.get(city) result get_weather(city) print(f调用工具 get_weather参数{city}结果{result}) messages.append({ role: tool, tool_call_id: tool_call.id, content: result, }) print(已达到最大步数任务结束。) return None if __name__ __main__: run_agent(北京今天天气怎么样)运行这段代码后你会看到类似这样的输出--- 第 1 次调用模型 --- 调用工具 get_weather参数北京结果晴25°C --- 第 2 次调用模型 --- 最终回答北京今天天气晴朗气温25°C。这段代码虽然简单但已经包含了 Agent 工作流最核心的骨架模型决定调用工具、程序执行工具、把结果回传给模型继续推理。5. 反射机制让 Agent 学会自我纠错5.1 反射是什么在 Agent 领域反射Reflection指的是让大模型对自己的输出进行审视、评估和修正的机制。为什么需要反射因为大模型生成的第一次答案往往不是最优答案。它可能会忽略关键条件、产生幻觉或者使用了错误的格式。通过反射我们让模型“再想一遍”、“再检查一遍”从而提升最终输出的质量。吴恩达老师在教程里用“生成 → 评估 → 反思 → 再生成”的循环来讲解这个机制。你在本地也可以非常简单地复现这个过程不需要任何框架。5.2 代码生成场景下的反射下面的例子模拟了一个能写代码并自我检查的 Agent。这里我们不实际执行代码而是让模型扮演“代码审查者”来检查候选代码。# 文件路径reflection_demo.py import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() client OpenAI( api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL, https://api.openai.com/v1), ) SYSTEM_PROMPT 你是一个资深 Python 工程师擅长编写高质量、可维护的代码。 def generate_code(task: str) - str: response client.chat.completions.create( modelgpt-4o-mini, messages[ {role: system, content: SYSTEM_PROMPT}, {role: user, content: f请编写 Python 代码完成任务{task}\n\n只输出代码不要输出额外解释。} ], ) return response.choices[0].message.content def review_code(task: str, code: str) - str: response client.chat.completions.create( modelgpt-4o-mini, messages[ {role: system, content: 你是一个严格的代码审查员负责找出代码中的 Bug、边界问题和不规范之处。请用中文回复审查意见。}, {role: user, content: f任务需求{task}\n\n候选代码\npython\n{code}\n\n\n请指出问题并给出修改建议。} ], ) return response.choices[0].message.content def improve_code(task: str, code: str, review: str) - str: response client.chat.completions.create( modelgpt-4o-mini, messages[ {role: system, content: SYSTEM_PROMPT}, {role: user, content: f任务需求{task}\n\n之前的代码\npython\n{code}\n\n\n审查意见\n{review}\n\n请根据审查意见改写代码输出完整代码。} ], ) return response.choices[0].message.content def run_with_reflection(task: str, rounds: int 2): code generate_code(task) print( 初始代码 ) print(code) for i in range(rounds): print(f\n 第 {i 1} 轮反思 ) review review_code(task, code) print(审查意见, review) code improve_code(task, code, review) print(改进后代码) print(code) return code if __name__ __main__: task 写一个函数从列表中找出所有重复两次以上的元素并返回去重后的列表。 run_with_reflection(task, rounds1)在这个例子中模型先生成一个候选方案然后由另一个角色“审查员”检查最后让生成者根据审查意见修改代码。这个过程显著提升了代码的鲁棒性尤其在处理边界条件时效果明显。5.3 反射机制的实现要点反思不是简单的“再生成一遍”而是要先进行批判性评估。评估角色和生成角色的 Prompt 要尽量分离避免思维惯性。反思轮数不是越多越好一般 1 到 3 轮就能获得明显收益超过 3 轮可能陷入过度修改。6. 工具使用与 MCP 接入6.1 模型为什么要使用工具大模型的知识截止到训练数据那一刻之后的新数据、企业内部数据、实时数据它都无法直接获取。工具调用的价值就是把模型的“推理能力”和外部世界的“实时能力”连接起来。常见的工具类型包括搜索 API数据库查询接口代码解释器文件读写能力HTTP 请求企业内部服务6.2 通过 Function Calling 调用自定义工具上一节的天气查询例子就是 Function Calling 的标准用法。这里我再补充一个更实用的案例让 Agent 读取数据库中的用户表统计不同状态下的用户数量。我们先用 SQLite 模拟一个用户表-- 文件路径init_db.sql CREATE TABLE users ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, status TEXT NOT NULL DEFAULT active ); INSERT INTO users (name, status) VALUES (张三, active); INSERT INTO users (name, status) VALUES (李四, active); INSERT INTO users (name, status) VALUES (王五, disabled); INSERT INTO users (name, status) VALUES (赵六, disabled);然后在 Python 中定义查询工具# 文件路径db_tool.py import sqlite3 import json from openai import OpenAI from dotenv import load_dotenv load_dotenv() client OpenAI( api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL, https://api.openai.com/v1), ) def query_user_stats() - str: 查询用户表中不同状态的用户数量 conn sqlite3.connect(app.db) cursor conn.cursor() cursor.execute(SELECT status, COUNT(*) FROM users GROUP BY status) rows cursor.fetchall() conn.close() return json.dumps(rows, ensure_asciiFalse) tools [ { type: function, function: { name: query_user_stats, description: 查询用户表中不同状态的用户数量返回 JSON 字符串 } } ] def run(): messages [ {role: system, content: 你是一个数据分析助手可以查询用户表。}, {role: user, content: 帮我统计一下不同状态的用户数量}, ] response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolstools, ) msg response.choices[0].message if msg.tool_calls: messages.append(msg) for tool_call in msg.tool_calls: if tool_call.function.name query_user_stats: result query_user_stats() messages.append({ role: tool, tool_call_id: tool_call.id, content: result, }) final_response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, ) print(final_response.choices[0].message.content) if __name__ __main__: run()这个例子里有个很重要的点工具返回的是原始数据大模型负责把原始数据转化为自然语言回答。数据格式越结构化模型的解读效果越好。6.3 MCP 协议统一工具接入标准MCPModel Context Protocol解决的核心问题是“工具接入标准化”。在没有 MCP 之前每接入一个新工具都需要为特定的大模型应用单独写一套工具定义和调用逻辑。而使用 MCP 协议后工具提供方只需要实现一个统一的 MCP Server任何支持 MCP 的客户端都可以直接复用这个工具。这很像数据库领域的 JDBC/ODBC你写的业务代码不直接依赖具体数据库而是通过统一接口访问。MCP 在 Agent 生态里扮演的就是这个“统一接口”的角色。不同的 MCP Server 可以暴露不同能力比如数据库 MCP Server提供表结构查看、SQL 查询能力文件系统 MCP Server提供文件读写能力浏览器 MCP Server提供网页抓取能力蓝湖/设计稿 MCP Server提供设计稿信息获取能力6.4 用 FastMCP 搭建一个最小 MCP Server下面使用fastmcp库搭建一个极简 MCP Server演示如何暴露一个天气查询工具。# 文件路径weather_mcp_server.py from fastmcp import FastMCP mcp FastMCP(weather-server) mcp.tool() def get_weather(city: str) - str: 查询指定城市的天气情况 weather_data { 北京: 晴25°C, 上海: 多云28°C, 广州: 小雨30°C, } return weather_data.get(city, 暂无该城市天气数据) if __name__ __main__: mcp.run(transportstdio)运行python weather_mcp_server.py服务会启动一个标准输入输出传输方式的 MCP Server。搭建 MCP Server 的核心收益在于一次编写多个支持 MCP 的 AI 应用都能复用同一个工具定义。7. 规划与多智能体协作7.1 自主规划Planning Agent规划能力是指 Agent 把一个复杂任务拆解成多个子任务并为每个子任务选择合适的工具或策略。最简单的实现方式是在 Prompt 里要求模型先输出一个执行计划然后再逐步执行。在不需要引入复杂框架的前提下我们可以先让模型生成一个“任务拆解清单”然后再针对每个子任务分别调用工具或模型。这种方式虽然简单但对于很多业务场景已经足够用了。在多智能体系统中规划往往由一个专门的“Planner Agent”负责。Planner Agent 只负责拆解任务和下发指令不负责具体执行。执行工作交给其他专门的 Agent。7.2 多智能体协作模式多智能体协作并不是简单地把多个 Agent 堆在一起而是要设计它们之间的通信机制和任务分配机制。目前常见的设计模式有主管/工人模式Supervisor/Worker 一个主管 Agent 负责理解用户请求把任务拆解后分发给多个工人 Agent最后汇总结果。流水线模式Pipeline 任务按顺序经过多个 Agent每个 Agent 只负责一个环节。例如需求分析 Agent → 代码生成 Agent → 测试 Agent。辩论模式Debate 多个 Agent 针对同一个问题给出答案然后互相评审最终投票或由裁判 Agent 综合判断。下面用伪代码演示主管/工人模式的基本结构# 文件路径multi_agent_demo.py from openai import OpenAI from dotenv import load_dotenv load_dotenv() client OpenAI( api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL, https://api.openai.com/v1), ) def call_model(system_prompt: str, user_prompt: str) - str: response client.chat.completions.create( modelgpt-4o-mini, messages[ {role: system, content: system_prompt}, {role: user, content: user_prompt}, ], ) return response.choices[0].message.content def supervisor_agent(user_request: str): plan_prompt ( 你是任务主管请将以下任务拆解为最多3个子任务 每个子任务用一行描述只输出任务列表不要输出其他内容。\n f任务{user_request} ) plan call_model(你是一个严谨的项目经理。, plan_prompt) print( 任务拆解 ) print(plan) results [] for line in plan.strip().splitlines(): if not line.strip(): continue # 去掉序号 subtask line.split(. , 1)[-1].strip() print(f\n 工人 Agent 执行{subtask} ) worker_result call_model( 你是业务专家请认真完成任务只输出结果。, subtask ) results.append(worker_result) print(worker_result) # 主管汇总 summary_prompt ( 请根据以下多个子任务的执行结果给出一个完整、连贯的最终回复。\n\n \n\n.join(results) ) summary call_model(你是任务主管负责汇总。, summary_prompt) print(\n 最终汇总 ) print(summary) return summary if __name__ __main__: supervisor_agent(帮我调研 Python 和 Java 在 Web 开发中的优缺点并给出选型建议。)这就是多智能体协作的最基础形态。生产环境中你会用 LangGraph、AutoGen 或 CrewAI 来管理更复杂的智能体状态、记忆和通信机制但底层的“分工 汇总”逻辑是一样的。8. 常用框架与工程化落地8.1 LangGraph 在 Agent 工作流中的角色LangGraph 是一个用于构建有状态 Agent 应用的框架它的核心思想是把工作流建模成一张图。图中的节点是“动作”边是“状态转移”。和普通的 LangChain Chain 不同LangGraph 支持循环、分支和持久化这让它非常适合实现 Agent 的“推理-行动-观察”循环。下面是一个使用 LangGraph 实现的最简 Agent 图# 文件路径langgraph_basic.py from typing import TypedDict, Literal from langgraph.graph import StateGraph, END class AgentState(TypedDict): messages: list def call_model(state: AgentState) - AgentState: # 这里可以替换为真实的模型调用 last_message state[messages][-1] state[messages].append(f模型处理了{last_message}) return state def should_continue(state: AgentState) - Literal[call_model, end]: # 如果消息数量超过3条就结束 if len(state[messages]) 3: return end return call_model graph StateGraph(AgentState) graph.add_node(call_model, call_model) graph.set_entry_point(call_model) graph.add_conditional_edges(call_model, should_continue, {call_model: call_model, end: END}) app graph.compile()这段代码展示了 LangGraph 的核心概念节点、边、条件跳转。你可以在节点里加入工具调用、数据库查询等真实逻辑。8.2 多智能体框架选型建议框架特点适合场景LangGraph图结构灵活度高状态管理强需要精细控制工作流的项目AutoGen多智能体对话机制成熟研究型任务、智能体辩论场景CrewAI上手快Role/Goal/Backstory 设计简单中小型业务自动化自己实现无额外依赖逻辑透明学习、原型验证、轻量业务选择框架时不要盲目追求新。如果你的业务只需要一个固定流程自己写一个循环可能比引入框架更省心。如果你需要处理复杂的条件分支、多人协作、状态持久化那么 LangGraph 更合适。9. 常见问题与排查思路9.1 模型反复调用同一个工具问题现象常见原因解决思路Agent 卡在工具调用循环里反复调用同一个工具工具返回的内容模型无法理解上下文过长导致模型迷失工具定义不够清晰检查工具返回数据格式在 system prompt 中要求“如果工具返回的结果与问题无关请直接回答不知道”设置最大步数限制建议在实际项目中始终设置max_steps避免无限循环导致费用失控。9.2 MCP Server 启动失败问题现象常见原因解决思路MCP Server 启动后立即退出缺少依赖包、Python 环境不对、stdio 模式未配置检查依赖是否安装完整确认 Python 解释器是否为虚拟环境在客户端配置中指定正确的启动命令9.3 Function Calling 返回格式错误问题现象常见原因解决思路模型返回的参数无法解析为 JSON模型幻觉、temperature 设置过高、工具 schema 描述不清晰将 temperature 调到 0 或 0.2 左右精简参数描述对解析失败做兜底重试9.4 多智能体之间上下文不一致问题现象常见原因解决思路工人 Agent 返回的内容和主管 Agent 预期不符子任务拆解不清晰、上下文传递不完整在子任务 Prompt 中带上背景摘要要求工人 Agent 严格按格式输出增加主管 Agent 的校验环节10. 最佳实践与工程建议10.1 Prompt 设计给每个 Agent 明确“角色 目标 限制”。先让模型输出计划再输出执行结果。凡是希望模型遵守的格式规则都提供最少一个示例few-shot。10.2 工具与权限管理工具权限要遵守最小权限原则。Agent 需要只读查询时就不要再给它删除权限。工具返回的数据中不要包含敏感字段。如果有一定要脱敏后再返回模型。所有涉及删除、更新、生产环境的操作要加人工确认环节不能直接让 Agent 自动执行。10.3 日志与可观测性记录每一次模型调用的输入输出便于排查。记录每次工具调用的参数和结果方便审计。为每个 Agent 会话生成唯一的 trace_id把同一会话的所有日志串起来。10.4 成本控制把历史消息做摘要压缩防止上下文无限膨胀。对简单任务使用小模型复杂任务使用大模型。设置单次任务的模型调用次数上限。10.5 测试与评估针对 Agent 的高频任务准备测试集每次修改 Prompt 后跑一遍回归测试。检查最终输出中是否包含非法或捏造的工具调用参数。定期用真实业务日志评估 Agent 的完成率和错误率。11. 总结与学习路线建议最后做一个简要收束。这篇文章从 Agent 智能体的基本概念出发梳理了 Agentic AI 的四种核心设计模式并用原生 Python 实现了最小 Agent 工作流、反射机制、工具调用和 MCP Server最后介绍了多智能体协作模式和工程化落地的注意点。如果你正在规划 Agent 学习路线可以按下面的优先级来推进先用原生 Python 跑通一个 Agent 循环理解模型、工具、上下文三者的闭环关系。再重点掌握 Function Calling 工具定义的方法这是所有 Agent 应用的基础。然后把反射机制用到自己的代码生成或数据整理任务中感受自我纠错带来的效果提升。之后可以试试搭建一个 MCP Server把公司内部的一个小工具封装成统一接口。最后当业务复杂度上来之后再开始用 LangGraph 或多智能体框架做工程化落地。需要特别提醒的是Agent 并不是越复杂越好。很多业务场景用一个简单的循环加两个工具调用就足够了。等你真正理解了底层机制再根据需求引入框架才能做到有的放矢。建议你直接从“最小 Agent 循环”动手写起。自己亲手写过一遍工具调用的数据流你对后面所有框架的理解都会快很多。如果这篇文章对你有帮助可以收藏备用后续我再补充更多关于 LangGraph 状态管理和多智能体面试题的实战内容。
返回列表