LangGraph 系列 13：使用 Bing 与 12306 MCP 构建 Workflow

langgraph workflow mcp
langgraph
发布日期:   2026-06-04
文章字数:   3k
阅读时长:   12 分
阅读次数:  

本文把两个公网 MCP Server 接入同一张 LangGraph：

  • Bing MCP 负责搜索网页和抓取搜索结果。
  • 12306 MCP 负责日期、车站编码、余票、中转和经停站查询。

模型继续使用本地 Qwen3。公网服务负责提供实时数据，本地模型负责选择工具和整理回答。

1. 完整案例结构

双 MCP 构成的 Workflow

一次请求可能经过下面的路径：

用户问题
-> 本地 Qwen3
-> tools_condition
-> ToolNode
-> MCP Tool
-> ToolMessage
-> 本地 Qwen3
-> 最终回答

如果问题不需要外部数据，tools_condition 会直接结束流程。如果模型生成 Tool Call，流程进入 ToolNode；工具结果回来后再回到 Qwen3。

2. 启动本地 Qwen3

激活现有的 .venv_tool_server：

cd /path/to/llm-learning
source .venv_tool_server/bin/activate

"$VIRTUAL_ENV/bin/python" -m mlx_lm server \
  --model Qwen3-14B-AWQ-4bit-MLX \
  --host 127.0.0.1 \
  --port 18080 \
  --prompt-cache-size 0 \
  --chat-template-args '{"enable_thinking": false}'

–prompt-cache-size 0 是本次实测后的必要补充。开启缓存连续测试多个不同问题时，最终回答曾混入上一个车站查询的内容。关闭 Prompt Cache 后，四组独立问题没有再发生串入。

3. 配置两个 MCP Server

客户端环境继续使用 .venv_mcp：

source .venv_mcp/bin/activate
python -m pip install -r langgraph/p13_mcp_secretary_workflow/requirements.txt

两个服务放在同一个 MultiServerMCPClient 配置中：

MCP_SERVERS = {
    "bing_cn": {
        "transport": "http",
        "url": "https://mcp.api-inference.modelscope.net/f07308ccca444b/mcp",
    },
    "railway_12306": {
        "transport": "http",
        "url": "https://mcp.api-inference.modelscope.net/5699af7157784b/mcp",
    },
}

client = MultiServerMCPClient(MCP_SERVERS)
tools = await client.get_tools()

配置名称 bing_cn 和 railway_12306 用于区分连接。真正发送给模型的是 MCP Server 返回的 Tool 名称。

4. 查看 MCP 提供的工具

运行：

python langgraph/p13_mcp_secretary_workflow/01_list_mcp_tools.py

本次连接共返回十个工具：

MCP Server 工具 作用
Bing bing_search 根据关键词搜索网页
Bing crawl_webpage 根据搜索结果抓取网页正文
12306 get-current-date 获取上海时区日期，解析相对日期
12306 get-stations-code-in-city 查询城市中的全部车站
12306 get-station-code-of-citys 查询城市代表编码
12306 get-station-code-by-names 根据具体车站名查询编码
12306 get-station-by-telecode 根据 Telecode 查询车站详情
12306 get-tickets 查询直达余票
12306 get-interline-tickets 查询中转余票
12306 get-train-route-stations 查询列车经停站

外部 MCP 的工具可能调整，因此代码不把“十个”作为永久断言，只检查本案例真正依赖的工具：

REQUIRED_TOOLS = {
    "bing_search",
    "get-current-date",
    "get-station-code-by-names",
    "get-tickets",
}

missing_names = REQUIRED_TOOLS - {tool.name for tool in tools}
if missing_names:
    raise RuntimeError(f"MCP Server 缺少必需工具：{sorted(missing_names)}")

本次实测中，MCP Adapter 返回的 args_schema 是普通 dict。应直接序列化：

print(json.dumps(current_tool.args_schema, ensure_ascii=False, indent=2))

不能假设它一定是 Pydantic 类并调用 .model_json_schema()，否则会产生 AttributeError。

5. 先直接测试 MCP Tool

在加入模型和 Workflow 前，先直接调用工具。这样能把“外部服务失败”和“模型没有选择工具”分开排查。

MCP_TIMEOUT_SECONDS = 45

# 先把工具列表转换为“名称 → 工具对象”的字典。
tools = {}
for current_tool in await client.get_tools():
    tools[current_tool.name] = current_tool


async def test_station_codes(tools: dict[str, Any]) -> None:
    """查询两个车站，并检查服务返回的车站编码。"""

    station_result = await asyncio.wait_for(
        tools["get-station-code-by-names"].ainvoke(
            {"stationNames": "杭州东|上海虹桥"}
        ),
        timeout=MCP_TIMEOUT_SECONDS,
    )
    station_codes = json.loads(extract_text(station_result))

    assert station_codes["杭州东"]["station_code"] == "HGH"
    assert station_codes["上海虹桥"]["station_code"] == "AOH"

余票具有时效性，脚本使用运行后三天的日期，只检查接口是否正常返回，不把票价和余票数量写成长期固定结论：

travel_date = (date.today() + timedelta(days=3)).isoformat()
ticket_result = await asyncio.wait_for(
    tools["get-tickets"].ainvoke(
        {
            "date": travel_date,
            "fromStation": "HGH",
            "toStation": "AOH",
            "trainFilterFlags": "G",
            "sortFlag": "startTime",
            "limitedNum": 2,
            "format": "text",
        }
    ),
    timeout=MCP_TIMEOUT_SECONDS,
)

如果公网 MCP 在 45 秒内没有返回，asyncio.wait_for() 会结束等待，并由脚本输出清晰错误。这样不会把外部服务偶发延迟误判成程序死循环。

真实稳定输出：

Bing 搜索返回非空： True
杭州东车站编码： HGH
上海虹桥车站编码： AOH
余票查询返回非空： True

6. 创建智能 Workflow

03_secretary_workflow.py 的核心代码如下：

async def create_secretary_graph():
    """异步加载 MCP Tools，再编译模型与工具循环。"""

    # 第 1 步：从两个 MCP Server 加载工具。
    mcp_client = MultiServerMCPClient(MCP_SERVERS)
    tools = await mcp_client.get_tools()

    # 名称集合用于检查缺失项，查找字典用于按名称取得工具。
    tool_names = set()
    tools_by_name = {}
    for current_tool in tools:
        tool_names.add(current_tool.name)
        tools_by_name[current_tool.name] = current_tool

    missing_names = REQUIRED_TOOLS - tool_names
    if missing_names:
        raise RuntimeError(f"MCP Server 缺少必需工具：{sorted(missing_names)}")

    # 第 2 步：创建本地模型并绑定 MCP 工具。
    # trust_env=False 确保本机模型请求不经过系统代理。
    http_client = httpx.Client(trust_env=False)
    model = ChatOpenAI(
        model="Qwen3-14B-AWQ-4bit-MLX",
        base_url="http://127.0.0.1:18080/v1",
        api_key="not-needed",
        temperature=0,
        max_tokens=512,
        http_client=http_client,
    )
    model_with_tools = model.bind_tools(tools)
    model_with_bing_search = model.bind_tools(
        [tools_by_name["bing_search"]],
        tool_choice="required",
    )

    def chatbot(state: MessagesState) -> dict:
        """让 Qwen3 根据用户消息和工具结果决定下一步。"""

        # 每次进入 chatbot 节点时，都把系统规则放在消息列表最前面。
        system_message = SystemMessage(
            content=(
                "你是一个中文智能。需要互联网资料时使用 Bing 工具；"
                "需要火车站或余票信息时使用 12306 工具。"
                "相对日期必须先调用 get-current-date。"
                "不要编造工具没有返回的信息。用户明确要求不调用工具时直接回答。"
            )
        )
        last_message = state["messages"][-1]
        if isinstance(last_message, HumanMessage) and "搜索" in last_message.content:
            response = model_with_bing_search.invoke(
                [system_message, *state["messages"]]
            )
        else:
            response = model_with_tools.invoke([system_message, *state["messages"]])
        return {"messages": [response]}

    # 第 3 步：构建“模型 → 工具 → 模型”的循环。
    builder = StateGraph(MessagesState)
    builder.add_node("chatbot", chatbot)
    builder.add_node("tools", ToolNode(tools))
    builder.add_edge(START, "chatbot")
    builder.add_conditional_edges(
        "chatbot",
        tools_condition,
        {"tools": "tools", "__end__": END},
    )
    builder.add_edge("tools", "chatbot")
    return builder.compile()

MCP Tools 需要异步加载，因此图工厂也是 async def。ToolNode 可以执行这些异步 MCP Tools。

7. 搜索MCP绑定工具

第一次实测时，Qwen3 收到了全部十个工具。用户明确要求“搜索 LangGraph ToolNode”，模型却没有调用 bing_search，而是直接生成了一段不可靠的回答。

仅设置：

tool_choice="bing_search"

在当前 MLX-LM 环境中也没有稳定生效。最终使用的方式是：

model.bind_tools(
    [tools_by_name["bing_search"]],
    tool_choice="required",
)

当用户明确写出“搜索”时，只向模型暴露 bing_search，并要求必须调用。其他问题仍使用完整工具集合。

这不是替模型执行工具，而是应用层根据明确用户意图缩小可选范围。执行仍由 ToolNode 和 MCP Server 完成。

8. 调用 12306 工具

“三天后杭州东到上海虹桥最早的两趟高铁余票”不是一个 Tool Call 就能完成。

余票问题中的连续工具调用

模型依次需要：

  1. 调用 get-current-date，把“三天后”转换成日期。
  2. 调用 get-station-code-by-names，得到 HGH 和 AOH。
  3. 调用 get-tickets，带上日期、车站编码、车型和排序条件。

每次 Tool Call 执行后，ToolMessage 都回到 chatbot。模型读取已有结果，再生成下一个 Tool Call。

图中的三个工具不是彼此直接调用。每一步都完整经过一次“Qwen3 生成 Tool Call → ToolNode 执行 MCP Tool → ToolMessage 返回 Qwen3”的循环，后一个工具的参数来自模型对前面结果的读取和整理。

9. 运行完整 Workflow

代码:

"""运行双 MCP 智能小秘书，并以稳定格式展示工具调用过程。"""

import asyncio
from importlib import import_module
from typing import Any

import httpx
from langchain_core.messages import AIMessage, ToolMessage


# 文件名带执行编号，使用 import_module() 导入其中的图工厂。
create_secretary_graph = import_module(
    "03_secretary_workflow"
).create_secretary_graph


def format_exception(error: BaseException) -> str:
    """递归展开 ExceptionGroup，输出可直接定位的底层错误。"""

    children = getattr(error, "exceptions", ())
    if children:
        return " | ".join(format_exception(child) for child in children)
    return f"{type(error).__name__}: {error}"


def stable_arguments(arguments: dict[str, Any]) -> dict[str, Any]:
    """隐藏运行时日期，避免把实时字段写进稳定日志。"""

    result = dict(arguments)
    for key in ("date", "departDate"):
        if key in result:
            result[key] = "<运行时日期>"
    return result


def print_tool_messages(messages: list) -> list[str]:
    """打印工具调用过程，并返回实际调用过的工具名。"""

    called_tools = []
    for message in messages:
        if isinstance(message, AIMessage) and message.tool_calls:
            for call in message.tool_calls:
                called_tools.append(call["name"])
                print("调用工具：", call["name"])
                print("工具参数：", stable_arguments(call["args"]))
        elif isinstance(message, ToolMessage):
            print("工具结果已返回：", bool(message.content))
    return called_tools


def check_called_tools(called_tools: list[str], expected_tools: set[str]) -> None:
    """确认模型实际调用的工具符合当前测试案例。"""

    missing_tools = expected_tools - set(called_tools)
    if missing_tools:
        raise RuntimeError(f"模型没有调用预期工具：{sorted(missing_tools)}")
    if not expected_tools and called_tools:
        raise RuntimeError(f"本题不应调用工具，实际调用：{called_tools}")


async def run_question(
    graph,
    question: str,
    expected_tools: set[str],
    show_answer: bool = True,
) -> None:
    """执行一个问题并打印工具名称、参数和最终回答。"""

    # 第 1 步：把问题交给完整工作流。
    print(f"\n用户问题：{question}")
    result = await graph.ainvoke(
        {"messages": [{"role": "user", "content": question}]},
        config={"recursion_limit": 20},
    )

    # 第 2 步：打印消息中的工具调用，并收集工具名称。
    called_tools = print_tool_messages(result["messages"])

    # 第 3 步：检查工具调用是否符合当前测试案例的预期。
    check_called_tools(called_tools, expected_tools)
    final_answer = str(result["messages"][-1].content).strip()

    # show_answer=False 用于实时余票，避免把动态结果写进固定日志。
    if show_answer:
        print("最终回答：", final_answer)
    else:
        print("最终回答非空：", bool(final_answer))


async def main() -> None:
    """检查本地模型服务后运行四种典型问题。"""

    # 先检查本地模型服务，避免在完整工作流中排查连接问题。
    with httpx.Client(trust_env=False, timeout=10) as client:
        response = client.get("http://127.0.0.1:18080/v1/models")
        response.raise_for_status()

    # 创建一次图，然后用四个问题复用它。
    graph = await create_secretary_graph()
    await run_question(
        graph,
        "请用一句话说明智能小秘书能做什么，不要调用工具。",
        expected_tools=set(),
    )
    await run_question(
        graph,
        "请查询杭州东和上海虹桥的车站编码。",
        expected_tools={"get-station-code-by-names"},
    )
    await run_question(
        graph,
        "请搜索 LangGraph ToolNode 的作用，并用一句中文总结。",
        expected_tools={"bing_search"},
    )
    await run_question(
        graph,
        "请查询三天后杭州东到上海虹桥最早的两趟高铁余票。",
        expected_tools={
            "get-current-date",
            "get-station-code-by-names",
            "get-tickets",
        },
        show_answer=False,
    )


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except Exception as exc:
        raise SystemExit(
            f"智能小秘书 Workflow 执行失败：{format_exception(exc)}"
        ) from exc

运行：

python langgraph/p13_mcp_secretary_workflow/04_run_secretary_workflow.py

运行脚本把输出和校验分别放进 print_tool_messages() 与 check_called_tools()。run_question() 只保留“调用图、展示消息、检查工具、输出答案”四个步骤。脚本会断言每组问题必须出现预期工具，防止“模型直接编造答案”被误写成成功结果。

一次实测的稳定字段与模型回答如下：

用户问题：请用一句话说明智能能做什么，不要调用工具。
最终回答：智能可以协助您查询火车票余票、搜索互联网信息、获取车站代码、查询列车路线等。

用户问题：请查询杭州东和上海虹桥的车站编码。
调用工具： get-station-code-by-names
工具参数： {'stationNames': '杭州东|上海虹桥'}
工具结果已返回： True
最终回答：杭州东的车站编码是 **HGH**，上海虹桥的车站编码是 **AOH**。

用户问题：请搜索 LangGraph ToolNode 的作用，并用一句中文总结。
调用工具： bing_search
工具参数： {'query': 'LangGraph ToolNode 的作用', 'count': 5}
工具结果已返回： True
最终回答：LangGraph 的 ToolNode 用于在构建有状态、多步骤的 LLM 应用时，提供一个低层级的 Agent 编排框架，帮助开发者管理复杂的 AI 工作流。

用户问题：请查询三天后杭州东到上海虹桥最早的两趟高铁余票。
调用工具： get-current-date
调用工具： get-station-code-by-names
调用工具： get-tickets
最终回答非空： True

最后一组不展示实时日期、余票和票价，只保留调用顺序。模型回答存在非确定性，工具名称、参数结构和调用链才是本案例需要稳定验证的部分。

11. 总结

智能的核心不是“绑定很多工具”，而是让模型、工具节点、外部 MCP 和状态路由形成可验证的循环。下一篇会在工具真正产生副作用前暂停流程，加入第一种人工审批方式。

文章作者: hnbian
文章链接: https://www.hnbian.cn/posts/beeb3f66.html
版权声明: 本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源 hnbian !
langgraph workflow mcp
评论
  目录