1. 项目概述从单点工具到协同智能的跨越最近在折腾AI应用开发的朋友估计没少被“Agent”这个词刷屏。从OpenAI的GPTs到各种开源框架仿佛一夜之间所有应用都得有个“智能体”才够时髦。但说实话很多所谓的Agent项目本质上还是个大号提示词工程把一堆指令塞给一个大模型然后祈祷它能按顺序执行。这种模式在处理简单、线性的任务时还行一旦任务稍微复杂需要多个步骤协作或者信息在不同模块间流转时就很容易“翻车”——要么模型忘了上一步的结果要么在不同工具间切换时逻辑混乱。这正是我这次想和大家深入聊聊Google A2AAgent-to-Agent协议的原因。它不是一个具体的工具库而是一种设计范式一种让多个AI智能体能够像人类团队一样通过清晰、结构化的“对话”来协同工作的“游戏规则”。我这次搭建的“双Agent天气查询系统”就是一个绝佳的入门沙盒。它麻雀虽小五脏俱全一个Agent负责理解用户的自然语言比如“北京明天下午会不会下雨”另一个Agent则专精于调用外部API获取精准的天气数据。它们之间不是简单的函数调用而是通过A2A协议定义的消息格式进行“协商”和“协作”。这个项目的价值远不止于查个天气。它为你理解未来AI应用架构提供了一个清晰的切片。当你掌握了如何让两个Agent可靠地对话你就掌握了构建更复杂智能工作流比如自动数据分析报告生成、多步骤客户服务、跨平台内容创作的核心拼图。而且我把它做成了一个开箱即用的GitHub项目里面包含了完整的代码、详细的配置说明以及我踩过的所有坑的解决方案。无论你是想快速体验A2A的威力还是希望以此为蓝本开发自己的多Agent应用这个项目都能给你提供一个扎实的起点。2. 核心架构与Google A2A协议深度解析2.1 为什么是“双Agent”单一模型的局限性在动手之前我们必须先想清楚为什么非要拆成两个Agent用一个强大的模型比如GPT-4一次性完成理解和查询不行吗理论上可以但实践中问题很多。这涉及到AI应用开发中一个核心的“关注点分离”原则。让一个模型既做自然语言理解NLU又做精确的API调用就像让一个建筑师既画设计图又去砌砖。虽然天才或许能做到但效率低且容易出错。具体来说单一模型的弊端有三提示词污染与指令遗忘复杂的系统提示词System Prompt会占用大量上下文窗口。当你要求模型“先理解用户意图再提取城市和日期最后调用某API”时它在生成长回复的过程中可能会遗忘或混淆早期步骤的指令细节。工具调用不精确天气API的参数格式是固定的比如城市名必须是“Beijing”而不是“北京”日期格式必须是“YYYY-MM-DD”。让一个通才模型去保证每次都能输出完全合规的JSON参数需要极其精细的提示工程且仍有失败概率。难以维护和迭代如果你想升级天气数据源或者增加空气质量查询功能你需要修改一个庞大而复杂的提示词牵一发而动全身测试和调试成本很高。而“双Agent”架构则将问题分解理解AgentNLU Agent只关心一件事——从用户模糊的、口语化的请求中精准提取出结构化信息。它的输出是高度规范的例如{intent: query_weather, parameters: {city: 北京, date: 2023-10-27}}。它的系统提示词可以专注于意图识别和实体抽取的优化。执行AgentAPI Agent只关心另一件事——根据收到的结构化参数调用正确的接口并格式化返回结果。它的系统提示词可以专注于错误处理、API重试和响应模板。两者各司其职通过一个明确的协议这就是A2A进行通信大大降低了单个模型的认知负荷提升了整个系统的可靠性和可维护性。2.2 Google A2A协议智能体间的“通用语言”那么这两个Agent怎么“说话”呢这就是Google A2A协议要解决的问题。你可以把它想象成智能体世界的HTTP协议或者gRPC框架。它定义了一套标准的消息格式和交互流程确保不同来源、不同能力的Agent能够无缝协作。A2A协议的核心在于其消息结构。一个典型的A2A消息不仅仅包含文本内容更是一个承载了丰富元数据的信封。在我的项目实现中我简化并实践了其核心思想消息体主要包含以下几个关键部分{ from: nlp_agent, to: weather_api_agent, type: task_request, content: { action: get_weather, parameters: { location: Beijing, date: 2023-10-27 } }, context_id: session_abc123 }from/to清晰标识消息的发送者和目标接收者。这是多Agent协作的基础避免了消息在复杂路由中丢失。type定义消息的类型。例如task_request任务请求、task_response任务响应、error错误、heartbeat心跳检测。不同类型的消息触发接收方不同的处理逻辑。content消息的实际负载。这里包含了具体的动作指令和参数。结构化是关键它让执行Agent无需再做二次解析。context_id会话或任务上下文ID。这对于关联同一个用户请求相关的所有消息至关重要尤其在异步或长时间运行的任务中。协议带来的核心优势解耦理解Agent和执行Agent可以独立开发、测试和部署。只要它们都遵守A2A消息格式就可以替换其中的任何一个。比如你可以把天气API Agent换成股票查询Agent而理解Agent完全不用改动。可观测性由于所有交互都是结构化的消息你可以非常容易地记录、追踪和调试整个对话流程。消息日志就是最好的调试工具。可扩展性当需要引入第三个Agent比如一个负责将天气结果生成诗意化描述的Agent时你只需要定义它如何消费和产生A2A消息即可无需重构整个系统。注意目前Google A2A协议本身还在不断演进中并没有一个官方的、所有框架都必须实现的“标准库”。我这里的实现是一种基于其核心思想的实践方案。在实际项目中你可以根据需求自定义消息字段例如增加priority优先级、require_response_by超时等使其更贴合你的业务场景。3. 系统搭建全流程实操指南3.1 环境准备与依赖安装这个项目基于Python因为它有最丰富的AI生态。我们主要依赖LangChain框架来快速构建Agent并使用OpenAI的GPT模型作为智能体的“大脑”。当然你也可以替换为其他兼容OpenAI API的模型或本地模型。首先创建一个干净的虚拟环境并安装核心依赖# 创建并激活虚拟环境以conda为例 conda create -n a2a-weather python3.10 conda activate a2a-weather # 安装核心库 pip install langchain langchain-openai python-dotenv requestslangchain 提供了构建Agent所需的高级抽象如AgentExecutor、Tool等。它就像乐高积木让我们能快速组装智能体。langchain-openai LangChain的OpenAI集成包方便我们调用GPT模型。python-dotenv 用于管理环境变量安全地存储你的OpenAI API密钥等敏感信息。requests 用于我们的执行Agent调用外部天气API。接下来获取并配置API密钥OpenAI API Key 前往OpenAI平台注册并获取。这是驱动两个Agent“思考”的燃料。天气API Key 我推荐使用和风天气、OpenWeatherMap等提供免费层级的服务。注册后获取你的密钥。在项目根目录创建一个.env文件内容如下OPENAI_API_KEYsk-your-openai-api-key-here WEATHER_API_KEYyour-weather-api-key-here OPENAI_BASE_URLhttps://api.openai.com/v1 # 如果你使用第三方代理可在此修改重要提示务必把.env文件加入到.gitignore中切勿将包含密钥的文件提交到GitHub等公开仓库3.2 理解AgentNLU Agent的实现细节这个Agent的目标是将用户输入“翻译”成机器可读的结构化指令。我们使用LangChain的create_react_agent模式来构建因为它鼓励模型进行“思考-行动”的循环适合需要推理的任务。首先定义它的“系统提示词”。这个提示词是Agent的“人格”和“职责说明书”至关重要。nlp_agent_prompt 你是一个专业的自然语言理解助手。你的唯一任务是从用户的查询中提取出查询天气所需的明确信息。 用户可能会用各种方式表达例如“北京明天天气怎么样”、“后天上海会下雨吗”、“帮我看看纽约下周一的温度”。 你必须严格按以下JSON格式输出且只输出这个JSON不要有任何其他解释 {{ intent: query_weather, parameters: {{ city: 提取出的城市名必须是英文如Beijing, Shanghai, New York, date: 提取出的日期格式必须为YYYY-MM-DD。如果用户说‘明天’则计算为明天的日期‘后天’、‘大后天’同理。如果用户未指明则默认为今天。 }} }} 请仔细分析用户输入{user_input} 关键设计点解析强约束输出格式提示词中明确要求“只输出JSON”这大大减少了模型“胡说八道”的概率。JSON的键intent,parameters是固定的为下游处理提供了稳定接口。清晰的规则对city和date字段的格式做了明确规定。特别是日期要求模型进行简单的相对日期计算将“明天”等口语词转换为具体日期。这一步的预处理能极大减轻执行Agent的负担。单一职责整个提示词都围绕“信息提取”这一件事不涉及任何调用API的知识。然后我们使用LangChain将其包装成一个Tool。为什么是Tool因为在A2A的思维里每个Agent都可以被视作一个对外提供服务的“工具”。这里NLU Agent提供的工具就是“解析天气查询”。from langchain.agents import Tool from langchain_openai import ChatOpenAI import json import re def parse_weather_query(user_input: str) - str: NLU Agent的核心函数解析用户输入返回结构化JSON。 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # temperature0使输出更确定 prompt nlp_agent_prompt.format(user_inputuser_input) response llm.invoke(prompt).content # 关键步骤清洗和验证模型输出 # 模型有时会在JSON外加引号或markdown代码块需要提取 json_match re.search(r\{.*\}, response, re.DOTALL) if json_match: json_str json_match.group(0) try: # 尝试解析验证JSON格式是否正确 parsed json.loads(json_str) # 进一步验证必要字段是否存在 if all(k in parsed for k in [intent, parameters]): return json_str except json.JSONDecodeError: pass # 如果解析失败返回一个安全的默认结构 return json.dumps({intent: query_weather, parameters: {city: Unknown, date: 2023-10-27}}) # 将函数封装成Tool nlp_tool Tool( nameWeatherQueryParser, funcparse_weather_query, description解析用户关于天气的自然语言查询提取城市和日期信息返回JSON格式的结构化数据。 )实操心得在定义Tool的description时要尽可能精确。这个描述会被用于Agent的“工具选择”推理过程。一个模糊的描述会导致主控Agent不知道在什么情况下该调用它。3.3 执行AgentWeather API Agent的构建这个Agent负责具体的“脏活累活”。它接收结构化的参数调用第三方天气API并将返回的原始数据格式化成对人类友好的文本。首先实现一个调用真实天气API的函数。这里以和风天气为例import requests from datetime import datetime def fetch_weather_from_api(city: str, date: str) - dict: 调用天气API获取数据。 # 1. 参数预处理 # 确保城市名是API支持的格式例如和风天气需要城市ID或英文名 city_code_map {Beijing: CN101010100, Shanghai: CN101020100, New York: USNY0996} location city_code_map.get(city, city) # 2. 构建请求 api_url https://devapi.qweather.com/v7/weather/3d params { location: location, key: os.getenv(WEATHER_API_KEY), # 从环境变量读取密钥 lang: en } try: response requests.get(api_url, paramsparams, timeout10) response.raise_for_status() # 如果状态码不是200抛出异常 data response.json() # 3. 从API响应中提取目标日期的数据 target_date datetime.strptime(date, %Y-%m-%d).date() for day_data in data.get(daily, []): api_date datetime.strptime(day_data[fxDate], %Y-%m-%d).date() if api_date target_date: return { city: city, date: date, weather_desc: day_data.get(textDay, N/A), temp_max: day_data.get(tempMax, N/A), temp_min: day_data.get(tempMin, N/A), precip: day_data.get(precip, N/A) } return {error: fNo data found for {date}} except requests.exceptions.RequestException as e: return {error: fAPI request failed: {str(e)}} except (KeyError, ValueError) as e: return {error: fFailed to parse API response: {str(e)}}接下来构建这个Agent。它也需要一个系统提示词指导它如何利用fetch_weather_from_api这个工具并组织回复。weather_agent_prompt 你是一个天气信息专家。你的任务是使用工具获取指定城市和日期的详细天气数据并以清晰、友好、完整的方式呈现给用户。 你拥有一个名为 fetch_weather 的工具它需要两个参数city (英文城市名) 和 date (格式为YYYY-MM-DD)。 请遵循以下步骤 1. 确认你收到的 city 和 date 参数是有效的。 2. 调用 fetch_weather 工具。 3. 分析工具返回的结果。 4. 如果结果中包含错误信息如实告知用户并建议其检查城市名或日期。 5. 如果成功获取数据请组织成以下格式的回复 “【{city}】{date}的天气情况 - 天气状况{weather_desc} - 最高温度{temp_max}°C - 最低温度{temp_min}°C - 降水概率{precip}%” 你可以根据数据适当添加一句温馨提示如“温差较大请注意增减衣物”或“有降水可能出门请带伞”。 当前请求参数city{city}, date{date} 请开始执行。同样我们将这个逻辑封装成一个Tooldef execute_weather_query(city: str, date: str) - str: Weather API Agent的核心函数。 # 调用底层API函数 weather_data fetch_weather_from_api(city, date) if error in weather_data: return f抱歉获取天气信息时出错{weather_data[error]}。请确认城市名和日期是否正确。 # 格式化输出 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.7) # 这里temperature可以稍高让回复更自然 prompt weather_agent_prompt.format(citycity, datedate) # 这里为了简化我们直接将数据传给LLM进行格式化。更复杂的做法是将数据放入上下文。 formatted_response llm.invoke(prompt).content return formatted_response weather_tool Tool( nameFetchWeather, funcexecute_weather_query, description根据提供的英文城市名和日期YYYY-MM-DD格式获取该地该日的详细天气预报信息。 )3.4 A2A协议控制器与主控Agent的集成现在我们有了两个专业的“工具人”Agent。我们需要一个“项目经理”主控Agent来协调它们。这个主控Agent负责接收用户原始输入决定工作流程并按照A2A协议在两个Agent间传递消息。我们使用LangChain的create_react_agent来创建这个主控Agent。它会根据对用户问题的分析自动决定是否调用以及按什么顺序调用我们提供的工具。from langchain.agents import create_react_agent, AgentExecutor from langchain.memory import ConversationBufferMemory # 1. 为主控Agent准备工具列表 tools [nlp_tool, weather_tool] # 注意顺序LangChain有时会根据描述匹配度选择工具 # 2. 创建主控Agent的系统提示词 master_agent_prompt 你是一个智能协调助手。你的目标是理解用户的问题并通过协调两个专业工具来解决问题。 你拥有以下两个工具 1. WeatherQueryParser: 当用户的问题看起来是在询问天气时使用这个工具。它会将用户模糊的语言转化为明确的城市和日期。 2. FetchWeather: 当你已经获得了明确的城市和日期信息后使用这个工具来获取具体的天气预报。 你的工作流程应该是 用户询问天气 - 调用 WeatherQueryParser 获得结构化数据 - 解析出其中的 city 和 date - 调用 FetchWeather 工具 - 将最终结果返回给用户。 如果用户的问题与天气无关请礼貌地告知你的能力范围。 请开始处理用户问题{input} # 3. 初始化LLM和记忆用于多轮对话 llm ChatOpenAI(modelgpt-4, temperature0) # 主控Agent使用更强的模型推理更准 memory ConversationBufferMemory(memory_keychat_history, return_messagesTrue) # 4. 创建ReAct Agent master_agent create_react_agent( llmllm, toolstools, promptmaster_agent_prompt ) # 5. 创建执行器 agent_executor AgentExecutor.from_agent_and_tools( agentmaster_agent, toolstools, memorymemory, verboseTrue, # 设置为True可以看到Agent的思考过程调试非常有用 handle_parsing_errorsTrue # 优雅处理解析错误 ) # 6. 运行测试 def run_weather_system(user_query: str) - str: 运行整个双Agent天气查询系统。 result agent_executor.invoke({input: user_query}) return result[output] # 示例 if __name__ __main__: print(run_weather_system(请问北京后天天气如何))当verboseTrue时你会在控制台看到类似以下的思考链这就是ReAct框架的威力 Entering new AgentExecutor chain... 思考用户问的是北京后天的天气。这是一个天气查询问题。我应该先使用WeatherQueryParser工具来解析出具体的城市和日期。 行动调用 WeatherQueryParser 行动输入请问北京后天天气如何 观察{intent: query_weather, parameters: {city: Beijing, date: 2023-10-28}} 思考我已经从解析结果中获得了明确的城市Beijing和日期2023-10-28。现在可以使用FetchWeather工具来获取具体天气。 行动调用 FetchWeather 行动输入cityBeijing, date2023-10-28 观察【Beijing】2023-10-28的天气情况...具体天气信息 思考我已经获得了天气信息可以将其作为最终答案返回给用户。 最终答案【Beijing】2023-10-28的天气情况... Finished chain.这个过程完美体现了A2A的思想主控Agent作为协调者在理解Agent产出结构化消息和执行Agent消费结构化消息之间传递信息驱动任务完成。4. 部署、优化与常见问题排查4.1 项目工程化与GitHub仓库管理一个可复现、易协作的项目离不开好的工程实践。我的GitHub仓库结构如下a2a-weather-system/ ├── README.md # 项目说明、快速开始 ├── requirements.txt # 依赖清单 ├── .env.example # 环境变量模板 ├── .gitignore # 忽略.env等文件 ├── src/ │ ├── __init__.py │ ├── agents/ │ │ ├── __init__.py │ │ ├── nlu_agent.py # 理解Agent实现 │ │ └── weather_agent.py # 执行Agent实现 │ ├── protocols/ │ │ ├── __init__.py │ │ └── a2a_message.py # A2A消息结构定义 │ ├── tools/ # 工具函数 │ └── main.py # 主程序入口集成Agent └── tests/ # 单元测试 ├── test_nlu_agent.py └── test_weather_agent.py关键点requirements.txt使用pip freeze requirements.txt生成确保他人能安装完全相同的依赖版本。.env.example文件列出了所有需要的环境变量键名用户只需复制为.env并填入自己的值即可。将不同功能的代码分模块存放提高了可读性和可维护性。protocols文件夹强调了A2A协议的核心地位。编写基本的单元测试特别是对NLU Agent的解析函数和API调用函数进行测试能极大提升代码的健壮性。4.2 性能优化与成本控制技巧在原型阶段后若想投入实际使用必须考虑性能和成本。缓存策略天气数据在短时间内变化不大。可以对fetch_weather_from_api函数的结果进行缓存。使用functools.lru_cache或外部缓存如Redis对(city, date)组合缓存一段时间例如10分钟能显著减少对天气API和OpenAI API的调用。from functools import lru_cache import time lru_cache(maxsize100) def fetch_weather_from_api_cached(city: str, date: str, cache_ttl: int 600): # 简易的带TTL的内存缓存 # 实际项目建议用Redis cache_key f{city}:{date} # ... 检查缓存逻辑 ...模型选型与降级主控Agent需要较强的推理能力可以使用GPT-4。但NLU Agent和Weather Agent的任务相对固定完全可以使用更便宜、更快的gpt-3.5-turbo甚至微调过的中小模型如Qwen、DeepSeek的API成本会大幅下降。异步处理如果系统需要处理高并发请求可以将Agent的执行过程异步化。LangChain支持异步调用结合asyncio可以避免在I/O等待如调用API时阻塞整个应用。输入验证与限流在parse_weather_query函数中我们已经看到了对模型输出的清洗和验证。此外还应在用户输入侧增加基础验证防止恶意或无效输入消耗API额度。对于公开服务设置每分钟/每天的调用频率限制也是必要的。4.3 常见问题与排查实录在开发和测试过程中我遇到了几个典型问题这里分享排查思路问题一NLU Agent偶尔返回非JSON格式。现象parse_weather_query函数中的正则表达式匹配失败返回了默认值。排查打印出模型response的原始内容。发现有时模型会在JSON外加上“json ...”的markdown标记或者多了一些解释性文字。解决强化提示词使用更严厉的指令如“你必须且只能输出一个合法的JSON对象不要有任何其他字符不要用代码块包裹”。同时在解析函数中增加更健壮的清洗逻辑比如先尝试json.loads如果失败再尝试用正则提取。问题二主控Agent错误地选择了工具。现象用户问“北京天气”主控Agent却直接调用了FetchWeather但因为没有日期参数而失败。排查将verbose设为True观察Agent的思考链。发现是因为FetchWeather工具的description描述不够准确让主控Agent误以为它能直接处理自然语言。解决精确化每个工具的description明确其输入和输出。将FetchWeather的描述改为“需要明确的英文城市名和YYYY-MM-DD格式的日期来获取天气预报”这样主控Agent就能更准确地判断必须先调用解析工具。问题三天气API返回错误或超时。现象系统返回“API请求失败”。排查检查WEATHER_API_KEY环境变量是否正确加载。检查网络连接。手动用curl或Postman测试天气API确认接口本身是否正常、参数格式是否正确。查看API返回的具体错误码和消息在fetch_weather_from_api函数的异常处理中打印出来。解决在代码中增加更全面的异常处理对不同的HTTP状态码如401鉴权失败、429超过限额、500服务器错误进行区别处理并给出更友好的用户提示。实现简单的重试机制如tenacity库对于网络波动导致的临时失败进行自动重试。考虑配置备用API源当主用API失败时自动切换。问题四处理复杂或歧义的用户输入。现象用户输入“我老家和上海下周的天气”NLU Agent无法处理多个实体。解决这是当前设计的局限。对于更复杂的场景需要升级NLU Agent的能力。可以在提示词中明确指导模型处理多实体情况并输出列表形式的JSON。修改主控Agent逻辑使其能循环处理多个(city, date)对并发起多次查询。这正体现了A2A架构的优势你只需要增强或替换NLU Agent这个模块而无需重构整个系统。你可以尝试用更专业的NLP模型或服务来替代当前的GPT解析。这个双Agent天气查询系统就像搭积木一样让我清晰地看到了智能体协作的脉络。A2A协议提供的标准化通信方式是解开复杂AI应用开发的一把钥匙。从这个小系统出发你可以尝试引入更多Agent比如一个负责将天气数据生成图表的Agent或者一个负责根据天气推荐穿衣的Agent让这个智能团队不断壮大。