第五次作业：MCP

05_MCP/agent.py

@@ -0,0 +1,43 @@
+from llm import create_model
+from config import load_config
+from langchain_mcp_adapters.client import MultiServerMCPClient
+from langchain.agents import create_agent
+import asyncio
+mcp_client=MultiServerMCPClient(
+    {
+        "amap": {
+      "transport": "streamable_http",
+      "url": "https://mcp.api-inference.modelscope.net/847991bb62714b/mcp"
+    }
+    }
+)
+
+config=load_config()
+llm = create_model(config)
+
+async def ask_agent(question:str)->str:
+    tools= await mcp_client.get_tools()
+    agent =  create_agent(
+        model=llm,
+        tools=tools,
+        system_prompt="你是⼀个出⾏助⼿，可以使⽤地图⼯具帮⽤户查天⽓、规划路线。回答要简洁，不要编造数据。",
+        )
+    # 第三步：执⾏对话，让 LLM ⾃行判断是否需要调⽤⼯具
+    result =await agent.ainvoke({
+    "messages": [{"role": "user", "content": question}]
+    })
+    # 第四步：返回最后⼀条消息（即 LLM 的最终回复）
+    return result["messages"][-1].content
+
+
+async def main():
+     # 查天⽓——LLM 会⾃动调⽤ amap 的天⽓⼯具
+    weather = await ask_agent("成都今天天⽓怎么样？")
+    print(f"天⽓查询结果：{weather}\n")
+    # 规划路线——不同的问题，LLM 会选择不同的⼯具
+    route =await ask_agent("帮我规划⼀条从春熙路到双流机场的地铁换乘路线")
+    print(f"路线规划结果：{route}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

05_MCP/config.py

@@ -0,0 +1,29 @@
+import os
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class AppConfig:
+    api_key: str
+    api_url: str
+    model: str
+    database_url: str
+    history_table_name: str
+    default_session_id: str
+
+def load_config() -> AppConfig:
+    api_key = os.getenv("API_KEY")
+    if not api_key:
+        raise ValueError("API_KEY is not set in environment variables.")
+
+    return AppConfig(
+        api_key=api_key,
+        api_url=os.getenv(
+            "API_URL",
+            "https://dashscope.aliyuncs.com/compatible-mode/v1",
+        ),
+        model=os.getenv("MODEL", "qwen3.6-plus"),
+        database_url=os.getenv("DATABASE_URL", "mysql+pymysql://root:1234@localhost:3306/mydb"),
+        history_table_name=os.getenv("HISTORY_TABLE_NAME", "chat_history"),
+        default_session_id=os.getenv("SESSION_ID", "user_001"),
+    )

05_MCP/llm.py

@@ -0,0 +1,10 @@
+from langchain_openai import ChatOpenAI
+from config import AppConfig
+
+def create_model(config: AppConfig) -> ChatOpenAI:
+    return ChatOpenAI(
+        model_name=config.model,
+        openai_api_key=config.api_key,
+        openai_api_base=config.api_url,
+        temperature=0.7,
+    )

05_MCP/main.py

@@ -0,0 +1,7 @@
+import travel_agent.graph
+import asyncio
+def main():
+    asyncio.run(travel_agent.graph.main())
+
+if __name__ == "__main__":
+    main()

05_MCP/travel_agent/agents.py

@@ -0,0 +1,122 @@
+from .state import TravelState
+from .tools import get_transport_tools,get_hotel_tools,get_attraction_tools
+from langchain.agents import create_agent
+from llm import create_model
+from config import load_config
+
+
+config=load_config()
+llm = create_model(config)
+
+def coordinator_node(state: TravelState)->dict:
+    return{}
+#路线agent节点
+async def transport_agent_node(state: TravelState) -> dict:
+
+    user_request= state["user_request"]
+    tools=await get_transport_tools()
+    agent = create_agent(
+        model=llm,
+        tools=tools,
+        system_prompt="""你是旅行规划系统中的交通专家，只负责交通与市内通勤建议。
+根据用户需求，必要时调用地图工具查询路线、距离或交通方式。
+优先提供公共交通建议；用户明确自驾时再提供驾车建议。
+若用户未提供出发地、日期等关键条件，应明确说明你的假设或待确认信息。
+不得推荐酒店、景点或编造工具未返回的路线数据。
+输出应包含：推荐交通方式、预计耗时、关键换乘/路线信息、出行提醒。"""
+    )
+    result = await agent.ainvoke({"messages":[{"role":"user","content":user_request}]})
+    return {"transport_plan":result["messages"][-1].content}
+
+#酒店agent节点
+async def hotel_agent_node(state:TravelState)->dict:
+
+    user_request= state["user_request"]
+    tools=await get_hotel_tools()
+    agent = create_agent(
+            model=llm,
+            tools=tools,
+            system_prompt="""你是旅行规划系统中的住宿专家，只负责查询与推荐住宿。
+
+当用户给出了目的地时，必要时调用 hotel_search_and_recommend。
+调用时：
+- destination 必填，从用户请求中提取城市、区域或地标；
+- query 传入完整用户原始需求；
+- scene 可映射为商务、亲子、度假、背包或通用；
+- 用户明确给出入住/退房日期时，才传 YYYY-MM-DD 格式的 check_in/check_out；
+- guests 取用户明确人数，未说明时可省略；
+- max_price 只在用户明确给出“每晚预算”时传入；总旅行预算不能直接当作每晚预算。
+
+如果用户没有目的地，不调用工具，并要求补充目的地。
+不得编造酒店名称、价格、位置、评分或是否有房。
+推荐结果不代表已完成预订。
+不得规划交通、景点或完整行程。
+
+输出格式：
+1. 推荐住宿
+2. 每晚价格与位置
+3. 适合原因
+4. 需用户确认的信息"""
+        )
+    result = await agent.ainvoke({"messages":[{"role":"user","content":user_request}]})
+    return {"hotel_plan":result["messages"][-1].content}
+
+#景点agent节点
+async def attraction_agent_node(state:TravelState)->dict:
+
+    user_request= state["user_request"]
+    tools=await get_attraction_tools()
+    agent = create_agent(
+                model=llm,
+                tools=tools,
+                system_prompt="""你是旅行规划系统中的景点与活动专家，只负责景点推荐和游玩建议。
+用户提供目的地时，必须调用地图工具查询景点或天气后再回答。
+优先使用关键词搜索查找与用户偏好匹配的 POI。
+需要了解特定地点时，使用详情搜索；需要查询某个地点周边内容时，先获得其坐标，再使用周边搜索。
+根据天气工具结果给出合理出行提醒。
+
+不得虚构景点地址、营业时间、门票、评分、距离或天气数据。
+不得推荐酒店、规划交通路线或声称已完成预约。
+
+输出应包含：
+1. 推荐景点或区域；
+2. 每个推荐的类型、位置或商圈信息；
+3. 适合用户偏好的原因；
+4. 天气与游玩提醒；
+5. 需要用户确认的信息。"""
+            )
+    result = await agent.ainvoke({"messages":[{"role":"user","content":user_request}]})
+    return {"attraction_plan":result["messages"][-1].content}
+
+#整合agent
+async def integrator_agent_node(state: TravelState) -> dict:
+
+    user_request= state["user_request"]
+    transport_plan=state.get("transport_plan")
+    hotel_plan=state.get("hotel_plan")
+    attraction_plan=state.get("attraction_plan")
+    agent = create_agent(
+                    model=llm,
+                    tools=[],
+                    system_prompt="""你是旅行规划系统中的方案整合专家。
+只能依据用户需求及交通、住宿、景点专家提供的结果生成方案。
+不得自行调用工具、补充未经专家结果支持的酒店、路线、景点价格、天气或营业信息。
+若专家结果冲突、缺失或信息不足，明确标记“待确认”，不要编造。
+
+输出格式：
+1. 旅行概览
+2. 交通与市内通勤
+3. 住宿建议
+4. 每日行程建议
+5. 注意事项与待确认事项"""
+                )
+    compose_text=f"""【用户需求】{user_request}
+    【交通专家结果】
+    {transport_plan}
+    【住宿专家结果】
+    {hotel_plan}
+    【景点专家结果】
+    {attraction_plan}
+    """
+    result=await agent.ainvoke({"messages":[{"role":"user","content":compose_text}]})
+    return {"final_plan":result["messages"][-1].content}

05_MCP/travel_agent/graph.py

@@ -0,0 +1,34 @@
+from langgraph.graph import StateGraph,START,END
+from .state import TravelState
+from .agents import transport_agent_node,hotel_agent_node,attraction_agent_node,coordinator_node,integrator_agent_node
+
+def build_graph():
+    builder=StateGraph(TravelState)
+    #添加节点
+    builder.add_node("coordinator",coordinator_node)
+    builder.add_node("transport",transport_agent_node)
+    builder.add_node("hotel",hotel_agent_node)
+    builder.add_node("attraction",attraction_agent_node)
+    builder.add_node("integrator",integrator_agent_node)
+    #添加边(并行)
+    builder.add_edge(START,"coordinator")
+    builder.add_edge("coordinator","transport")
+    builder.add_edge("coordinator","hotel")
+    builder.add_edge("coordinator","attraction")
+
+    builder.add_edge("transport","integrator")
+    builder.add_edge("hotel","integrator")
+    builder.add_edge("attraction","integrator")
+    builder.add_edge("integrator",END)
+
+    #编译图
+    graph= builder.compile()
+
+    return graph
+
+async def main():
+    graph=build_graph()
+    result=await graph.ainvoke({"user_request":"成都三日游，两人，喜欢美食和人文"})
+    print(result["final_plan"])
+
+    

05_MCP/travel_agent/state.py

@@ -0,0 +1,10 @@
+from typing import TypedDict,NotRequired
+class TravelState(TypedDict):
+    """user_request：用户原始自然语言请求；
+    三个 *_plan：对应专家 Agent 的输出；
+    final_plan：整合 Agent 的最终结果"""
+    user_request: str
+    transport_plan: NotRequired[str]
+    hotel_plan: NotRequired[str]
+    attraction_plan: NotRequired[str]
+    final_plan: NotRequired[str]

05_MCP/travel_agent/tools.py

@@ -0,0 +1,59 @@
+from langchain_mcp_adapters.client import MultiServerMCPClient
+
+import asyncio
+mcp_client=MultiServerMCPClient(
+    {
+        "amap": {
+      "transport": "streamable_http",
+      "url": "https://mcp.api-inference.modelscope.net/75ce1803fffa45/mcp"
+    },
+    
+    
+ 
+    "hotel-recommend": {
+      "transport": "streamable_http",
+      "url": "https://mcp.api-inference.modelscope.net/0c1ea913c4ef4f/mcp"
+    }
+}
+)
+_mcp_all_tools=None
+_tools_lock = asyncio.Lock()
+
+async def get_tools():
+    global _mcp_all_tools
+    if _mcp_all_tools is None:
+        async with _tools_lock:
+            if _mcp_all_tools is None:
+                _mcp_all_tools = await mcp_client.get_tools()
+    return _mcp_all_tools
+
+
+async def get_transport_tools():
+    """获取地图交通工具"""
+    all_mcp_tools=await get_tools()
+    transport_tools_names=["maps_geo","maps_direction_transit_integrated","maps_direction_driving","maps_direction_walking","maps_bicycling","maps_distance"]
+    transport_tools=[tool for tool in all_mcp_tools if tool.name in transport_tools_names]
+    return transport_tools
+
+async def get_hotel_tools():
+    """获取酒店工具"""
+    all_mcp_tools=await get_tools()
+    hotel_tools_names=["hotel_search_and_recommend"]
+    hotel_tools=[tool for tool in all_mcp_tools if tool.name in hotel_tools_names]
+    return hotel_tools
+
+async def get_attraction_tools():
+    """获取POI信息工具"""
+    all_mcp_tools=await get_tools()
+    attraction_tools_names=["maps_geo","maps_text_search","maps_around_search","maps_search_detail","maps_weather"]
+    attraction_tools=[tool for tool in all_mcp_tools if tool.name in attraction_tools_names]
+    return attraction_tools
+
+async def main():
+    tools = await get_tools()
+    print(f"可用工具：{[tool.name for tool in tools]}\n")
+
+
+
+if __name__ == "__main__":
+    asyncio.run(main())