LANGCHAIN MODULE 2 · LESSON 2.4 Wedding Planners

1 项目结构：一个协调者，三个专家

这个 notebook 是 Module 2 的综合实战。主 Agent 扮演 wedding coordinator，先收集必要字段并写入 state，然后委托三个专家 Agent 分别处理航班、场地和歌单。

角色	工具来源	负责内容
Travel agent	Kiwi MCP tools	搜索目的地往返航班
Venue agent	`web_search`	搜索符合地点和人数的婚礼场地
Playlist agent	`query_playlist_db`	查询 Chinook SQLite 数据库并策划歌单
Coordinator	包装后的专家工具 + `update_state`	收集需求、分派任务、整合结果

2 给远程 MCP 加重试拦截器

真实远程工具会失败。Notebook 先定义 RetryMCPInterceptor，用于重试 transient MCP 错误，并把不可恢复错误转成可读信息。

Python

RETRYABLE_MCP_CODES = {-32603}

class RetryMCPInterceptor:
    """Intercept MCP tool calls: retry transient failures, surface all errors gracefully."""

client = MultiServerMCPClient(
    {
        "travel_server": {
            "transport": "streamable_http",
            "url": "https://mcp.kiwi.com",
        }
    },
    tool_interceptors=[RetryMCPInterceptor()],
)

tools = await client.get_tools()

生产意识

课程前几节重点是“能调用”。这一节开始处理“调用会失败怎么办”。远程 MCP、搜索和数据库都需要明确的错误处理边界。

3 准备 Web Search 和 SQL 工具

场地专家使用 Tavily 搜索；歌单专家查询本地 Chinook 数据库。两个工具都通过 @tool 暴露给对应专家 Agent。

Python

tavily_client = TavilyClient()
search_count = 0

@tool
def web_search(query: str) -> Dict[str, Any]:
    """Search the web for information."""
    # call Tavily and return search results


db = SQLDatabase.from_uri("sqlite:///resources/Chinook.db")

@tool
def query_playlist_db(query: str) -> str:
    """Query the database for playlist information"""
    try:
        return db.run(query)
    except Exception as e:
        return f"Error: {e}"

工具描述很重要

多 Agent 场景中，工具 docstring 就是模型的操作说明。描述越清楚，专家 Agent 越不容易调用错参数。

4 定义 WeddingState

Coordinator 需要先从用户输入中抽取婚礼需求，并写进 state。Notebook 的 WeddingState 包含出发地、目的地、宾客人数和音乐类型。

Python

class WeddingState(AgentState):
    origin: str
    destination: str
    guest_count: str
    genre: str

origin

从哪里出发。

destination

婚礼目的地。

guest_count

场地容量约束。

genre

歌单风格。

5 创建三个专家 Agent

每个专家 Agent 都只拿自己的工具和 system prompt。这样比让一个大 Agent 同时理解航班、场地、SQL 更容易控制。

Python

today = date.today().isoformat()

travel_agent = create_agent(
    model=deepseek_model(),
    tools=tools,
    system_prompt=f"""
    You are a travel agent. Search for flights to the desired destination wedding location.
    Today is {today}. Any departureDate you pass to tools must be strictly after {today}.
    You are not allowed to ask any more follow up questions.
    """,
)

venue_agent = create_agent(
    model=deepseek_model(),
    tools=[web_search],
    system_prompt="""You are a venue specialist. Search for venues in the desired location.""",
)

playlist_agent = create_agent(
    model=deepseek_model(),
    tools=[query_playlist_db],
    system_prompt="""You are a playlist specialist. Query the SQL database and curate a wedding playlist.""",
)

职责隔离

专家 Agent 的 prompt 可以更短、更具体；主 Agent 只负责收集信息、委托和汇总。

6 Coordinator 通过 ToolRuntime 读取 state

Coordinator 的工具不是直接搜索，而是从 runtime.state 读取结构化字段，再调用对应专家 Agent。

Python

@tool
async def search_flights(runtime: ToolRuntime) -> str:
    """Travel agent searches for flights to the desired destination wedding location."""
    origin = runtime.state["origin"]
    destination = runtime.state["destination"]
    response = await travel_agent.ainvoke(
        {"messages": [HumanMessage(content=f"Find flights from {origin} to {destination}")]}
    )
    return response["messages"][-1].content

@tool
def search_venues(runtime: ToolRuntime) -> str:
    guest_count = runtime.state["guest_count"]
    destination = runtime.state["destination"]
    response = venue_agent.invoke(
        {"messages": [HumanMessage(content=f"Find venues in {destination} for {guest_count} guests")]}
    )
    return response["messages"][-1].content

为什么先写 state

如果不先结构化保存需求，每个专家都要从原始用户句子里重新抽字段，错误率会更高。

7 运行和流式输出

最后创建 coordinator，并用较高的 recursion_limit 给多轮工具调用留空间。Notebook 同时演示了普通 ainvoke 和 astream 流式输出。

Python

coordinator = create_agent(
    model=deepseek_model(),
    tools=[search_flights, search_venues, suggest_playlist, update_state],
    state_schema=WeddingState,
    system_prompt="""
    You are a wedding coordinator.
    First find all the information you need to update the state.
    Then delegate tasks to your specialists for flights, venues, and playlists.
    """,
)

response = await coordinator.ainvoke(
    {
        "messages": [
            HumanMessage(content="I'm from London and I'd like a wedding in Paris for 100 guests, jazz-genre")
        ]
    },
    config={"recursion_limit": 40},
)

综合 takeaway

这一课把 Module 2 的主线串起来：外部工具来自 MCP/搜索/SQL，临时需求进入 state，专家 Agent 各自解决子任务，Coordinator 负责组织最终答案。