LangGraph workflow详解以及MCP调用

from typing import TypedDict, Literal
import os
from langgraph.graph import StateGraph, START, END
from langchain_community.chat_models.tongyi import ChatTongyi


# --- 1. State 定义 (白板) ---
class JokeState(TypedDict):
    topic: str
    joke: str
    feedback: str
    # 记录重试次数，防止死循环
    retry_count: int

# 设置环境变量 - 使用阿里云千问API Key
os.environ["DASHSCOPE_API_KEY"] = "your-dashscope-api-key-here"


# --- 2. Nodes 定义 (干活的人) ---
llm = ChatTongyi(model="qwen-plus")  # 使用千问plus模型


def generate_joke(state: JokeState):
    msg = f"讲一个关于 {state['topic']} 的笑话。"
    response = llm.invoke(msg)
    # 更新 state：覆盖旧笑话，并增加重试计数
    return {
        "joke": response.content,
        "retry_count": state.get("retry_count", 0) + 1
    }


def critique_joke(state: JokeState):
    msg = f"评价这个笑话是否好笑，只回答'yes'或'no'：{state['joke']}"
    response = llm.invoke(msg)
    return {"feedback": response.content.lower()}


# --- 3. Edges 定义 (SOP 流程) ---
def router(state: JokeState) -> Literal["generate", END]:
    # 强制熔断机制：如果试了3次还不行，强制结束
    if state["retry_count"] > 3:
        return END

    if "no" in state["feedback"]:
        return "generate"  # 循环！回到生成节点
    return END


# --- 4. 组装图 ---
builder = StateGraph(JokeState)

builder.add_node("generator", generate_joke)
builder.add_node("critic", critique_joke)

builder.add_edge(START, "generator")
builder.add_edge("generator", "critic")

# 关键的循环逻辑
builder.add_conditional_edges("critic", router, {
    "generate": "generator",
    END: END
})

app = builder.compile()

# --- 5. 运行 ---
inputs = {"topic": "程序员", "retry_count": 0}
# 像流一样运行
for output in app.stream(inputs):
    print(output)

from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient(
    {
        "weather": {
            "url": "http://localhost:8000/mcp",
            "transport": "http",
        }
    }
)
tools = await client.get_tools()

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode

# 1. 建立 MCP 连接 (这是一个上下文管理器)
async def create_mcp_tools():
    # 配置你的 MCP Server 路径
    server_params = StdioServerParameters(
        command="python", 
        args=["mcp_server_script.py"] 
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # 2. 获取 MCP Server 里的所有工具列表
            result = await session.list_tools()
            mcp_tools = result.tools

            langchain_tools = []

            # 3. 【核心步骤】将 MCP 工具动态转换为 LangChain 工具
            for mcp_tool in mcp_tools:

                # 定义一个闭包函数来执行具体的 MCP 调用
                async def _dynamic_tool_func(**kwargs):
                    return await session.call_tool(mcp_tool.name, arguments=kwargs)

                # 设置函数的元数据（这很重要，LLM 靠这个理解工具怎么用）
                _dynamic_tool_func.__name__ = mcp_tool.name
                _dynamic_tool_func.__doc__ = mcp_tool.description

                # 使用 LangChain 的装饰器或构造函数包装
                # 这里简化演示，实际需要处理 input_schema 转换
                lc_tool = tool(_dynamic_tool_func) 
                langchain_tools.append(lc_tool)

            return langchain_tools

from langchain_core.tools import tool

def _create_tools(self):
    """创建工具集，包括 MCP 工具"""
    @tool
    def calculate(expression: str) -> str:
        """执行数学计算

        Args:
            expression: 数学表达式，例如 '2+2' 或 '10*5'

        Returns:
            计算结果
        """
        try:
            result = eval(expression)
            return f"计算结果：{expression} = {result}"
        except Exception as e:
            return f"计算错误：{str(e)}"

    return [calculate]

# 绑定工具到 LLM
llm = ChatOpenAI(model="gpt-4o")
llm_with_tools = llm.bind_tools(tools)

builder.add_node("tools", ToolNode(tools)) 

你是一个自主规划助手。你的任务是：
1. 理解用户的请求
2. 分析需要执行哪些操作
3. 规划执行步骤
4. 调用相应的工具完成任务
你可以使用的工具包括：
- get_weather: 查询天气预报
- search_web: 搜索互联网信息
- calculate: 执行数学计算
请根据用户的需求，智能地规划并执行任务。

def _execution_node(self, state: AgentState) -> AgentState:
    """执行节点：执行规划好的任务"""
    messages = state["messages"]
    last_message = messages[-1]

    # 如果最后一条消息包含工具调用，则执行工具
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return {
            "messages": messages,
            "next": "continue"
        }

    # 如果没有工具调用，说明任务完成
    return {
        "messages": messages,
        "next": "end"
    }