Agent协作原理:条件路由与智能调度

架构设计哲学

传统的多Agent系统往往采用"全员并行"策略——每次请求都调用所有Agent。这在旅游规划场景下存在明显问题:

  1. 资源浪费:用户只问"杭州有什么好吃的",不需要调用路线规划Agent
  2. 响应延迟:所有Agent串行等待,增加总耗时
  3. 信息冗余:不相关的Agent输出干扰用户体验

TravelAgent-5采用条件路由策略:由Supervisor Agent分析用户意图,只调度真正需要的Agent。

LangGraph状态图

[用户输入] → [Supervisor] → 条件路由 → [Agent1] ─┐
                                    → [Agent2] ─┤
                                    → [Agent3] ─┼→ [聚合] → [输出]
                                    → [Agent4] ─┘

核心代码:

from langgraph.graph import StateGraph

graph = StateGraph(AgentState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("route", route_agent_node)
graph.add_node("hotel", hotel_agent_node)
# ...

# 条件路由:supervisor输出决定调用哪些Agent
graph.add_conditional_edges(
    "supervisor",
    route_decision,  # 返回目标节点列表
    {"route": "route", "hotel": "hotel", ...}
)

Supervisor的决策逻辑

Supervisor接收用户消息后,通过LLM分析意图,输出JSON格式的路由决策:

{
  "agents": ["route", "hotel", "food", "info"],
  "destination": "杭州",
  "days": 3,
  "budget": "5000元"
}
用户输入 调度的Agent
"规划3天杭州旅行" route, hotel, food, info
"推荐西湖附近酒店" hotel
"杭州有什么好吃的" food, info
"西湖的开放时间" route

工具层:飞猪MCP集成

每个Agent通过flyai-cli调用飞猪实时数据:

  • 路线Agentsearch-poi(景点搜索)
  • 住宿Agentsearch-hotel(酒店搜索)
  • 美食Agentkeyword-search(美食搜索)
  • 讲解Agentkeyword-search(文化搜索)

工具封装为Python函数,通过subprocess调用CLI,解析JSON返回。

SSE流式输出

后端通过FastAPI的StreamingResponse实现SSE推送,每完成一个Agent就立即向前端发送结果:

data: {"agent": "路线规划", "content": "## Day 1\n..."}
data: {"agent": "住宿安排", "content": "推荐3家酒店..."}
data: {"agent": "主管", "content": "完整方案...", "done": true}

前端根据agent字段显示不同颜色的卡片,实现渐进式加载体验。

导航

← 上一篇:前端架构设计:2核2G服务器上的极致轻量化方案 → 下一篇:技术选型与Agent设计详解