一、Overview
1.1 一句话理解
- LangChain = 大模型应用的"乐高积木"，提供模型、提示词、记忆等组件，拼装出 AI 应用
- LangGraph = “流程图编排器”，把复杂任务画成节点和边的图，控制每一步怎么走
LangChain 1.0 之后，LangGraph 成为整个生态的底座。两者关系：
| 维度 | LangChain | LangGraph |
|---|---|---|
| 定位 | LLM 应用开发框架 | 有状态工作流编排框架 |
| 核心抽象 | Chain（链）、组件 | Graph（图）、节点、边、状态 |
| 适用场景 | 简单线性流程 | 复杂带循环、分支、并行的流程 |
1.2 智能体（Agent）
智能体 = 能感知环境、自己做决定、调工具干活的"数字员工"。以 LLM 为大脑，整合感知、规划、记忆、工具调用完成复杂任务。
核心架构（类比人体）：
- 大脑（LLM）：决策中枢，负责拆任务、做推理
- 感官（感知）：处理文本/图片/语音等输入
- 手脚（行动）：调搜索引擎、API 等外部工具
- 记忆：短期（对话上下文）+ 长期（向量数据库）
常用智能体框架对比：
| 名称 | 是否开源 | 核心能力 |
|---|---|---|
| LangChain/LangGraph | 开源 | 模块化，精细控制每步状态，适合复杂业务 |
| CrewAI | 开源 | 基于角色分工的多 Agent 协作 |
| AutoGen | 开源 | 多 Agent 对话，内置人类反馈 |
| Dify | 开源 | 低代码可视化编排 |
| 扣子（Coze） | 商用 | 拖拽式工作流，60+ 插件 |
本文选 LangGraph：生态成熟、配合 LangSmith 调试、能精细控制每一步。
二、LangChain 六大组件
LangChain 的核心思想是模块化：把 LLM 和外部资源封装成标准接口的组件。
2.1 Models 模型
对接各种大模型的统一接口，分两类：LLM（文本生成）和 Embedding（向量化）。
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
import os
llm = ChatOpenAI(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o-mini")
embeddings = OpenAIEmbeddings(model="text-embedding-3-small", api_key=os.getenv("OPENAI_API_KEY"))
2.2 Prompts 提示词
模板化用户输入，把变量动态注入提示。
from langchain_core.prompts import PromptTemplate
prompt = PromptTemplate(
input_variables=["context", "question"],
template="基于以下已知信息回答问题：\n已知内容：{context}\n问题：{question}"
)
final_prompt = prompt.format(context="北京是首都", question="首都是哪？")
2.3 Chains 链
把多个组件按顺序串起来。LangChain 1.0 后推荐用 LCEL 管道符 | 组装。
rag_chain = (
RunnablePassthrough.assign(context=lambda x: "检索到的上下文")
| prompt | llm | StrOutputParser()
)
2.4 Agents 智能体
让 LLM 自主选择调哪个工具。LangGraph 提供 create_react_agent 快速创建。
from langgraph.prebuilt import create_react_agent
def get_weather(city: str) -> str:
"""查询指定城市的天气"""
return f"{city}今天晴"
agent = create_react_agent(model=llm, tools=[get_weather], prompt="你是一个天气助手")
2.5 Memory 记忆
多轮对话保持上下文。LangGraph 用 Checkpointer 机制，配合 thread_id 区分会话。
from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
agent = create_react_agent(model=llm, tools=[], checkpointer=checkpointer)
config = {"configurable": {"thread_id": "user-001"}} # 同一 thread_id 保留上下文
2.6 Indexes 索引
和外部数据交互的桥梁，含四个子模块：文档加载、文本分割、向量库、检索器。
from langchain_community.document_loaders import PyMuPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
# 1. 加载 + 分割
docs = PyMuPDFLoader("data/文档.pdf").load()
texts = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=20).split_documents(docs)
# 2. 存入向量库
vectordb = Chroma.from_documents(texts, embeddings, persist_directory="./chroma")
# 3. 检索
retriever = vectordb.as_retriever(search_kwargs={'k': 2})
docs = retriever.invoke("查询内容")
支持的向量库：Chroma、ElasticSearch、Milvus、Redis、FAISS、Pinecone 等。
2.7 结构化输出
把 LLM 的自由文本强制转成预定义结构（Pydantic 模型），适合信息抽取。
from pydantic import BaseModel, Field
class PersonInfo(BaseModel):
name: str = Field(description="姓名")
age: int = Field(description="年龄")
agent = create_agent(model=llm, response_format=PersonInfo)
response = agent.invoke({"messages": [{"role": "user", "content": "我叫李明，28岁"}]})
print(response["structured_response"])
三、RAG 检索增强生成
3.1 一句话理解
RAG = 让大模型"开卷考试"。不改模型权重，从知识库检索相关文档喂给模型，解决知识滞后和缺乏私有知识的问题。比微调便宜、更新快。
3.2 三阶段流程
[索引阶段·离线] [检索阶段·在线] [生成阶段·在线]
加载文件 → 提取内容 → 分块 → query 向量化 → 上下文+问题组装
→ 向量化 → 存向量库 匹配 top_k 文档 提交 LLM 生成答案
3.3 最小示例
# 1. 加载 + 分割 + 入库
docs = TextLoader("data/知识.txt").load()
texts = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=20).split_documents(docs)
vectordb = Chroma.from_documents(texts, embeddings, persist_directory="./chroma")
retriever = vectordb.as_retriever(search_kwargs={'k': 2})
# 2. 检索 + 生成
query = "你的问题"
context = "\n".join([d.page_content for d in retriever.invoke(query)])
prompt = PromptTemplate(
input_variables=["context", "question"],
template="基于以下已知信息回答问题，不要编造：\n已知内容：{context}\n问题：{question}"
)
result = llm.invoke(prompt.format(context=context, question=query))
print(result.content)
RAG 的核心优化点：知识库构建质量和检索策略（query 改写、混合检索、重排序）。
四、LangGraph 核心概念
LangGraph 的设计就一句话：用图来组织工作流。节点 = 干什么，边 = 下一步去哪，状态 = 节点间传的纸条。
4.1 四要素
| 概念 | 作用 | 类比 |
|---|---|---|
| 图（Graph） | 整个工作流，有向图，含 START/END | 一张流程图 |
| 状态（State） | 节点间传递的数据，本质是字典 | 传纸条 |
| 节点（Node） | 一个操作单元，封装成函数 | 工位 |
| 边（Edge） | 控制执行顺序 | 传送带 |
from typing_extensions import TypedDict
class State(TypedDict):
question: str
answer: str
def think_node(state: State):
state["answer"] = "思考结果"
return state
边分两种：
- 普通边：
add_edge(src, dst)固定顺序 - 条件边：
add_conditional_edges(src, fn, mapping)根据返回值选分支
4.2 状态图三步法（万能模板）
所有 LangGraph 工作流都遵循三步：建状态 → 建节点 → 串边。
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
# 1️⃣ 创建状态
class MyState(TypedDict):
question: str
answer: str
# 2️⃣ 创建节点
def think_node(state: MyState):
state["answer"] = "思考结果"
return state
# 3️⃣ 串边
def build_graph():
gb = StateGraph(MyState)
gb.add_node("think", think_node)
gb.add_edge(START, "think")
gb.add_edge("think", END)
return gb.compile()
graph = build_graph()
result = graph.invoke({"question": "你好"})
后续所有高级特性都建立在这三步之上，只是在不同环节加料。
五、流程控制：分支 / 循环 / 并行
5.1 分支（条件边）
条件节点返回字符串，条件边根据返回值映射到后续节点。
def evaluate(state):
return "通过" if state["score"] >= 60 else "重试"
gb.add_conditional_edges("exam", evaluate, {"通过": "next", "重试": "study"})
5.2 循环
条件边指回自身，构造循环。常用于"未达标则重复"。
def check(state):
return "完成" if state["temp"] >= 100 else "继续加热"
gb.add_conditional_edges("heat", check, {"完成": END, "继续加热": "heat"})
5.3 并行（fan-out / fan-in）
- 并行：一个节点向多个节点连边
- 汇总：多个节点向一个节点连边，用 list 传多个源，LangGraph 会等全部完成
gb.add_edge("start", "task_a") # fan-out
gb.add_edge("start", "task_b")
gb.add_edge(["task_a", "task_b"], "merge") # fan-in
5.4 并行状态合并
并行节点写同一个字段默认会覆盖。用 Annotated 指定 reducer 自动合并。
from typing import Annotated
import operator
class State(TypedDict):
menu: Annotated[list, operator.add] # 多节点同时写 menu 时自动拼接而非覆盖
| 写法 | 效果 |
|---|---|
menu: list |
并行写会覆盖（默认） |
menu: Annotated[list, operator.add] |
自动拼接 |
5.5 人工介入（Human-in-the-loop）
关键步骤需人工确认。三要素：interrupt + memory + Command 恢复。
from langgraph.types import interrupt, Command
def human_approval(state):
approval = interrupt("请输入审批结果：") # 触发中断，图暂停
state["approval"] = approval
return state
memory = InMemorySaver()
graph = gb.compile(checkpointer=memory) # ⚠️ 必须挂载 memory
config = {"configurable": {"thread_id": "1"}}
graph.invoke({"input": "..."}, config=config) # 执行到 interrupt 暂停
result = graph.invoke(Command(resume="批准"), config=config) # 人工输入后恢复
⚠️ 人工介入必须有 checkpointer，否则暂停后状态丢失无法恢复。
5.6 子图（Subgraph）
把一组节点封装成可复用子流程，主图把它当一个节点用。
subgraph = build_subgraph().compile()
gb.add_node("sub", subgraph) # 直接加入，主子图隐式共享同名字段
六、高级特性
6.1 RunnableConfig 配置注入
节点需要"运行时参数"（用户名、API key 等）时，通过 RunnableConfig 注入，避免污染 state。
from langgraph.types import RunnableConfig
def step1(state, config: RunnableConfig): # 第二个参数固定为 RunnableConfig
prefix = config.get("configurable", {}).get("prefix", "")
state["result"] = f"{prefix}{state['input']}"
config = RunnableConfig(configurable={"prefix": "[前缀] ", "user_id": "u001"})
graph.invoke({"input": "你好"}, config=config)
典型应用：多租户传 tenant_id、多语言传 language、调试传 trace_id。
6.2 thread_id 线程隔离
thread_id 区分不同会话。同一 thread_id = 同一对话，不同 thread_id 互不干扰。
6.3 InMemorySaver 记忆
默认 invoke 每次从零开始。InMemorySaver 把每次调用后的 state 按 thread_id 保存，同 id 再次调用自动接续。
memory = InMemorySaver()
graph = gb.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "1"}}
graph.invoke({"new_value": 5}, config=config) # total = 5
graph.invoke({"new_value": 7}, config=config) # total = 5+7 = 12（自动接续）
config2 = {"configurable": {"thread_id": "2"}}
graph.invoke({"new_value": 10}, config=config2) # total = 10（全新开始）
⚠️ 内存存储重启即丢失，生产环境换
SqliteSaver/PostgresSaver。 memory 决定"在哪存"，thread_id 决定"存哪份"，两者缺一不可。
6.4 四种调用方式
| 方法 | 同步/异步 | 是否流式 | 适用场景 |
|---|---|---|---|
graph.invoke(input) |
同步 | 否 | 命令行脚本、简单测试 |
graph.stream(input) |
同步 | 是 | 本地调试、看每步输出 |
graph.ainvoke(input) |
异步 | 否 | Web 服务单次返回 |
graph.astream(input) |
异步 | 是 | Web 服务流式响应（推荐） |
stream 让每个节点完成时立刻吐出结果，提升交互体验：
for chunk in graph.stream({"question": "为什么天空是蓝色的？"}):
print(chunk) # {"思考节点": {...}} {"回答节点": {...}}
七、智能体设计模式
7.1 两种思路
- 工作流（workflow）：预先编排好代码路径，掌控力强但需量身定制
- 智能体（agent）：LLM 动态决定调什么工具，灵活但可控性差
总原则：如无必要，勿增实体。简单任务直接用 LLM，复杂智能体才上框架。
Anthropic 在《Building Effective Agents》提出六种构建块，前五种是工作流，最后一种是真正的智能体。
7.2 五种工作流模式
① 提示链：复杂任务拆成多步逐步完成
START → 生成大纲 → 撰写初稿 → 润色 → END
② 分支：按输入类别路由到不同处理
START → 分类 → ┬→ 儒家专家 → END
├→ 法家专家 → END
└→ 道家专家 → END
③ 并行化：互不干扰的任务同时做
START → ┬→ 查发货日期 → ┐
└→ 查运输时间 → ┴→ 发送通知 → END
④ 计划-执行：编排器规划任务列表，worker 并行执行后合并
START → 生成梗概 → 规划章节 → ┬→ 写第1章 → ┐
├→ 写第2章 → ┤→ 合并 → END
└→ ... → ┘
⑤ 生成-评估：生成后评估，不合格带反馈重写，循环优化
START → 生成 → 评估 → ┬→ 合格 → END
└→ 不合格 → 生成（带反馈）
7.3 智能体（Agent）
工作流是定制的，不通用。真正的智能体 = 大模型 + 工具集，由 LLM 自主决策调哪些工具。
from langgraph.prebuilt import create_react_agent
def search_train(date, start, end):
"""查询列车班次"""
return "D81 次 12:24 发车"
def search_hotel(date, city):
"""查询可用旅店"""
return "丽晶酒店 1200元"
agent = create_react_agent(model=llm, tools=[search_train, search_hotel])
result = agent.invoke({"messages": [{"role": "user", "content": "帮我安排6月15日北京到青岛的车次和住宿"}]})
print(result["messages"][-1].content)
八、LangSmith 监控
LangSmith = LLM 应用的"监控仪表盘"，用于开发、调试、测试、监控。
核心能力：追踪每次调用链路、调试问题、回归测试、线上监控。
import os
from langsmith import traceable
os.environ['LANGSMITH_TRACING'] = 'true'
os.environ['LANGSMITH_ENDPOINT'] = 'https://api.smith.langchain.com'
os.environ['LANGSMITH_API_KEY'] = '你的API密钥'
os.environ['LANGSMITH_PROJECT'] = 'my-project'
@traceable # 用 @traceable 装饰需要追踪的函数
def call_agent():
return agent.invoke({"messages": [{"role": "user", "content": "你好"}]})
注册地址：https://smith.langchain.com/
九、核心 API 速查
9.1 LangGraph 核心 API
| API | 作用 |
|---|---|
StateGraph(State) |
创建图构造器 |
gb.add_node(name, fn) |
注册节点 |
gb.add_edge(src, dst) |
加普通边 |
gb.add_edge([a, b], c) |
多源汇总（fan-in） |
gb.add_conditional_edges(src, fn, mapping) |
加条件边 |
gb.compile(checkpointer=...) |
编译图，挂载记忆器 |
graph.invoke / stream / ainvoke / astream |
同步/异步 + 普通/流式执行 |
interrupt(msg) |
触发人工中断 |
Command(resume=...) |
恢复中断执行 |
9.2 关键导入
from langgraph.graph import StateGraph, START, END
from langgraph.types import RunnableConfig, interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import create_react_agent
from typing import TypedDict, Annotated
十、学习路径与要点
1. LangChain 六大组件 → 理解模块化设计
2. RAG 流程 → 掌握知识库问答
3. LangGraph 三步法 → 学会图、状态、节点、边
4. 条件边 / 循环 → 学会分支和循环控制
5. 并行 + Annotated → 解决并行写冲突
6. memory + thread_id → 多轮记忆和会话隔离
7. 人工介入 / 子图 → 中断恢复、复用子流程
8. stream / astream → 流式输出
9. 六种设计模式 → 掌握智能体构建方法
核心要点：
- memory + thread_id 是核心：多轮对话、人工介入都依赖它
- 生产环境三件套：
astream+ 持久化 Checkpointer +thread_id - 如无必要勿增实体：简单任务不用框架，复杂任务再上 LangGraph
- 从基础功能开始：不要一次开发得过于复杂，逐步增加复杂度