1. 为什么你需要这份协同开发指南如果你最近在捣鼓AI应用特别是想让它不只是个“聊天机器人”而是能真正帮你干活、甚至自己协调几个“小弟”去完成复杂任务的智能系统那你肯定绕不开这几个词Agent、Function Calling、MCP、A2A。网上资料不少但要么是干巴巴的概念解释看完还是不知道怎么动手要么是零散的代码片段拼凑不起来。我自己在构建智能系统时就经历过这种“一看就会一写就废”的阶段。所以我决定写这篇指南目标很明确让你从“知道是什么”到“真正能用起来”。我不会只给你讲理论而是会用一个贯穿始终的“旅行规划”案例手把手带你把这四项技术像搭积木一样组合起来。你会看到一个能听懂你模糊需求、自动查天气、订酒店、生成行程单的智能助理是如何从一行行代码中“长”出来的。这不仅仅是技术栈的堆砌更是一种开发范式的转变——从编写死板的流程到设计能自主协作的智能体。准备好了吗我们开始吧。2. 核心概念拆解它们到底是什么以及如何协同工作在深入代码之前我们必须先统一“语言”。这四项技术不是孤立的它们像一支足球队各有位置协同作战。我画了一张关系图来帮你理解用户需求 | v [智能体 Agent] -- 大脑负责规划和决策 | (使用) v [Function Calling] -- 手和脚执行单一动作 | (通过) v [MCP 协议] -- 万能适配器连接各种工具 | (协调) v [A2A 通信] -- 团队沟通语言连接多个大脑 | v 最终结果Agent智能体这是整个系统的“大脑”和“指挥官”。你可以把它想象成一个有经验的项目经理。它不直接写代码或调用API但它有目标比如“规划一次旅行”、有记忆记得用户喜欢海景房、会规划先查天气再订酒店最后排行程、能决策根据预算选择酒店。它的核心公式是Agent 大模型思考 记忆经验 工具能力 规划策略。单独一个大模型只是智库加上后面这些才成了能行动的智能体。Function Calling函数调用这是大模型伸出“手”去触碰现实世界的标准方式。以前大模型只能“说”不能“做”。Function Calling 让它能说“嘿去调用那个‘查询天气’的函数把‘北京’作为参数传进去。” 这就像是给大脑连接上了机械臂。它是 Agent 能够执行动作的底层技术基础。没有它Agent 就是个光说不练的“思想家”。MCP模型上下文协议这是由 Anthropic 推动的一个开放协议。你可以把它理解为智能硬件界的USB-C 接口。在 MCP 出现之前每个大模型厂商比如 OpenAI、Anthropic的 Function Calling 方式略有不同你要为每个模型写适配代码很麻烦。MCP 定义了一套统一的、模型无关的通信标准。无论后端是 GPT、Claude 还是开源模型只要工具服务遵循 MCP 协议前端就能用同样的方式调用。它解决了工具生态的“碎片化”问题让 Function Calling 变得通用和可移植。A2A智能体间通信这是 Google 提出的让多个 Agent 能一起工作的“团队协作规范”。当一个任务太复杂一个 Agent 搞不定时比如规划旅行涉及酒店、机票、景点多个环节就需要多个专业 Agent 分工合作。A2A 定义了它们之间如何打招呼、如何传递任务、如何回复结果、出错时怎么沟通。它解决了多智能体系统的“沟通混乱”问题让 Agent 们能像训练有素的团队一样有序协作。它们如何协同想象一下旅行规划场景主规划 Agent大脑收到任务“规划东京之旅”。它通过A2A协议向酒店预订 Agent发送一条结构化消息“兄弟订一下东京的酒店。”酒店预订 Agent 内部为了查询房态通过MCP协议调用一个统一的“数据库查询工具”。这个工具的具体实现可能就是通过Function Calling去执行一段查询 SQL 的代码。酒店 Agent 将结果通过A2A协议回复给主规划 Agent。主规划 Agent 汇总所有信息最终给用户一个完整方案。看它们环环相扣。接下来我们就从最基础的 Function Calling 开始一步步把它们全部实现。3. 基石深入理解并实战 Function CallingFunction Calling 是整个技术栈的“手脚”是动作执行单元。很多教程只教你怎么用 OpenAI 的 API 调一下但没讲明白背后的机制和坑。我这里带你从原理到安全实践彻底搞懂。3.1 不仅仅是“调用函数”它的技术本质它的核心流程是一个清晰的闭环用户提问“上海现在多少度”LLM 判断模型分析这个问题发现需要实时数据自己不知道于是决定调用工具。生成调用指令模型不会直接写代码而是输出一个严格格式的 JSON 对象包含要调用的函数名和参数比如{name: get_weather, arguments: {city: Shanghai}}。开发者执行你的代码收到这个 JSON去本地找到真正的get_weather函数传入参数执行。结果返回LLM你把执行结果比如“多云28°C”以特定格式再传回给模型。LLM 组织回答模型拿到真实数据组织成自然语言回复用户“上海目前多云气温28摄氏度。”关键点模型从未直接执行你的代码。它只是“建议”调用哪个函数以及参数是什么。真正的执行权牢牢掌握在你的服务器上。这是一个重要的安全设计。3.2 手把手实现绕过框架使用原生 API很多人一上来就用 LangChain 等框架虽然快但容易变成“黑盒”。我建议你先用原生 API 实现一遍理解底层发生了什么。我们继续用天气查询当例子。import openai import json import requests import os from dotenv import load_dotenv # 1. 加载你的 API 密钥 load_dotenv() client openai.OpenAI(api_keyos.getenv(OPENAI_API_KEY)) # 2. 定义真实的工具函数这是你的安全边界 def get_weather(city: str) - str: 查询城市天气。内部可以调用任何安全可控的API或数据库。 # 这里用一个简单的公共API示例生产中请替换为稳定服务 url fhttps://wttr.in/{city}?format%C%t try: resp requests.get(url, timeout5) resp.raise_for_status() return resp.text.strip() except Exception as e: return fError: {str(e)} # 3. 定义工具描述这是给模型看的“说明书” tools [ { type: function, function: { name: get_weather, # 必须和上面函数名一致 description: 获取指定城市的当前天气状况和温度。这是获取实时天气信息的唯一方式。, # 描述至关重要 parameters: { type: object, properties: { city: { type: string, description: 城市名称支持中文或英文例如北京、Shanghai } }, required: [city] # 声明必填参数 } } } ] # 4. 模拟用户请求 messages [{role: user, content: 上海和北京的天气分别怎么样}] # 5. 首次调用让模型决定是否、以及如何调用工具 response client.chat.completions.create( modelgpt-4o, # 或 gpt-3.5-turbo messagesmessages, toolstools, # 把工具“说明书”传给模型 tool_choiceauto, # 让模型自己决定。也可强制调用某个工具。 ) # 6. 检查模型的回复 response_message response.choices[0].message print(模型初始回复:, response_message) # 7. 如果模型决定调用工具 if response_message.tool_calls: # 注意模型可能决定同时调用多个工具并行查询 available_functions {get_weather: get_weather} # 安全函数映射 messages.append(response_message) # 把模型的回复也加入对话历史 # 处理每一个工具调用 for tool_call in response_message.tool_calls: function_name tool_call.function.name function_to_call available_functions.get(function_name) if function_to_call: # 解析模型生成的参数 function_args json.loads(tool_call.function.arguments) # 重要在这里可以添加参数校验、权限检查等安全逻辑 print(f准备调用函数: {function_name}, 参数: {function_args}) # 执行实际函数 function_response function_to_call(**function_args) # 将结果以特定格式追加到消息列表 messages.append({ role: tool, tool_call_id: tool_call.id, # 必须对应模型才知道是哪个调用的结果 content: str(function_response), # 内容需要是字符串 }) else: # 处理模型试图调用未授权函数的情况 print(f警告尝试调用未定义的函数 {function_name}) messages.append({ role: tool, tool_call_id: tool_call.id, content: Error: Function not available., }) # 8. 将工具执行结果送回模型让它生成最终回答 second_response client.chat.completions.create( modelgpt-4o, messagesmessages, ) final_answer second_response.choices[0].message.content print(\n最终回答:, final_answer) else: # 模型认为不需要调用工具直接给出了答案 print(模型直接回答:, response_message.content)运行这段代码你会看到模型可能先调用get_weather查询上海天气拿到结果后再在同一个对话上下文中自动去查询北京天气最后汇总成一个回答。这体现了模型的多步推理能力。注意生产环境中available_functions这个映射表是你的安全防火墙。永远不要让模型能调用任何它想调用的函数必须严格限制在白名单内。同时对function_args进行严格的输入验证防止注入攻击。4. 升级用 MCP 打造可移植的工具层当你熟练使用 Function Calling 后很快就会遇到新问题如果你的应用既要支持 OpenAI又想接 Claude甚至后端想换成一个本地部署的 Llama 模型怎么办难道要为每个模型重写一遍工具调用逻辑这时MCP 的价值就凸显出来了。4.1 MCP 的核心优势一次定义到处运行MCP 协议的核心思想是解耦。它将“工具提供者”Server和“工具调用者”Client通常是大模型分离开并规定了它们之间通信的标准格式。这样做的好处是对模型透明无论是 GPT 还是 Claude它们作为 Client都通过同样的 MCP 接口与工具对话。对工具开发者友好你只需用任何语言Python、JavaScript、Go等按照 MCP 标准实现一次工具服务就可以被所有兼容 MCP 的模型使用。传输方式灵活支持标准输入输出stdio适合本地、HTTP SSE服务器发送事件、WebSocket 等多种通信方式适应不同场景。4.2 实战构建一个数学计算 MCP 服务让我们构建一个简单的 MCP 服务它提供加法和乘法功能。之后无论是通过 Claude 的桌面应用还是通过我们自己的程序都能以统一的方式调用它。首先安装官方库pip install mcp第一步创建 MCP 服务端 (math_server.py)服务端负责声明它提供哪些工具并实现工具的具体逻辑。# math_server.py from mcp.server import Server from mcp.server.models import Tool import asyncio # 创建 Server 实例 app Server(simple_math_server) # 使用装饰器注册工具。这是 MCP 的标准方式。 app.call_tool() async def add(numbers: list[float]) - dict: 计算一组数字的总和。 参数: numbers: 一个包含数字的列表例如 [1, 2, 3.5] 返回: 一个包含总和sum的字典。 total sum(numbers) return {sum: total} app.call_tool() async def multiply(a: float, b: float) - dict: 计算两个数字的乘积。 参数: a: 第一个乘数 b: 第二个乘数 返回: 一个包含乘积product的字典。 product a * b return {product: product} # 我们还可以注册一个“列出所有工具”的工具这在动态发现时很有用。 app.list_tools() async def handle_list_tools(): # 这个方法会自动返回我们通过 app.call_tool() 注册的所有工具信息。 pass # 以 stdio 模式运行服务。这是最简单的本地调试方式。 if __name__ __main__: print(MCP 数学服务启动... (使用 stdio 传输), filesys.stderr) asyncio.run(app.run_stdio())第二步创建 MCP 客户端 (mcp_client.py)客户端连接到服务端并调用其工具。这个客户端可以集成到你的 AI 应用中。# mcp_client.py from mcp import Client import asyncio import sys async def main(): # 1. 初始化客户端指定如何启动服务端和传输方式。 # 这里我们使用 stdio客户端会启动 math_server.py 进程并与之通信。 client Client( command[sys.executable, math_server.py], # 用当前 Python 解释器运行服务脚本 transportstdio ) try: # 2. 建立连接 await client.connect() print(已连接到 MCP 服务。) # 3. 可选列出服务端提供的所有工具 tools await client.list_tools() print(f可用工具: {[t.name for t in tools]}) # 4. 调用加法工具 print(\n调用加法工具: add([5, 10, 15])) add_result await client.call_tool(add, {numbers: [5, 10, 15]}) print(f结果: {add_result}) # 输出: {sum: 30.0} # 5. 调用乘法工具 print(\n调用乘法工具: multiply(7, 8)) mul_result await client.call_tool(multiply, {a: 7, b: 8}) print(f结果: {mul_result}) # 输出: {product: 56.0} except Exception as e: print(f发生错误: {e}) finally: # 6. 断开连接 await client.disconnect() print(连接已关闭。) if __name__ __main__: asyncio.run(main())运行python mcp_client.py你会看到客户端成功调用了服务端的工具并得到了结果。这里的魔力在于客户端完全不需要知道服务端是用 Python 写的还是用其他语言写的。它们只通过 MCP 定义的标准 JSON 格式进行通信。第三步在 AI 应用中使用 MCP 工具现在我们可以修改之前的 Function Calling 代码让它不直接调用本地函数而是去调用 MCP 服务。这只需要改动工具定义和函数执行部分。# 在原有的 Function Calling 代码中修改工具定义 tools_for_llm [ { type: function, function: { name: call_mcp_math, description: 调用远程数学计算服务进行加法或乘法运算。, parameters: { type: object, properties: { operation: {type: string, enum: [add, multiply], description: 运算类型}, args: {type: object, description: 运算参数} }, required: [operation, args] } } } ] # 然后在函数映射表里指向一个调用 MCP 客户端的函数 async def call_mcp_math(operation: str, args: dict): 这个函数作为桥梁连接 LLM 和 MCP 服务。 client Client(command[sys.executable, math_server.py], transportstdio) await client.connect() try: result await client.call_tool(operation, args) return json.dumps(result) finally: await client.disconnect() available_functions {call_mcp_math: call_mcp_math}这样一来你的 AI 应用就与具体的工具实现解耦了。未来要把数学服务换成更强大的计算引擎或者迁移到另一台服务器上使用 SSE 传输只需要更新 MCP 服务端AI 应用侧的代码几乎不用动。5. 构建大脑从 Function Calling 到智能体 Agent有了灵活的工具调用能力Function Calling和统一的工具接入层MCP我们现在可以来打造系统的“大脑”——Agent。Agent 的核心是自主性和规划能力。它不再是你问一句、它调用一个函数的“提线木偶”而是能根据目标自己决定先做什么、后做什么、失败了怎么办的“智能管家”。5.1 使用 LangChain 快速构建 ReAct Agent我们使用流行的 LangChain 框架来简化 Agent 的构建。它会帮我们处理思维链Chain-of-Thought、工具选择、记忆等复杂逻辑。我们构建一个能进行多轮对话、并自主使用天气和计算工具的智能助手。# agent_with_memory.py import os from dotenv import load_dotenv from langchain.agents import AgentExecutor, create_react_agent from langchain.tools import Tool from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate from langchain.memory import ConversationBufferMemory import requests import json load_dotenv() # 1. 定义我们的工具 def get_weather(city: str) - str: 获取天气的工具函数 url fhttps://wttr.in/{city}?format%C%t try: return requests.get(url, timeout5).text.strip() except: return 无法获取天气信息 def calculator(expression: str) - str: 一个简单的计算器工具注意生产环境请使用更安全的评估方法 try: # 警告直接使用 eval 有安全风险此处仅用于演示。 # 真实场景应使用 ast.literal_eval 或专用库如 numexpr。 allowed_chars set(0123456789-*/(). ) if all(c in allowed_chars for c in expression): result eval(expression) return str(result) else: return 错误表达式包含非法字符 except Exception as e: return f计算错误: {e} # 2. 将函数封装为 LangChain Tool tools [ Tool( nameGetWeather, funcget_weather, description当用户询问某个城市的天气时使用此工具。输入应为城市名称。 ), Tool( nameCalculator, funccalculator, description当用户需要进行数学计算时使用此工具。输入为一个数学表达式字符串例如 3 5 * 2。 ) ] # 3. 创建带有记忆的提示词模板 prompt_template 你是一个乐于助人的助手可以回答问题和使用工具。 你拥有之前对话的记忆。 工具 {tools} 使用工具的格式如下 思考我需要先做什么 行动使用的工具名 行动输入工具的输入 观察工具返回的结果 ...这个思考/行动/观察循环可以重复多次 当你有最终答案时必须严格以“最终答案”开头。 之前的对话 {chat_history} 新问题{input} {agent_scratchpad} # LangChain 会自动在这里填充工具的思考过程 prompt PromptTemplate.from_template(prompt_template) # 4. 初始化 LLM 和记忆 llm ChatOpenAI(modelgpt-4o, temperature0) memory ConversationBufferMemory(memory_keychat_history, return_messagesTrue) # 5. 创建 ReAct 模式的 Agent agent create_react_agent(llmllm, toolstools, promptprompt) # 6. 创建执行器并传入记忆 agent_executor AgentExecutor( agentagent, toolstools, memorymemory, verboseTrue, # 打印详细的思考过程便于调试 handle_parsing_errorsTrue, # 处理模型输出格式错误 max_iterations5 # 防止无限循环最多尝试5步 ) # 7. 运行多轮对话 queries [ 北京今天的天气怎么样, 把刚才的温度加上15是多少, # 这里会用到记忆 那么上海呢 ] for query in queries: print(f\n用户: {query}) result agent_executor.invoke({input: query}) print(f助手: {result[output]})运行这段代码你会看到类似以下的输出用户: 北京今天的天气怎么样 Entering new AgentExecutor chain... 思考用户询问北京天气我需要使用 GetWeather 工具。 行动GetWeather 行动输入Beijing 观察Sunny 22°C 思考我得到了北京的天气信息可以给出最终答案。 最终答案北京今天天气晴朗气温22°C。 助手: 北京今天天气晴朗气温22°C。 用户: 把刚才的温度加上15是多少 Entering new AgentExecutor chain... 思考用户指的是刚才北京的温度22°C。这是一个数学计算我需要使用 Calculator 工具。首先提取数字22。 行动Calculator 行动输入22 15 观察37 思考计算完成可以给出答案。 最终答案22°C 加上 15 等于 37°C。 助手: 22°C 加上 15 等于 37°C。看到了吗在第二轮对话中Agent自动从记忆chat_history里提取了“22”这个数字并正确使用了计算器工具。这就是一个初级智能体的样子有记忆能规划先提取数字再计算会使用工具。6. 迈向协作用 A2A 协议连接多个智能体单个 Agent 能力再强也有瓶颈。复杂的任务需要分工协作这就需要 A2AAgent-to-Agent通信协议。它定义了 Agent 之间如何交换信息、理解彼此意图、协同完成任务。我们来实现一个简单的双 Agent 旅行规划场景。6.1 设计消息协议A2A 没有唯一的官方实现但其核心是定义一套双方都能理解的消息格式。我们设计一个简单的 JSON 格式{ id: unique_msg_id_123, from: travel_planner_agent, to: hotel_booking_agent, type: request, intent: find_hotels, payload: { city: Tokyo, check_in: 2024-12-01, check_out: 2024-12-05, guests: 2 }, timestamp: 2024-11-05T10:30:00Z }6.2 实现酒店预订 Agent服务端这个 Agent 扮演一个专业的酒店预订服务它通过 HTTP API 接收来自其他 Agent 的请求。# hotel_agent_server.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from datetime import datetime import uvicorn import uuid app FastAPI(title酒店预订 Agent API) # 定义 A2A 请求消息格式 class A2AMessage(BaseModel): id: str from_agent: str to_agent: str type: str # request, response, error intent: str payload: dict timestamp: str # 模拟一个简单的“酒店数据库” fake_hotel_db [ {id: 1, name: 东京晴空塔酒店, city: Tokyo, price_per_night: 15000, available: True}, {id: 2, name: 东京湾希尔顿, city: Tokyo, price_per_night: 22000, available: True}, {id: 3, name: 京都花园酒店, city: Kyoto, price_per_night: 12000, available: True}, ] app.post(/a2a/request) async def handle_a2a_request(message: A2AMessage): 处理来自其他 Agent 的 A2A 请求 # 1. 验证消息基本格式 if message.type ! request: raise HTTPException(status_code400, detail仅处理 request 类型消息) # 2. 根据意图intent路由到不同的处理函数 if message.intent find_hotels: return await handle_find_hotels(message) elif message.intent book_hotel: return await handle_book_hotel(message) else: raise HTTPException(status_code400, detailf不支持的意图: {message.intent}) async def handle_find_hotels(msg: A2AMessage) - dict: 处理查询酒店请求 city msg.payload.get(city) if not city: return create_a2a_response(msg, error, {error: 缺少城市参数}) # 模拟查询逻辑 found_hotels [h for h in fake_hotel_db if h[city] city and h[available]] hotel_list [{name: h[name], price: h[price_per_night]} for h in found_hotels] response_payload { city: city, hotels: hotel_list, count: len(hotel_list) } return create_a2a_response(msg, success, response_payload) async def handle_book_hotel(msg: A2AMessage) - dict: 处理预订酒店请求 hotel_name msg.payload.get(hotel_name) # 这里省略真实的预订逻辑如调用第三方API、更新数据库等 booking_id str(uuid.uuid4())[:8].upper() response_payload { booking_id: fHTL-{booking_id}, hotel_name: hotel_name, status: confirmed, message: f酒店 {hotel_name} 预订成功。 } return create_a2a_response(msg, success, response_payload) def create_a2a_response(original_msg: A2AMessage, status: str, payload: dict) - dict: 构建一个标准的 A2A 响应消息 return { id: str(uuid.uuid4()), from_agent: original_msg.to_agent, # 接收方变成发送方 to_agent: original_msg.from_agent, # 发送方变成接收方 type: response, intent: original_msg.intent, # 保持意图一致 payload: payload, timestamp: datetime.utcnow().isoformat() Z, in_response_to: original_msg.id, # 指明是对哪条消息的回复 status: status } if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8001)6.3 实现旅行规划 Agent客户端这个 Agent 是用户的主要交互对象它负责分解任务并通过 A2A 协议调用酒店预订 Agent。# travel_planner_agent.py import httpx import asyncio import uuid from datetime import datetime from langchain.agents import AgentExecutor, create_react_agent from langchain.tools import Tool from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate # 1. 定义与酒店 Agent 通信的工具函数 async def call_hotel_agent(intent: str, payload: dict) - str: 通过 A2A 协议调用远程酒店预订 Agent a2a_message { id: str(uuid.uuid4()), from_agent: travel_planner_main, to_agent: hotel_booking_service, type: request, intent: intent, payload: payload, timestamp: datetime.utcnow().isoformat() Z } async with httpx.AsyncClient(timeout30.0) as client: try: # 发送请求到酒店 Agent 的 API resp await client.post(http://localhost:8001/a2a/request, jsona2a_message) resp.raise_for_status() result resp.json() if result.get(status) success: # 将响应结果格式化为字符串供 LLM 理解 return f酒店Agent回复: {result[payload]} else: return f酒店Agent处理失败: {result.get(payload, {})} except Exception as e: return f调用酒店Agent时发生网络错误: {str(e)} # 2. 将 A2A 调用封装为 LangChain Tool tools [ Tool( nameHotelBookingAgent, funclambda intent, payload: asyncio.run(call_hotel_agent(intent, payload)), # 注意处理异步 description用于与酒店预订专家沟通。当你需要查询或预订酒店时使用此工具。 输入应该是一个逗号分隔的字符串第一部分是意图intent第二部分是 JSON 格式的参数。 例如find_hotels, {\city\: \Tokyo\} 或 book_hotel, {\hotel_name\: \东京晴空塔酒店\}。 ), # 可以继续添加其他工具如天气查询、航班查询等 ] # 3. 创建主规划 Agent使用 LangChain llm ChatOpenAI(modelgpt-4o, temperature0) prompt PromptTemplate.from_template( 你是一个旅行规划大师。你可以使用工具来获取信息或执行操作。 你的目标是帮助用户规划完美的行程。 工具 {tools} 请严格按照以下格式使用工具 思考分析用户需求决定下一步 行动使用的工具名 行动输入工具的输入 观察工具返回的结果 ...重复直到任务完成 最终答案必须以“最终答案”开头。 用户问题{input} {agent_scratchpad} ) agent create_react_agent(llmllm, toolstools, promptprompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, max_iterations6) # 4. 运行一个复杂查询 async def main(): complex_query 我想去东京玩先帮我找找那里有什么酒店价格在2万日元以下的。 print(f用户: {complex_query}) result await agent_executor.ainvoke({input: complex_query}) # 使用异步调用 print(f\n规划师Agent最终回答: {result[output]}) if __name__ __main__: # 首先启动酒店 Agent 服务 (hotel_agent_server.py) # 然后运行此规划师 Agent asyncio.run(main())这个例子展示了 A2A 的精髓标准化通信和职责分离。旅行规划 Agent 不需要知道酒店预订的具体逻辑它只需要按照约定好的消息格式intent 和 payload发送请求。酒店预订 Agent 可以独立开发、部署和升级。这种架构使得构建复杂的多智能体系统成为可能每个 Agent 都可以是微服务通过 A2A 协议总线连接。7. 综合实战构建旅行规划全家桶系统现在让我们把前面学的所有东西串起来构建一个微缩但完整的“旅行规划全家桶系统”。这个系统将包含主控 Agent基于 LangChain负责与用户对话、理解复杂需求、协调其他 Agent。天气查询工具通过 Function Calling 实现。酒店预订 Agent一个独立的 A2A 服务。行程生成服务一个通过 MCP 协议暴露的文档生成工具。系统架构图用户 | | (自然语言) v [主控 Agent (LangChain)] | | | | | | (A2A 请求) | | v | | [酒店预订 Agent (FastAPI)] | | | | (Function Calling) | v | [天气查询工具] | | (MCP 调用) v [行程生成 MCP 服务] | v PDF 行程单关键集成点主控 Agent 的工具箱主控 Agent 的tools列表里会包含三种工具一个直接调用本地天气函数的Tool。一个封装了 A2A 客户端、用于与酒店 Agent 通信的Tool。一个封装了 MCP 客户端、用于调用行程生成服务的Tool。工作流程当用户说“为我规划一个下周去东京的三天行程预算中等”主控 Agent 会思考需要天气、酒店、景点信息最后生成文档。行动1调用天气工具获取东京下周天气。行动2通过 A2A 工具请求酒店 Agent 查找东京的酒店。行动3假设还有景点查询工具获取景点信息。行动4汇总所有信息通过 MCP 工具请求行程生成服务创建 PDF。最终答案将 PDF 文件链接或内容概要回复给用户。由于篇幅所限这里不展开全部代码但核心的集成思路如下# 在主控 Agent 的 tools 定义中集成三者 tools [ # 工具1: 原生 Function Calling 工具 Tool( nameGetWeather, funcget_weather, # 本地函数 description查询城市天气。 ), # 工具2: 封装 A2A 调用的工具 Tool( nameConsultHotelAgent, funccall_hotel_agent_a2a, # 内部使用 httpx 发送 A2A 消息 description与酒店预订专家沟通用于查询或预订酒店。输入为 JSON 字符串包含 city、date 等信息。 ), # 工具3: 封装 MCP 调用的工具 Tool( nameGenerateItinerary, funccall_mcp_itinerary_service, # 内部使用 mcp.Client 连接 MCP 服务 description生成精美的旅行行程 PDF 文档。输入为包含行程详情的 JSON 字符串。 ) ] # 然后将这个 tools 列表交给 LangChain 的 create_react_agent # Agent 就会在规划过程中自主判断并调用这些工具。通过这个案例你可以清晰地看到四项技术如何各司其职、协同工作Function Calling是原子操作MCP让工具调用标准化Agent提供规划和决策的大脑而A2A则将多个大脑连接成一个协作网络。构建这样的系统你不再是写死流程的程序员而是设计智能生态的架构师。8. 避坑指南与性能优化实战在实际开发中你会遇到各种预料之外的问题。我把自己踩过的坑和解决方案分享给你希望能帮你节省大量调试时间。坑1Function Calling 描述不清导致模型乱调用或不敢调用问题工具的描述description写得太模糊比如“处理数据”模型可能无法判断何时该调用。或者描述太长挤占了宝贵的上下文窗口。解决方案描述要具体、场景化不要写“获取信息”要写“当用户询问某个城市的当前天气状况和温度时使用此工具”。说明参数格式在描述中明确参数的格式例如“城市名称请使用拼音或英文例如Beijing, Shanghai”。使用向量检索如果工具非常多可以将工具描述存入向量数据库如Chroma。当用户提问时先用查询检索最相关的几个工具再动态注入给模型而不是一次性给所有工具描述。坑2Agent 陷入思考循环或执行无用步骤问题Agent 可能在一个简单问题上反复思考或者调用一些不必要的工具消耗 Token 和时间。解决方案设置硬性限制在使用AgentExecutor时务必设置max_iterations最大迭代次数和max_execution_time最大执行时间。优化提示词Prompt在提示词中明确指令例如“请用最少的步骤解决问题”、“如果第一次工具调用失败请直接告知用户并停止尝试”。实现成本监控在每次调用模型后累计 Token 消耗设置预算上限超标则强制终止。坑3MCP 服务在 stdio 模式下性能瓶颈问题stdio传输是同步的如果工具执行很慢比如查询大型数据库会阻塞整个请求。解决方案切换到 SSE 或 WebSocket对于生产环境使用sse或websocket传输模式支持异步和流式响应。服务端异步化确保你的工具函数是异步的使用async def并且内部执行的是异步 IO 操作如使用asyncpg查数据库aiohttp发请求。实现超时和重试在客户端设置合理的超时时间并对临时性失败实现重试机制。坑4A2A 通信中的消息丢失或乱序问题在网络不稳定的情况下A2A 消息可能丢失或者后发的消息先到导致状态混乱。解决方案消息唯一 ID 与关联 ID每条消息必须有全局唯一 ID (id)响应消息必须包含其所回复消息的 ID (in_response_to)。实现确认机制ACK重要的指令性消息如book_hotel要求接收方必须返回确认消息。发送方可以设置一个超时未收到 ACK 则重发。引入消息队列对于高并发或关键任务不要直接 Agent 对 Agent 调用。引入一个消息中间件如 Redis Streams、RabbitMQ、Kafka让 Agent 将消息发布到队列由中间件保证可靠投递和顺序。这虽然增加了复杂度但大幅提升了系统的鲁棒性。坑5工具调用的安全性风险问题模型可能生成恶意参数调用危险函数或者工具函数本身存在漏洞。解决方案严格的白名单制度维护一个{function_name: function_object}的映射字典。模型返回的函数名必须在这个字典里否则直接拒绝执行。参数验证与清洗在调用真实函数前对模型生成的参数进行严格的类型、范围、格式验证。对于数据库查询避免直接拼接 SQL使用参数化查询。沙箱环境对于执行代码、访问文件系统等高危操作务必在沙箱环境如 Docker 容器、e2b 等安全沙箱中运行限制其权限和资源。构建智能体系统是一场充满挑战但也极具成就感的旅程。从让一个大模型调用一个函数开始到设计出多个能自主协作的智能体每一步都伴随着新的问题和解决方案。我的经验是从小处着手快速迭代。先实现一个能稳定完成单一任务的 Agent再逐步添加记忆、规划能力最后通过 MCP 和 A2A 将其扩展成分布式系统。多利用 LangChain、CrewAI、AutoGen 等成熟框架的生态它们帮你解决了很多底层难题。最重要的是始终保持对系统行为的观察和测试因为智能体的行为有时会出乎你的意料而这正是其魅力所在。