LangGraph 与 MCP 模型上下文协议简介与演示

一.LangGraph 与 MCP 简介

1.LangGraph

LangGraph 是由 LangChain 团队开发的开源框架，主要用于构建状态化、多代理的系统，能与大型语言模型很好地结合

核心定位

LangChain 生态成员：它是 LangChain 生态系统的重要组成部分，能与 LangChain 无缝集成，不过也可独立使用。
突破传统框架限制：传统的框架常依赖有向无环图，LangGraph 则支持循环图，这使其更契合真实业务场景中的迭代需求。
企业级生产工具：LangGraph 提供了持久化、错误恢复、人工审核等特性，可应用于企业生产环境，像 Uber、Klarna 等企业已经将其用于客服自动化、代码测试生成等方面。

核心能力

循环与分支：支持条件逻辑和循环流程，代理能够反复优化查询结果，直到达到满意的效果，适用于像是 RAG 增强检索等场景。
持久化状态管理：可自动保存图的每一步状态，支持在任何时间点暂停和恢复执行，以此实现错误恢复、时间回溯等功能。
人类在环机制 ：允许人工暂停流程，对决策进行审核，比如在医疗诊断场景中，专家可以介入审核 AI 给出的建议。
流式处理与实时监控 ：支持按照节点生成顺序进行流式输出，包括令牌流，还可以实时监控工具调用、数据检索等中间步骤。
多代理协作 ：可以构建分工明确的代理团队，例如安排 “规划代理” 去制定计划，“执行代理” 来调用工具，从而协同完成任务。

底层设计

状态管理：使用类似 JSON 的共享状态对象，每个节点执行后会更新全局状态，状态可以是消息列表或自定义数据结构。
节点与边：节点代表执行步骤，可能是 LLM 调用、工具函数或者自定义逻辑；边则用于定义节点之间的流转规则，支持根据 LLM 输出等条件判断来选择分支。

优势特点

灵活性：提供了更灵活的 Agent 框架，可支持定义复杂的流程，实现复杂的循环与条件逻辑，让多智能体系统的构建更为灵活。
可控性：能提供细致的流程和状态控制，同时内置持久性功能，支持高级的人工干预和记忆能力，方便用户对整个系统进行掌控和调整。

应用场景

客户服务自动化：可用来构建机器人和虚拟助手，自动完成客户服务流程，为客户提供个性化的支持。
数据检索与分析：能够创建可以执行复杂数据检索和分析任务的代理，使其自动从多个数据源里提取和汇总信息。

2.MCP 全称 Model Context Protocol

模型上下文协议，为什么会有这么一个协议？因为大模型都是基于先验知识训练而来，很多依赖于实时数据和现有工具集的任务，难以无缝集成到系统。因此， Anthropic 公司于 2024 年 11 月提出并开源了一种开放标准协议，旨在解决大型语言模型（LLM）与外部数据源和工具之间的集成问题 。
核心目标
MCP 主要是为了通过标准化接口，让 AI 系统能够无缝访问外部数据源、工具和服务，以此打破数据孤岛，提升 AI 助手的实际操作能力 。

架构组成

MCP Hosts：指的是 LLM 运行的应用程序环境，像 Cursor、Claude Desktop、Cline 等都属于此类，这些应用负责启动模型，并让模型能够运行在支持插件、具备上下文的环境中。
MCP Clients：客户端是运行在 Host 内部的桥梁，负责维护与 Server 的一对一连接，承担着请求转发、结果返回、权限验证等任务。
MCP Servers：是整个协议的关键角色，负责向 Client 暴露所有可以使用的工具、数据源、API，并通过统一格式，如 JSON Schema，来提供参数信息、调用权限、功能描述等内容，是模型和外部世界之间的连接桥梁。
Local Data Sources：也就是本地数据源，像用户的文件系统、本地数据库等，这些资源能够通过 MCP Server 封装后供模型调用。
Remote Services：指代外部的 API、SaaS 服务、云数据库等，例如 Notion、Google Drive 等，它们统一经 MCP Server 接入。

关键优势

标准化：提供了通用的接口，降低了为每个数据源去创建自定义连接器的必要，使得不同的 AI 模型和工具之间可以高效协作 。
安全性：采用本地服务器与数据源连接，防止敏感数据上传到第三方平台，并且支持加密通信与细粒度权限管理，能确保数据安全 。
灵活性：允许使用者在不同的 LLM 提供商和供应商之间进行切换，开发人员可根据需求选择不同模型和工具，不会被单一供应商锁定 。
可扩展性：架构较为灵活，支持模块化开发，多个 MCP 服务器可连接到单个主机，便于开发者自由组合服务器，构建复杂的功能链 。

应用场景

数据集成：能够连接 LLM 到企业数据源，例如文档库、CRM 系统或者内部知识库等，实现数据的统一整合与利用。
工具使用：可让 LLM 使用各类计算工具、API 或服务，拓展模型的功能边界，如使模型具备查询天气、发送邮件等能力。
开发环境：用于增强 IDE 和开发工具，让 AI 助手可以理解代码库和项目上下文，辅助开发人员进行编程工作。
企业系统：可在保障数据安全和隐私的情况下，将 AI 功能集成到企业系统中，助力企业实现智能化升级。

二.LangGraph 调用 Qianwen32B

本段代码使用 LangGraph 构键了一个简单的工作流，并用过 Langchain 的 ChatOpenAI 类访问硅基流动平台的 QianWen 32B 模型

1. 创建虚拟环境

## 查看已有环境
conda env list
## 新建
conda create -n mcp-env python=3.10
## 激活环境
conda activate mcp-env
## 升级 PIP
python -m pip install --upgrade pip
## 安装依赖
pip install langgraph langchain langchain_mcp_adapters
pip install requests pydantic
pip install langchain-openai
pip install flask flask-restful
pip install mcp

2.基于 LangGraph 实现一个简单聊天服务

from typing import List, Optional, Dict, Any, finalfrom langchain_core.messages import AIMessage
from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, MessagesPlaceholder, \HumanMessagePromptTemplate
from langchain_core.runnables import RunnableLambda
from langchain_openai import ChatOpenAI
from langgraph._internal._typing import TypedDictLikeV1, TypedDictLikeV2, DataclassLike
from langgraph.graph import StateGraph, END
from langgraph.graph.state import CompiledStateGraph
from pydantic import BaseModel, Field, SecretStr# 1. 聊天助手响应解析方法 原始结构 {'messages': [{'role': 'user', 'content': '你好'}, {'role': 'assistant', 'content': '\n\n你好呀！今天过得怎么样？有什么有趣的事情想分享吗？😊'}], 'user_input': '你好'}
def _extract_content(response) -> str:"""统一提取模型响应内容"""if isinstance(response, str):return responseelif isinstance(response, AIMessage):  # 处理AIMessage对象return response.contentelif isinstance(response, dict):  # 处理API原始响应# 优先查找assistant的最后一条消息if 'messages' in response:for msg in reversed(response['messages']):if msg.get('role') == 'assistant':return msg.get('content', '')return response.get('content', '')return str(response)class GeneralChatComponent:"""通用聊天组件"""def __init__(self, llm):self.llm = llmself.prompt = ChatPromptTemplate.from_messages([SystemMessagePromptTemplate.from_template("""你是一个友好的聊天助手，能够回答各种日常问题，进行轻松愉快的对话。请根据上下文和用户问题，给出自然、有帮助的回应。"""),MessagesPlaceholder(variable_name="history"),HumanMessagePromptTemplate.from_template("{current_input}")])self.chain = self.prompt | self.llm | RunnableLambda(_extract_content)def process(self, history, current_input):response = self.chain.invoke({"history": history,"current_input": current_input})return response# 2. 定义对话状态（使用LangGraph的消息处理机制）
class ChatState(BaseModel):"""对话状态管理"""messages: List[Dict] = Field(default_factory=list)  # 消息历史responses: str = ""user_input: Optional[str] = None  # 当前用户输入# 3. 定义LangGraph节点
def add_user_message(state: ChatState) -> ChatState:"""添加用户消息到对话历史"""if state.user_input:state.messages.append({"role": "user", "content": state.user_input})return statedef generate_response(state: ChatState, chat_component: GeneralChatComponent) -> ChatState:"""调用Qwen-32B生成回复"""# 准备对话历史history = state.messages[:-1]  # 排除当前用户输入current_input = state.messages[-1]["content"] if state.messages else ""# 调用模型response = chat_component.process(history, current_input)# 添加模型回复到状态state.messages.append({"role": "assistant", "content": response})# 写到响应结果return state.model_copy(update={"responses": response.lstrip('\n')})def should_continue(state: ChatState) -> str:"""判断对话是否继续"""if state.user_input and state.user_input.lower() in ["退出", "quit"]:return "end"return "generate"# 4. 构建对话流程图
def build_chat_graph(chat_component: GeneralChatComponent) -> CompiledStateGraph[TypedDictLikeV1 | TypedDictLikeV2 | DataclassLike | BaseModel | Any, Any, Any, Any]:"""构建LangGraph状态图"""graph_builder = StateGraph(ChatState)# 添加节点graph_builder.add_node("add_user_message", add_user_message)graph_builder.add_node("generate_response", lambda s: generate_response(s, chat_component))# 定义流程graph_builder.set_entry_point("add_user_message")graph_builder.add_conditional_edges("add_user_message",should_continue,{"generate": "generate_response","end": END})graph_builder.add_edge("generate_response", END)return graph_builder.compile()# 5. 运行对话系统
if __name__ == "__main__":# 初始化模型,通过openai类调用qw 演示代码内采用的是硅基流动平台的免费流量qwen_llm = ChatOpenAI(model='Qwen/QwQ-32B',api_key=SecretStr('sk-*******************'),base_url='https://api.siliconflow.cn/v1',temperature=0.7,max_tokens=2048)