LangChain+LangGraph+MCP智能体与工作流开发知识点全整理

一、大模型选择与私有化部署

1. 主流大模型特性与选型

（1）核心推荐模型

模型系列	关键型号	核心特性	适用场景
DeepSeek	DeepSeek-V3、DeepSeek-R1	支持思维链（reasoning_content）输出，API最大上下文64K，支持Function Calling/JSON Output	复杂逻辑推理、代码生成、多轮对话
Qwen3（通义千问）	Qwen3-235B-A22B（MoE）、Qwen3-30B-A3B、Qwen3-8B	覆盖0.6B-235B参数量级，支持MoE与Dense两种架构；思维/非思维模式切换；支持119种语言；Agent能力增强（MCP支持）	企业级应用、多模态处理、大参数量需求场景
其他主流模型	GPT-4o、Claude-3.7-Sonnet、Gemini-1.5	闭源模型，功能全面（多模态、长上下文），但需依赖第三方API，私有化成本高	无需私有化、追求极致效果的场景

（2）模型关键参数对比（以Qwen3为例）

• Qwen3-235B-A22B：总参数量2350亿，激活参数量220亿，编码/数学/通用能力对标GPT-o1、DeepSeek-R1
• Qwen3-30B-A3B：小型MoE模型，激活参数量30亿（为Qwen3-32B的10倍）
• 小型模型（Qwen3-4B/8B）：Qwen3-4B性能接近Qwen2.5-72B-Instruct，适合资源受限场景

2. 私有化部署方案

（1）部署工具选型：vLLM vs Ollama

对比维度	Ollama	vLLM
定位	轻量化本地部署（个人/原型验证）	高性能推理框架（企业级生产环境）
安全风险	默认无鉴权（11434端口暴露风险）	支持APIKey鉴权、SSL配置，安全性更高
并发能力	有限（单会话/少量并发）	支持Continuous Batching，高吞吐低延迟
硬件要求	支持CPU/GPU（最低16GB内存）	强制GPU（NVIDIA Tesla系列），显存要求高
模型格式支持	GGUF格式	HuggingFace格式（.bin/.safetensors）
生产环境推荐度	不推荐（安全/并发短板）	推荐（企业级特性完善）

（2）vLLM部署关键步骤

1. 模型下载：通过ModelScope下载（推荐Qwen3系列），代码示例：

   from modelscope import snapshot_download
   # 下载Qwen3-8B到指定目录
   model_dir = snapshot_download('Qwen/Qwen3-8B', cache_dir='/root/autodl-tmp/models', revision='master')
   

2. vLLM安装：pip install vllm
3. 核心部署命令（以Qwen3-8B为例）：

   python -m vllm.entrypoints.openai.api_server \
     --model /root/autodl-tmp/models/Qwen/Qwen3-8B \
     --served-model-name qwen3-8b \
     --max-model-len 8k \
     --host 0.0.0.0 \
     --port 6006 \
     --dtype bfloat16 \
     --gpu-memory-utilization 0.8 \
     --enable-auto-tool-choice \
     --tool-call-parser hermes \
     --enable-reasoning \
     --reasoning-parser deepseek_r1
   

4. API调用示例（OpenAI SDK兼容）：

   from openai import OpenAI
   client = OpenAI(api_key="EMPTY", base_url="http://localhost:6006/v1")
   response = client.chat.completions.create(
       model="qwen3-8b",
       messages=[{"role":"user", "content":"介绍大语言模型的核心原理"}],
       temperature=0.7,
       max_tokens=8192
   )
   print(response.choices[0].message.content)
   

（3）部署辅助工具

• 显存计算器：LlamaFactory显存估算工具，提前评估模型所需显存
• 镜像加速：HuggingFace镜像站（https://hf-mirror.com/），解决模型下载慢问题

二、新版LangChain应用开发

1. LangChain核心架构与组件

（1）三大核心包

包名	核心作用	依赖特点
langchain-core	定义基础抽象（LLM、向量存储、检索器接口），无第三方集成	轻量级，依赖极少
langchain	提供通用链、Agent、检索策略（非第三方集成）	核心逻辑层，不绑定具体工具
langchain-community	社区维护的第三方集成（LLM、向量数据库、工具）	依赖可选，按需引入

（2）核心功能模块

• 大模型连接：统一封装第三方LLM接口，支持标准化参数（model/temperature/max_tokens等），示例：

  from langchain_openai import ChatOpenAI
  # 连接私有化部署的Qwen3
  llm = ChatOpenAI(
      model='qwen3-8b',
      base_url="http://localhost:6006/v1",
      api_key="EMPTY",
      temperature=0.8
  )
  

• 提示词模板（Prompt Template）：分两类
  1. 字符串模板：适用于简单输入，示例：

     from langchain_core.prompts import PromptTemplate
     template = PromptTemplate.from_template("生成关于{topic}的300字介绍")
     prompt = template.invoke({"topic": "人工智能"})
     

  2. 聊天模板：适用于多轮对话，支持系统/用户/助手消息，示例：

     from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
     template = ChatPromptTemplate.from_messages([
         ("system", "你是专业的技术顾问"),
         MessagesPlaceholder("chat_history"),  # 历史消息占位符
         ("user", "{input}")
     ])
     

• 输出解析器（结构化输出）：三种核心方式
  1. with_structured_output()：基于Pydantic定义结构，示例：

     from pydantic import BaseModel, Field
     class Joke(BaseModel):
         setup: str = Field(description="笑话铺垫")
         punchline: str = Field(description="笑话笑点")
     # 绑定结构化输出
     runnable = llm.with_structured_output(Joke)
     result = runnable.invoke("生成关于程序员的笑话")
     

  2. SimpleJsonOutputParser：强制输出JSON格式，示例：

     from langchain_core.output_parsers import SimpleJsonOutputParser
     prompt = ChatPromptTemplate.from_template("输出JSON：{{'answer':'{question}的答案','followup':'后续问题'}}")
     chain = prompt | llm | SimpleJsonOutputParser()
     result = chain.invoke({"question":"地球自转周期"})
     

  3. 工具调用（bind_tools()）：通过工具定义结构化输出，示例：

     class WeatherResponse(BaseModel):
         city: str = Field(description="城市名")
         temp: float = Field(description="温度")
     # 绑定工具
     runnable = llm.bind_tools([WeatherResponse])
     response = runnable.invoke("查询北京今天的温度")
     print(response.tool_calls[-1]['args'])  # 提取结构化结果
     

2. RAG（检索增强生成）核心模块

（1）Embedding模型选型与部署

• 主流开源模型对比

  模型	研发机构	向量维度	语言支持	开源协议	核心优势
  BGE-Large-zh-v1.5	北京智源	1024	中英文（中文优）	MIT	中文语义理解强，适合中文RAG场景
  GTE-Large	阿里巴巴	1024	多语言	部分商用授权	多语言扩展性好，支持8k长上下文
  Qwen3-Embedding	阿里云	768/1024	119种语言	Apache 2.0	与Qwen3生态兼容，多模态支持

• 私有化部署示例（BGE-Large）：

  from langchain.embeddings import HuggingFaceBgeEmbeddings
  model = HuggingFaceBgeEmbeddings(
      model_name="BAAI/bge-large-zh-v1.5",
      model_kwargs={'device': 'cuda'},  #  GPU加速
      encode_kwargs={'normalize_embeddings': True},  # 归一化（便于余弦相似度计算）
      query_instruction="为这个句子生成表示以用于检索相关文章："  # 查询指令（提升检索精度）
  )
  

（2）向量数据库选型

数据库	核心特点	适用场景
Chroma	轻量级，支持DuckDB/ClickHouse存储，Python/JS SDK完善	快速原型、中小规模数据（百万级向量）
FAISS	Facebook开源，支持GPU加速，适合高维向量快速检索	大规模数据（亿级向量），需自行管理存储
Milvus	分布式架构，支持水平扩展，提供完整的向量搜索服务（索引、备份、监控）	企业级生产环境，高并发检索需求

（3）RAG完整流程

1. 数据预处理：文档加载（PDF/Word）→ 文本切片（按语义分割，如500token/片）→ 向量化（Embedding模型）
2. 向量存储：将向量写入数据库（如Chroma），示例：

   from langchain.vectorstores import Chroma
   # 向量化文档并写入Chroma
   vector_db = Chroma.from_texts(
       texts=document_chunks,  # 文本切片列表
       embedding=model,        # Embedding模型
       persist_directory="./chroma_db"  # 持久化目录
   )
   

3. 检索与生成：用户查询→向量化→相似度匹配（Top-K）→ 拼接上下文→LLM生成答案，示例：

   # 检索相似文档（Top3）
   retriever = vector_db.as_retriever(search_kwargs={"k": 3})
   # 构建RAG链
   from langchain.chains import RetrievalQA
   rag_chain = RetrievalQA.from_chain_type(
       llm=llm,
       chain_type="stuff",  # 上下文拼接方式
       retriever=retriever
   )
   # 执行查询
   result = rag_chain.run("介绍LangChain的核心组件")
   

三、基于LangGraph的Agent开发

1. Agent核心概念与定位

• 定义：具备“推理+工具调用”能力的智能体，通过LLM自主规划任务、调用工具、迭代执行，直至完成目标。
• 与Workflow的区别：

  类型	核心逻辑	灵活性
  Workflow	预定义代码路径，按固定流程执行（如“步骤1→步骤2→步骤3”）	低，需手动修改流程
  Agent	动态推理决策，根据实时上下文调整工具调用与流程（如“需查询天气→调用天气工具→根据结果生成建议”）	高，自主适应场景

• LangGraph中Agent的本质：Agent是StateGraph中的一个节点，通过create_react_agent快速创建，示例：

  from langgraph.prebuilt import create_react_agent
  # 定义工具
  def get_weather(city: str) -> str:
      return f"{city}今日天气晴朗，温度25℃"
  # 创建Agent
  agent = create_react_agent(llm=llm, tools=[get_weather], prompt="你是贴心的天气助手")
  # 执行Agent
  response = agent.invoke({"messages": [{"role": "user", "content": "查询上海天气"}]})
  

2. LangGraph本地服务搭建

（1）环境准备（Python≥3.11）

1. 创建虚拟环境：

   pip install virtualenv
   virtualenv langgraph-env
   # 激活环境（Windows）：langgraph-env\Scripts\activate
   # 激活环境（Linux/Mac）：source langgraph-env/bin/activate
   

2. 安装LangGraph CLI：pip install --upgrade "langgraph-cli[inmem]"

（2）服务搭建步骤

1. 创建LangGraph应用：

   langgraph new my-agent-app --template new-langgraph-project-python
   cd my-agent-app
   

2. 安装依赖：pip install -e .（编辑模式，支持本地代码修改）
3. 修改核心逻辑（graph.py）：

   from langchain_openai import ChatOpenAI
   from langgraph.prebuilt import create_react_agent
   # 连接私有化Qwen3
   llm = ChatOpenAI(
       model='qwen3-8b',
       base_url="http://localhost:6006/v1",
       api_key="EMPTY"
   )
   # 定义工具
   def get_weather(city: str) -> str:
       return f"{city}今日天气：晴朗，25℃"
   # 构建Agent图
   graph = create_react_agent(llm=llm, tools=[get_weather])
   

4. 启动服务：langgraph dev（默认端口2024）
5. 测试服务（Python SDK）：

   from langgraph_sdk import get_sync_client
   client = get_sync_client(url="http://localhost:2024")
   # 发送请求
   for chunk in client.runs.stream(
       None,  # 无线程ID（单次请求）
       "agent",  # 助手ID（langgraph.json配置）
       input={"messages": [{"role": "user", "content": "查询广州天气"}]}
   ):
       print(chunk.data)
   

3. Agent工具定义与上下文管理

（1）工具定义的三种方式

1. 函数装饰器（@tool）：最简单方式，示例：

   from langchain_core.tools import tool
   @tool
   def calculator(a: int, b: int) -> int:
       """两数相加的计算器工具"""
       return a + b
   

2. Runnable对象转换：通过as_tool()将LangChain Runnable转为工具，示例：

   from langchain_core.runnables import RunnableLambda
   # 定义Runnable
   weather_runnable = RunnableLambda(lambda city: f"{city}天气：25℃")
   # 转为工具
   weather_tool = weather_runnable.as_tool(name="get_weather", description="查询城市天气")
   

3. BaseTool子类化：最灵活，支持自定义参数校验与异常处理，示例：

   from langchain_core.tools import BaseTool
   from pydantic import BaseModel, Field
   # 定义工具参数模型
   class SearchArgs(BaseModel):
       query: str = Field(description="搜索关键词")
   # 自定义工具
   class SearchTool(BaseTool):
       name = "search_tool"
       description = "搜索互联网公开信息"
       args_schema = SearchArgs
       def _run(self, query: str) -> str:
           # 调用搜索API逻辑（示例）
           return f"搜索结果：{query}的相关信息..."
   

（2）上下文与记忆管理

• 短期记忆（会话级）：通过Checkpointer保存会话状态，支持多轮对话，示例：

  from langgraph.checkpoint.memory import InMemorySaver
  # 开发环境：内存存储
  checkpointer = InMemorySaver()
  # 生产环境：Postgres/Redis存储
  # from langgraph.checkpoint.postgres import PostgresSaver
  # checkpointer = PostgresSaver.from_conn_string("postgresql://user:pass@localhost:5432/db")
  
  # 创建带记忆的Agent
  agent = create_react_agent(llm=llm, tools=[get_weather], checkpointer=checkpointer)
  # 带线程ID执行（保持会话上下文）
  config = {"configurable": {"thread_id": "user-123"}}
  response1 = agent.invoke({"messages": [{"role": "user", "content": "查询北京天气"}], config=config)
  response2 = agent.invoke({"messages": [{"role": "user", "content": "那上海呢"}], config=config)  # 保留历史对话
  

• 长期记忆（跨会话）：通过Store存储用户偏好/历史数据，示例：

  from langgraph.store.memory import InMemoryStore
  # 开发环境：内存存储
  store = InMemoryStore()
  # 生产环境：Redis/Postgres存储
  # from langgraph.store.redis import RedisStore
  # store = RedisStore.from_conn_string("redis://:pass@localhost:6379")
  
  # 存储用户偏好
  store.put("user-123", "preference", "喜欢简洁的天气播报")
  # 读取用户偏好
  preference = store.get("user-123", "preference")
  

四、MCP（Model Context Protocol）开发

1. MCP核心概念与价值

• 定义：2024年由Anthropic推出的开放协议，统一LLM与外部数据源/工具的通信标准，解决“数据孤岛”问题，支持本地/远程数据交互。
• 与Function Calling的核心区别：

  维度	MCP（协议）	Function Calling（功能）
  性质	通用通信标准	模型调用工具的机制
  适用范围	多数据源/工具/系统（LLM、应用均可对接）	单一模型与工具的交互
  复用性	一次开发，多场景复用	需为每个任务单独开发函数
  灵活性	支持动态扩展与适配	功能扩展需额外开发

2. MCP工作原理与通信机制

（1）核心架构（客户端-服务器-资源）

1. 客户端：发起请求（如LLM/Agent），维护与服务器的1:1连接
2. 服务器：转发请求至资源，控制访问权限，返回处理结果
3. 资源：实际提供服务的组件（数据库、工具、API等）

（2）三种通信机制

机制	核心特点	适用场景
stdio（本地）	基于标准输入输出，适用于本地工具交互	本地部署的轻量工具（如本地文件读取）
SSE（远程）	基于HTTP的单向流，服务器实时推送数据	需实时通知的场景（如任务进度更新）
Streamable-HTTP（推荐）	统一HTTP端点（支持POST/GET），简化通信模型，支持无状态/有状态模式	企业级生产环境，高可扩展性需求

3. FastMCP（Python MCP服务器实现）

（1）环境安装

pip install langchain-mcp-adapters  # LangChain与MCP适配
pip install fastmcp                 # FastMCP核心库

（2）核心功能与认证

• 认证机制：基于JWT（Bearer Token），流程如下：
  1. 生成RSA密钥对：

     from fastmcp.auth import RSAKeyPair
     key_pair = RSAKeyPair.generate()  # 生成公钥/私钥
     

  2. 配置认证提供方：

     from fastmcp.auth import BearerAuthProvider
     auth = BearerAuthProvider(
         public_key=key_pair.public_key,
         issuer="http://localhost:8000",  # 签发方
         audience="mcp-server"           # 目标服务标识
     )
     

  3. 生成Token（服务器端）：

     token = key_pair.create_token(
         subject="user-123",  # 用户ID
         scopes=["tool:read", "tool:write"],  # 权限范围
         expires_in_seconds=3600  # 1小时过期
     )
     

  4. 客户端携带Token请求：

     import requests
     headers = {"Authorization": f"Bearer {token}"}
     response = requests.post("http://localhost:8000/mcp/request", headers=headers, json={"query": "查询天气"})
     

五、基于LangGraph的WorkFlow开发

1. StateGraph核心概念（图结构）

LangGraph的WorkFlow基于有向图（Directed Graph） 实现，核心三要素：

（1）State（状态）

• 图的共享数据结构，通常用TypedDict或Pydantic BaseModel定义，支持reducer函数（控制状态更新方式），示例：

  from typing_extensions import TypedDict
  from langgraph.graph.message import add_messages
  # 定义状态（含消息历史与额外字段）
  class State(TypedDict):
      messages: Annotated[list, add_messages]  # 消息历史（add_messages：追加而非覆盖）
      documents: list[str]                   # 额外字段（文档列表）
  

• Reducer函数：控制状态更新逻辑，常用add_messages（消息追加）、operator.add（列表拼接）。

（2）Node（节点）

• 执行具体逻辑的Python函数（同步/异步），输入为当前State，输出为更新后的State，示例：

  # 定义“生成笑话”节点
  def generate_joke(state: State) -> dict:
      prompt = f"生成关于{state['topic']}的笑话"
      response = llm.invoke(prompt)
      return {"messages": [{"role": "assistant", "content": response.content}]}
  # 添加节点到图
  from langgraph.graph import StateGraph
  builder = StateGraph(State)
  builder.add_node("generate", generate_joke)  # 节点名："generate"，节点逻辑：generate_joke
  

（3）Edge（边）

• 定义节点间的路由关系，分两类：
  1. 普通边：固定路由（从节点A到节点B），示例：

     from langgraph.graph import START, END
     builder.add_edge(START, "generate")  # 从START到"generate"节点
     

  2. 条件边：根据State动态路由，示例：

     # 路由函数：根据笑话是否幽默决定下一步
     def route_joke(state: State) -> str:
         last_msg = state["messages"][-1]["content"]
         return "approve" if "幽默" in last_msg else "reject"
     # 添加条件边（从"evaluate"节点根据路由函数跳转）
     builder.add_conditional_edges(
         "evaluate",  # 源节点
         route_joke,  # 路由函数
         {"approve": END, "reject": "generate"}  # 路由映射（批准→结束，拒绝→重新生成）
     )
     

2. 典型WorkFlow案例：笑话生成与评估流程

（1）流程设计

START → generate（生成笑话） → evaluate（评估） → 条件路由（approve→END / reject→generate）

（2）完整代码

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain_core.pydantic_v1 import BaseModel, Field

# 1. 定义状态
class State(TypedDict):
    topic: str                # 笑话主题
    messages: list            # 消息历史
    is_funny: bool            # 是否幽默（评估结果）

# 2. 定义评估结构化输出模型
class Evaluation(BaseModel):
    is_funny: bool = Field(description="笑话是否幽默")
    feedback: str = Field(description="改进建议")

# 3. 定义节点
def generate_joke(state: State) -> dict:
    """生成笑话节点"""
    prompt = f"生成关于{state['topic']}的短笑话"
    response = llm.invoke(prompt)
    return {
        "messages": [{"role": "assistant", "content": response.content}],
        "topic": state["topic"]
    }

def evaluate_joke(state: State) -> dict:
    """评估笑话节点"""
    last_joke = state["messages"][-1]["content"]
    # 结构化评估
    evaluator = llm.with_structured_output(Evaluation)
    result = evaluator.invoke(f"评估笑话：{last_joke}")
    return {
        "is_funny": result.is_funny,
        "messages": [{"role": "system", "content": f"评估结果：{result.feedback}"}]
    }

# 4. 定义路由函数
def route_evaluation(state: State) -> str:
    return END if state["is_funny"] else "generate"

# 5. 构建并编译图
builder = StateGraph(State)
builder.add_node("generate", generate_joke)
builder.add_node("evaluate", evaluate_joke)
# 连接边
builder.add_edge(START, "generate")
builder.add_edge("generate", "evaluate")
builder.add_conditional_edges("evaluate", route_evaluation)
# 编译图
workflow = builder.compile()

# 6. 执行流程
result = workflow.invoke({"topic": "程序员", "messages": [], "is_funny": False})
print("最终笑话：", result["messages"][0]["content"])

六、实战项目方向

1. 携程AI智能助手：集成多模态交互（语音/文本/图片）、行程规划工具、实时票务查询（对接12306 MCP服务）
2. RAG企业知识库：基于BGE-Large+Milvus，实现企业文档（PDF/Word）的语义检索与问答
3. TEXT2SQL+Qwen3：通过Qwen3的代码生成能力，将自然语言转为SQL，对接业务数据库实现数据查询
4. 多智能体监督系统：基于langgraph-supervisor，构建中心监督Agent，协调多个专业Agent（天气查询、日程管理、文档分析）