1. 项目概述
最近在开发一个跨平台的智能出行服务系统时,遇到了一个棘手的问题:如何让不同平台的AI助手(Agent)协同工作,完成从机票预订到酒店入住再到打车的全流程服务?经过多次尝试,我发现LangChain1.0提供的Agent-to-Agent(A2A)协作模式完美解决了这个问题。
这个方案的核心思想是"专业分工+统一调度":每个业务领域(机票、酒店、打车)都有自己专属的Agent,它们只专注于自己的专业领域;同时有一个总协调Agent负责统筹全局,确保整个流程顺畅执行。这种架构不仅保证了每个环节的专业性,还能实现全流程的自动化闭环。
需要模型API调用? 免费领10W Token,多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 技术选型与架构设计
2.1 为什么选择LangChain1.0?
LangChain1.0相比之前的版本有几个关键改进特别适合我们的需求:
- 更稳定的Agent调用机制:1.0版本重构了Runnable接口,Agent之间的调用更加可靠
- 简化的工具封装:新的@tool装饰器让工具定义更加简洁直观
- 更好的错误处理:内置的异常捕获和重试机制提高了系统稳定性
2.2 分层架构设计
我们的系统采用三层架构:
- 工具层:封装各平台的核心业务能力(携程机票、美团酒店、滴滴打车)
- 专属Agent层:每个Agent只负责单一业务领域
- 协调层:总协调Agent负责流程控制和结果整合
这种分层设计带来的好处是:
- 职责单一,每个组件只做一件事
- 易于扩展,新增业务只需添加新的专属Agent
- 维护方便,修改一个业务不会影响其他业务
3. 环境准备与基础配置
3.1 安装依赖
首先需要安装必要的Python包,这里特别指定版本以避免兼容性问题:
bash复制pip install langchain==1.0.0 langchain-openai==0.1.0 python-dotenv==1.0.1
3.2 大模型配置
我们选择通义千问(qwen-plus)作为底层大模型,主要考虑是:
- 国内可直接使用
- 兼容OpenAI接口,迁移成本低
- 性价比高
配置代码如下:
python复制import os
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="qwen-plus",
api_key=os.getenv("aliQwen-api"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
注意:需要提前设置环境变量aliQwen-api,或者直接在代码中替换为你的API Key
4. 工具层实现
4.1 携程机票预订工具
使用@tool装饰器封装机票预订功能:
python复制from langchain.tools import tool
@tool("CtripBookFlight", description="预订机票的唯一工具,必须调用,参数是departure出发地、arrival目的地、date出行日期(格式2026-02-01)")
def ctrip_book_flight(departure: str, arrival: str, date: str) -> str:
"""携程订机票:模拟接口返回"""
return f"【携程机票预订成功】\n出发地:{departure}\n目的地:{arrival}\n出行日期:{date}\n航班号:CA1885(北京首都T3→上海浦东T2)\n起飞时间:14:00\n降落时间:16:30\n座位:经济舱34A\n电子客票号:999-1234567890\n舱位等级:经济舱超级经济座"
关键点:
- 工具描述必须清晰明确,大模型靠这个理解如何调用工具
- 参数类型和格式要定义清楚
- 目前是模拟返回,实际应用可替换为真实API调用
4.2 美团酒店预订工具
python复制@tool("MeituanBookHotel", description="预订酒店的唯一工具,必须调用,参数是city城市、near_by附近地标、check_in入住日期、check_out离店日期")
def meituan_book_hotel(city: str, near_by: str, check_in: str, check_out: str) -> str:
"""美团订酒店:模拟接口返回"""
return f"【美团酒店预订成功】\n城市:{city}\n位置:{near_by}附近\n入住日期:{check_in}\n离店日期:{check_out}\n酒店名称:上海浦东机场铂尔曼大酒店\n房型:豪华大床房(含双人自助早餐)\n房号:1508\n预订号:MT20260201001\n入住人:张三\n退房政策:入住后24小时内可免费取消"
4.3 滴滴打车预约工具
python复制@tool("DidiBookTaxi", description="预约打车的唯一工具,必须调用,参数是start起点、end终点、time用车时间")
def didi_book_taxi(start: str, end: str, time: str) -> str:
"""滴滴打车:模拟接口返回"""
return f"【滴滴打车预约成功】\n起点:{start}\n终点:{end}\n用车时间:{time}\n车型:滴滴快车(舒适型)\n司机姓名:王师傅\n车牌号:沪A12345\n司机电话:13800138000\n预估费用:35元(券后立减5元,实付30元)\n预计接驾时间:16:35\n车型空间:5座,可放2件24寸行李箱"
5. 专属Agent实现
5.1 携程机票Agent
python复制from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
def create_ctrip_agent(llm):
"""创建携程机票专属Agent"""
llm_with_tools = llm.bind_tools([ctrip_book_flight])
prompt = ChatPromptTemplate.from_messages([
("system", "你是专业的机票预订助手,只能调用CtripBookFlight工具,调用格式必须正确,直接传入参数:departure='北京', arrival='上海', date='2026-02-01',调用后直接返回工具执行的完整字符串结果,不能有任何其他内容!"),
("human", "{input}")
])
return prompt | llm_with_tools | output_parser
设计要点:
- 严格限制Agent只能调用指定工具
- Prompt中明确指定参数格式
- 返回结果不做任何额外处理
5.2 美团酒店Agent
python复制def create_meituan_agent(llm):
"""创建美团酒店专属Agent"""
llm_with_tools = llm.bind_tools([meituan_book_hotel])
prompt = ChatPromptTemplate.from_messages([
("system", "你是专业的酒店预订助手,只能调用MeituanBookHotel工具,调用格式必须正确,直接传入参数:city='上海', near_by='浦东机场', check_in='2026-02-01', check_out='2026-02-02',调用后直接返回工具执行的完整字符串结果,不能有任何其他内容!"),
("human", "{input}")
])
return prompt | llm_with_tools | output_parser
5.3 滴滴打车Agent
python复制def create_didi_agent(llm):
"""创建滴滴打车专属Agent"""
llm_with_tools = llm.bind_tools([didi_book_taxi])
prompt = ChatPromptTemplate.from_messages([
("system", "你是专业的打车预约助手,只能调用DidiBookTaxi工具,调用格式必须正确,直接传入参数:start='上海浦东机场T2', end='上海浦东机场铂尔曼大酒店', time='2026-02-01 16:40',调用后直接返回工具执行的完整字符串结果,不能有任何其他内容!"),
("human", "{input}")
])
return prompt | llm_with_tools | output_parser
6. 总协调Agent实现
总协调Agent是整个系统的核心,负责调度各个专属Agent并整合结果:
python复制from langchain_core.runnables import RunnableLambda
def create_travel_coordinator_agent(llm, ctrip_chain, meituan_chain, didi_chain):
"""创建出行总协调Agent,负责调度各专属Agent"""
def a2a_schedule(input_dict):
print("🔍 开始执行A2A协作流程,依次调用各业务Agent...\n")
# 提取工具原始函数(用于兜底)
ctrip_func = ctrip_book_flight.func
meituan_func = meituan_book_hotel.func
didi_func = didi_book_taxi.func
# 1. 调用携程机票Agent
print("1. 调用【携程机票Agent】>>>")
try:
ctrip_result = ctrip_chain.invoke({"input": "订机票"})
except Exception as e:
print(f"携程Agent调用异常:{e},启用兜底逻辑")
ctrip_result = ""
if not ctrip_result.strip():
ctrip_result = ctrip_func("北京", "上海", "2026-02-01")
print(f"✅ 携程执行结果:\n{ctrip_result}\n" + "-"*80 + "\n")
# 2. 调用美团酒店Agent
print("2. 调用【美团酒店Agent】>>>")
try:
meituan_result = meituan_chain.invoke({"input": "订酒店"})
except Exception as e:
print(f"美团Agent调用异常:{e},启用兜底逻辑")
meituan_result = ""
if not meituan_result.strip():
meituan_result = meituan_func("上海", "浦东机场", "2026-02-01", "2026-02-02")
print(f"✅ 美团执行结果:\n{meituan_result}\n" + "-"*80 + "\n")
# 3. 调用滴滴打车Agent
print("3. 调用【滴滴打车Agent】>>>")
try:
didi_result = didi_chain.invoke({"input": "预约打车"})
except Exception as e:
print(f"滴滴Agent调用异常:{e},启用兜底逻辑")
didi_result = ""
if not didi_result.strip():
didi_result = didi_func("上海浦东机场T2", "上海浦东机场铂尔曼大酒店", "2026-02-01 16:40")
print(f"✅ 滴滴执行结果:\n{didi_result}\n" + "-"*80 + "\n")
# 整合最终报告
total_report = f"""
📋 【携程-美团-滴滴 A2A协作最终报告】
{('='*90)}
📌 协作场景:2026-02-01 北京飞上海完整出行服务
📌 执行状态:所有Agent调用成功(含异常兜底保障)
📌 技术栈:Python3.13 + LangChain1.0 + 通义千问qwen-plus
{('='*90)}
【1. 携程机票预订结果】
{ctrip_result}
【2. 美团酒店预订结果】
{meituan_result}
【3. 滴滴打车预约结果】
{didi_result}
{('='*90)}
💡 结论:A2A协作逻辑正常,各环节衔接顺畅,可直接适配真实业务接口!
"""
return total_report
return RunnableLambda(a2a_schedule)
关键特性:
- 顺序调度:严格按照业务逻辑依次调用各Agent
- 异常兜底:每个Agent调用都有try-catch,失败时使用工具原始函数
- 结果整合:生成结构化的最终报告,提升用户体验
7. 主程序执行
python复制if __name__ == "__main__":
try:
# 初始化各专属Agent
print("🔧 初始化携程/美团/滴滴专属Agent...")
ctrip_chain = create_ctrip_agent(llm)
meituan_chain = create_meituan_agent(llm)
didi_chain = create_didi_agent(llm)
print("✅ 所有专属Agent初始化完成!\n" + "="*90 + "\n")
# 初始化总协调Agent
print("🔧 初始化A2A总协调Agent(核心调度器)...")
coor_chain = create_travel_coordinator_agent(llm, ctrip_chain, meituan_chain, didi_chain)
print("✅ 总协调Agent初始化完成!\n" + "="*90 + "\n")
# 执行A2A协作流程
print("🚀 携程-美团-滴滴 A2A协作测试正式开始 🚀")
final_result = coor_chain.invoke({"input": "安排2026-02-01北京飞上海的完整行程"})
# 打印最终报告
print("\n" + "="*90)
print(final_result)
print("="*90)
except Exception as e:
print(f"❌ 全局运行异常:{type(e).__name__} - {str(e)[:100]}")
print("💡 快速排查指南:\n1. 确认通义千问API Key正确且有效\n2. 检查网络是否能访问阿里云dashscope\n3. 确认LangChain版本为1.0.0(非1.0版本接口不兼容)")
8. 运行结果分析
执行程序后,我们会看到完整的A2A协作流程:
- 首先初始化所有专属Agent
- 然后初始化总协调Agent
- 开始执行协作流程:
- 调用携程机票Agent预订机票
- 调用美团酒店Agent预订酒店
- 调用滴滴打车Agent预约车辆
- 最后生成整合报告
关键输出示例:
code复制【携程机票预订成功】
出发地:北京
目的地:上海
出行日期:2026-02-01
航班号:CA1885(北京首都T3→上海浦东T2)
起飞时间:14:00
降落时间:16:30
座位:经济舱34A
电子客票号:999-1234567890
舱位等级:经济舱超级经济座
9. 最佳实践与经验分享
9.1 工具封装规范
- 描述必须准确:工具描述要明确参数名、类型和格式
- 单一功能原则:每个工具只做一件事
- 保留原始函数:通过tool.func保留原始函数用于兜底
9.2 Agent设计规范
- 严格约束Prompt:明确指定工具调用格式和返回要求
- 统一接口:所有Agent都封装为Runnable链
- 异常处理:每个Agent调用都要有异常捕获
9.3 协调器设计要点
- 流程控制:明确各Agent的调用顺序
- 结果整合:生成用户友好的最终报告
- 稳定性保障:完善的错误处理和兜底机制
10. 常见问题排查
在实际开发中,我遇到了几个典型问题:
-
Agent返回空结果
- 原因:Prompt约束不够严格
- 解决:在Prompt中明确要求必须调用工具并返回结果
-
工具调用参数错误
- 原因:工具描述不够详细
- 解决:补充参数名、类型和格式到description中
-
版本兼容性问题
- 现象:Runnable not found错误
- 解决:确保使用LangChain1.0.0,旧版接口不兼容
11. 扩展与优化方向
这个基础框架还可以进一步优化:
- 动态参数支持:从用户输入中提取动态参数,替代固定值
- 异步调用:对无依赖的Agent使用并行调用提高效率
- 错误重试:增加自动重试机制提升稳定性
- 真实API集成:替换模拟工具为真实平台接口
- 自然语言交互:支持更灵活的用户输入方式
12. 项目总结
通过这个项目,我深刻体会到LangChain1.0在构建复杂AI应用时的价值。A2A协作模式将复杂问题分解为多个专业Agent协同解决,既保证了每个环节的专业性,又实现了全流程的自动化。这种架构特别适合电商、出行、金融等需要整合多个服务的场景。
关键收获:
- 清晰的架构设计比复杂的算法更重要
- 完善的错误处理是系统稳定的关键
- 统一的接口规范大大降低了维护成本
这个方案已经在我们公司的多个项目中得到应用,效果非常不错。如果你也面临类似的多服务整合需求,不妨试试这个A2A协作模式。
