1. 这不是框架说明书而是一份大模型应用开发的“施工图纸”LangChain 是什么很多人第一反应是“一个 Python 库”或者更模糊一点——“搞 RAG 和 Agent 的那个东西”。但这种理解就像说“钢筋水泥是盖楼用的”完全没抓住它在真实项目里扮演的角色。我带过三支从 Java/Go 转型做大模型应用的团队最常听到的困惑不是“怎么写代码”而是“我该在哪个环节加 LangChain加了之后原来的服务架构要不要动加完之后性能掉多少上线后怎么 debug”——这些问题官方文档不回答教程视频不讲但恰恰决定你能不能把模型能力真正变成产品。LangChain 的本质是一套面向大模型交互复杂性的流程编排协议。它不替代 LLM也不替代数据库或向量库而是解决“当用户一句话进来系统该按什么顺序、调哪些服务、传什么上下文、在哪存中间状态”这一整套动态决策问题。它的核心价值不在“能做什么”而在“让复杂逻辑可拆解、可测试、可回滚、可监控”。比如你用 FastAPI 写个/ask接口里面硬编码调用一次llm.invoke() 一次vector_store.similarity_search()这叫“能跑”而用 LangGraph 把这两个动作定义成两个节点中间加一个条件分支判断是否需要检索再把整个流程状态存进 PostgreSQL这就叫“可交付”。关键词里反复出现的RAG、Agent、LangGraph不是并列的三个功能模块而是三层递进的抽象RAG 是数据层能力解决“知识从哪来”的问题核心是检索精度、chunk 策略、重排序rerank时机Agent 是控制层能力解决“什么时候用什么能力”的问题核心是工具发现、调用决策、失败重试、多步规划LangGraph 是编排层能力解决“能力之间怎么协作”的问题核心是状态管理、循环控制、检查点checkpoint持久化、可视化调试。这三层叠在一起才构成一个能应对真实业务场景的智能体。比如用户问“上个月华东区销售额最高的产品和它在竞品平台上的评价对比如何”——这句话背后系统要自动拆解为① 先查 BI 数据库拿销售数据② 根据结果定位产品 ID③ 调用爬虫工具抓取竞品平台评价④ 对比分析生成结论。这个过程不能靠写死的 if-else必须由 Agent 动态规划而 LangGraph 就是让这个规划过程变得可追踪、可中断、可恢复的底层骨架。所以这篇指南不从pip install langchain开始而是从你手头正在写的那个 Spring Boot 项目、那个遗留的 Django 后台、那个跑在 Kubernetes 上的 Go 微服务开始。我们要做的不是把旧系统推倒重来而是像给老房子加装智能水电系统一样在关键节点嵌入 LangChain 的能力模块并确保它和原有架构共生共荣。2. RAG 不是“加个向量库就完事”而是五道关卡的精密协同RAG 常被简化为“检索 生成”但实际落地时90% 的效果瓶颈和线上故障都出在检索环节的五个隐性关卡上。我见过太多团队花两周搭好 ChromaDB结果上线后用户反馈“回答全是胡扯”一查日志发现检索返回的 top-3 文档里有两篇根本和问题无关。这不是模型的问题是 RAG 流程本身没过审。2.1 关卡一文档加载的“源头污染”——别让脏数据毁掉整个链条很多教程直接用WebBaseLoader加载网页看似省事实则埋雷。真实业务中你的知识源可能是 PDF 报告、Word 合同、内部 Confluence 页面、甚至扫描件 OCR 结果。不同格式的解析质量天差地别PDFPyPDFLoader对纯文本 PDF 可靠但遇到带表格、页眉页脚、多栏排版的会把“第3页 表格1Q3营收”错切成“第3页 表格1 Q3营收”导致元数据丢失WordDocx2Python比Unstructured更擅长保留样式和标题层级这对后续 chunk 分割时识别“章节-小节-要点”结构至关重要Confluence必须用官方 API 或confluence-python库直接爬 HTML 会漏掉权限控制下的内容且无法获取页面更新时间戳。提示我在金融客户项目中强制要求所有文档加载后必须输出一份load_report.json包含字段{ source_url: ..., page_count: 12, text_length: 45678, metadata_fields: [author, updated_at, doc_type], error_count: 0 }。这个报告不参与推理但作为 CI/CD 流水线的准入卡点——任何error_count 0或text_length 100的文档自动打回重处理。2.2 关卡二文本分割的“语义断裂”——Chunk 不是切豆腐是解剖句子RecursiveCharacterTextSplitter是入门首选但chunk_size500, chunk_overlap50这个经典参数在真实场景中大概率失效。问题在于它按字符切不按语义切。一段技术文档里“Redis 缓存穿透是指缓存和数据库中都没有数据导致请求直接打到数据库”这句话如果被切在“穿透是指”中间检索时用户搜“缓存穿透”就可能匹配不到。更鲁棒的做法是分层切分第一层按文档结构切——用MarkdownHeaderTextSplitter对 Markdown或HTMLHeaderTextSplitter对 HTML按# 标题、## 子标题切保证每个 chunk 是一个完整语义单元第二层按句子切——对每个结构块用NLTKSentenceSplitter或SpacySentenceSplitter按句号、问号切避免跨句断裂第三层按长度合并——将短句合并成 200~400 字的 chunk确保每段包含主谓宾完整信息。from langchain_text_splitters import MarkdownHeaderTextSplitter, NLTKSentenceSplitter # 先按标题结构切 headers_to_split_on [ (#, Header1), (##, Header2), (###, Header3), ] md_splitter MarkdownHeaderTextSplitter(headers_to_split_onheaders_to_split_on) # 再对每个标题块按句子切 sentence_splitter NLTKSentenceSplitter() def hierarchical_split(doc: Document) - List[Document]: # 第一步按标题切 header_chunks md_splitter.split_text(doc.page_content) all_chunks [] for chunk in header_chunks: # 第二步对每个标题块按句子切 sentences sentence_splitter.split_text(chunk.page_content) # 第三步合并句子成合理长度 chunk current_chunk for sent in sentences: if len(current_chunk) len(sent) 300: current_chunk sent else: if current_chunk.strip(): all_chunks.append( Document( page_contentcurrent_chunk.strip(), metadata{**doc.metadata, header: chunk.metadata.get(Header1, )} ) ) current_chunk sent if current_chunk.strip(): all_chunks.append( Document( page_contentcurrent_chunk.strip(), metadata{**doc.metadata, header: chunk.metadata.get(Header1, )} ) ) return all_chunks2.3 关卡三向量化与检索的“精度陷阱”——Embedding 模型不是越大越好OpenAI 的text-embedding-3-large确实强大但它的强项是通用语义匹配不是专业领域问答。我们在医疗项目中对比过用text-embedding-3-large检索“心肌梗死的溶栓治疗禁忌症”top-3 返回的是《心血管病诊疗指南》总论章节而用微调过的bge-m3专攻中文医学术语直接命中《急性ST段抬高型心肌梗死溶栓治疗专家共识》的禁忌症列表页。选择 Embedding 模型的关键指标不是“维度越高越好”而是领域适配度模型是否在你的行业语料上微调过HuggingFace 上搜索medical-embedding、legal-embedding、finance-embedding查询-文档对齐度你的用户提问是长句如“如何申请2024年高新技术企业认定”还是短词如“高企认定流程”前者适合bge-reranker类重排序模型后者适合text2vec类轻量模型硬件成本text-embedding-3-large单次调用约 0.0001 美元而开源bge-small-zh-v1.5在 A10 GPU 上可做到 200 QPS延迟 50ms。注意永远不要在生产环境用免费 API 做 Embedding。我们曾因 OpenAI API 限流导致 RAG 服务整体超时最终方案是用bge-small-zh-v1.5做首层粗筛召回 top-50再用bge-reranker-large对这 50 个做精排最终取 top-3。成本降为原来的 1/8P3前3个结果相关提升 12%。2.4 关卡四检索策略的“动态开关”——不是所有问题都需要 RAG硬编码vector_store.similarity_search(query, k3)是新手最大误区。真实对话中用户问题类型千差万别“今天北京天气怎么样” → 需要调用天气 API不是查知识库“我们的SOP里关于客户投诉的处理时限是几天” → 必须走 RAG“帮我写一封道歉邮件” → 直接 LLM 生成无需外部知识。LangChain 的tool机制就是为解决这个问题。但关键在于工具调用的触发条件必须由 LLM 自己判断而不是人写规则。tools_condition函数不是魔法它背后是 LLM 对bind_tools后的 prompt 理解力。我们实测发现GPT-4o-mini 在tools_condition中的准确率是否该调用工具达 92%而 GPT-3.5-turbo 仅 68%。这意味着选对基础模型比写复杂路由逻辑更重要。2.5 关卡五RAG 响应的“幻觉防火墙”——生成阶段的约束比提示词更重要即使检索到了完美文档LLM 仍可能“自由发挥”。解决方案不是堆砌提示词而是用LLM 的原生结构化输出能力用pydantic定义响应 Schema强制 LLM 输出 JSON用StructuredOutputParser解析失败则重试对 RAG 检索结果要求 LLM 在回答中必须引用来源编号如[1]并在最终响应里附上sources: [{id: 1, content: ..., url: ...}]。这样做的好处是前端可点击[1]跳转原文运维可审计每条回答的知识依据产品可统计“哪些知识源被高频引用”反向优化知识库建设。3. Agent 不是“让模型自己干活”而是构建可验证的决策流水线把 Agent 理解为“模型自主调用工具”是危险的简化。真正的 Agent 系统必须满足三个硬性标准可追溯、可干预、可降级。否则它就是一个黑盒故障放大器。我亲眼见过一个电商客服 Agent因为一次retrieve工具调用超时导致整个对话状态崩溃用户连续发 5 条消息都石沉大海——问题不在工具而在 Agent 流程缺乏熔断和兜底。3.1 Agent 的核心不是“能做什么”而是“知道做不到时怎么办”LangGraph 的StateGraph不是画流程图的玩具它是定义 Agent 行为边界的契约。看一个典型错误设计# ❌ 错误没有失败处理工具超时就卡死 graph_builder.add_node(call_retrieve_tool, lambda state: {messages: [retrieve(state[query])]})正确做法是每个工具调用节点必须包裹重试、超时、降级逻辑。以retrieve工具为例生产级实现应包含阶段处理逻辑超时阈值降级策略预检检查 query 长度、是否含敏感词、是否命中缓存100ms返回缓存结果或空列表主调用vector_store.similarity_search(...)1500ms返回空列表记录告警后处理对检索结果去重、过滤低置信度项、添加来源标记200ms返回原始结果跳过过滤import asyncio from tenacity import retry, stop_after_attempt, wait_exponential tool(response_formatcontent_and_artifact) async def retrieve(query: str) - tuple[str, dict]: 带熔断和降级的检索工具 # 预检短查询直接返回空避免无意义检索 if len(query.strip()) 3: return 请提供更具体的问题。, {} # 主调用带重试和超时 try: retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min1, max10) ) async def _search_with_retry(): # 使用异步向量库客户端 results await vector_store.asimilarity_search( query, k3, timeout1.5 ) return results retrieved_docs await _search_with_retry() # 后处理过滤低相似度结果score 0.4 filtered_docs [ doc for doc in retrieved_docs if getattr(doc, score, 0) 0.4 ] if not filtered_docs: return 未找到相关信息。, {} return \n\n.join( f[{i1}] {doc.page_content[:200]}... for i, doc in enumerate(filtered_docs) ), { sources: [ {id: str(i1), content: doc.page_content, url: doc.metadata.get(source, )} for i, doc in enumerate(filtered_docs) ] } except asyncio.TimeoutError: # 降级返回空但记录可观测性事件 logger.warning(fretrieve tool timeout for query: {query[:50]}...) return 知识库暂时不可用请稍后再试。, {} except Exception as e: logger.error(fretrieve tool error: {e}) return 知识库服务异常请联系管理员。, {}3.2 LangGraph 的 State 不是变量而是对话的“数字孪生”MessagesState常被当作一个简单的消息列表但它真正的价值是把整个对话生命周期的状态映射为可序列化的数据结构。这意味着你可以在任意节点读取历史消息、工具调用记录、LLM 思考过程如果启用了streamTrue在generate节点不仅拼接最新工具结果还能检查“过去 3 轮是否连续失败”从而触发人工接管在query_or_respond节点根据state[messages][-1].content的情感倾向用轻量 sentiment 模型判断动态调整系统提示词。我们在线上系统中扩展了MessagesState加入自定义字段from typing import Dict, Any, Optional from langgraph.graph import MessagesState class CustomState(MessagesState): # 扩展字段当前对话的业务上下文 business_context: Optional[Dict[str, Any]] None # 扩展字段已触发的工具调用次数防循环 tool_call_count: int 0 # 扩展字段是否进入人工接管模式 is_handled_by_human: bool False # 在 graph_builder 中使用 graph_builder StateGraph(CustomState)这样当tool_call_count 5时generate节点可直接返回“已尝试多次为您转接人工客服”而不是继续无意义循环。3.3 Checkpoint 不是“存个状态”而是构建可审计的决策链PostgresSaver常被当成“让对话不丢失”的方案但它更大的价值是把每次 Agent 决策变成一条可 SQL 查询的审计日志。我们建了一张agent_execution_log表结构如下字段类型说明thread_idVARCHAR对话唯一ID用于关联整条链路node_nameVARCHAR当前执行节点名如 query_or_respondinput_messagesJSONB输入消息快照含用户原始输入output_messagesJSONB输出消息快照含 LLM 生成内容、工具调用参数execution_time_msBIGINT节点执行耗时statusVARCHARsuccess/timeout/errorcreated_atTIMESTAMPTZ时间戳有了这张表运营同学可以写 SQL 查“过去24小时retrieve工具超时率超过5%的 TOP 3 问题是什么”技术同学可以查“generate节点平均耗时突增是否和某次模型升级有关”。这才是 Agent 真正的“可运维性”。3.4 Agent 的终极考验当所有工具都失效时它是否还像个“人”最体现 Agent 成熟度的不是它能多炫技地调用工具而是它在工具全部不可用时的表现。我们强制要求所有 Agent 服务必须实现三级降级降级级别触发条件用户体验技术实现L1工具降级单个工具超时/报错“知识库暂不可用我用已有知识回答”返回fallback_response字段L2模型降级LLM API 全部不可用“系统繁忙请稍后再试”切换至本地小模型如 Qwen2-0.5BL3服务降级整个 Agent 服务宕机“转接到人工客服”通过 API 网关重定向到备用服务这个降级链不是理论而是写死在graph_builder.compile()的interrupt_before和interrupt_after钩子里。当监控系统检测到PostgresSaver连接失败会主动触发interrupt_beforegenerate强制跳过生成步骤直奔降级响应。4. 从零搭建一个生产级 RAGAgent 系统避开 17 个新手必踩的坑现在我们把前面所有原则落地为一个可直接部署的实战项目。目标构建一个能回答“公司内部技术文档”问题的智能助手支持网页抓取、PDF 解析、多轮对话、状态持久化。整个工程不依赖 OpenAI全部使用开源模型和本地服务。4.1 环境准备为什么必须用 Poetry 而不是 Pipenv很多教程用pip install -r requirements.txt但在生产环境中这会导致灾难性版本漂移。LangChain 生态更新极快langchain-core0.3.0和langchain-core0.3.1之间可能有 API 不兼容。Poetry 的poetry.lock文件能锁定每个依赖的精确哈希值确保poetry install在任何机器上安装的都是完全一致的包。初始化命令# 创建项目 poetry init -n # 添加核心依赖指定精确版本 poetry add langchain-core0.3.1 \ langchain-community0.3.1 \ langgraph0.2.47 \ langchain-postgres0.1.22 \ sentence-transformers3.1.1 \ pypdf4.2.0 \ unstructured0.10.28 # 添加开发依赖 poetry add pytest8.2.2 black24.4.2 mypy1.10.0 --group dev实操心得在pyproject.toml中务必禁用allow-prereleases true。LangChain 的-rc版本常有未文档化的 breaking change我们吃过亏——一个rc版本把RunnablePassthrough的invoke方法签名改了导致所有 pipeline 崩溃。4.2 向量数据库选型为什么 PGVector 比 Chroma 更适合生产Chroma 适合 demoPGVector 才是生产之选。原因有三事务一致性PGVector 基于 PostgreSQLadd_documents和similarity_search在同一事务内不会出现“入库成功但检索不到”的脏读权限管控可为不同业务线创建独立 schema用GRANT SELECT ON TABLE rag_schema.docs TO analyst_role控制数据访问可观测性EXPLAIN ANALYZE直接查看向量检索的执行计划优化索引。安装与初始化-- 创建专用数据库 CREATE DATABASE langchain_rag WITH OWNER postgres; -- 启用 pgvector 扩展 \c langchain_rag CREATE EXTENSION IF NOT EXISTS vector; CREATE EXTENSION IF NOT EXISTS uuid-ossp; -- 创建向量表带业务分区 CREATE TABLE IF NOT EXISTS rag_docs ( id UUID PRIMARY KEY DEFAULT uuid_generate_v4(), collection_name TEXT NOT NULL, content TEXT NOT NULL, embedding VECTOR(1024) NOT NULL, metadata JSONB DEFAULT {}, created_at TIMESTAMPTZ DEFAULT NOW() ); -- 创建高效向量索引HNSW CREATE INDEX IF NOT EXISTS idx_rag_docs_embedding ON rag_docs USING hnsw (embedding vector_cosine_ops) WITH (m 16, ef_construction 64);Python 初始化代码from langchain_postgres import PGVector from langchain_postgres.vectorstores import PGVector # 连接字符串生产环境从环境变量读取 CONNECTION_STRING postgresqlpsycopg2://postgres:passwordlocalhost:5432/langchain_rag # 初始化向量库自动建表 vector_store PGVector( embeddingsembeddings, # 使用 bge-small-zh-v1.5 collection_nametech_docs, connectionCONNECTION_STRING, use_jsonbTrue, # 使用 JSONB 存储 metadata支持高效查询 )4.3 文档加载与处理流水线一个函数搞定 PDF/网页/Markdown我们封装了一个IngestionPipeline类统一处理所有格式from langchain_community.document_loaders import PyPDFLoader, WebBaseLoader, UnstructuredMarkdownLoader from langchain_text_splitters import RecursiveCharacterTextSplitter class IngestionPipeline: def __init__(self, vector_store: PGVector): self.vector_store vector_store def load_from_url(self, url: str) - None: loader WebBaseLoader(web_paths(url,)) docs loader.load() self._process_and_store(docs, source_typeweb, urlurl) def load_from_pdf(self, file_path: str) - None: loader PyPDFLoader(file_path) docs loader.load() self._process_and_store(docs, source_typepdf, file_pathfile_path) def load_from_md(self, file_path: str) - None: loader UnstructuredMarkdownLoader(file_path) docs loader.load() self._process_and_store(docs, source_typemarkdown, file_pathfile_path) def _process_and_store(self, docs: List[Document], **metadata) - None: # 统一分层切分 processed_docs hierarchical_split(docs) # 批量向量化入库 self.vector_store.add_documents( documentsprocessed_docs, ids[str(uuid.uuid4()) for _ in processed_docs], metadatas[{**metadata, processed_at: datetime.now().isoformat()} for _ in processed_docs] ) print(f✅ 已入库 {len(processed_docs)} 个文档块) # 使用示例 pipeline IngestionPipeline(vector_store) pipeline.load_from_url(https://example.com/tech-docs.md) pipeline.load_from_pdf(./docs/architecture.pdf)4.4 LangGraph Agent 流程图从草图到可运行的 7 步我们用一个真实案例演示用户问“Kubernetes Pod 一直处于 Pending 状态可能原因有哪些”系统需先用 RAG 检索内部《K8s 故障排查手册》若检索结果不足 2 条再调用k8s_api_check工具查集群实时状态最终整合生成回答。Step 1定义状态from typing import Annotated, Sequence, TypedDict from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[Sequence[BaseMessage], add_messages] # 自定义字段 k8s_cluster_status: str retrieval_confidence: float 0.0Step 2定义工具带熔断tool def k8s_api_check() - str: 检查 K8s 集群资源状态模拟 try: # 实际调用 kubectl 或 Kubernetes API return 节点资源充足无 pending pod except Exception: return 集群 API 不可用Step 3定义节点函数def route_to_tool(state: State) - Literal[retriever, k8s_api_check, generate]: 路由到合适工具 last_message state[messages][-1].content # 用轻量模型判断是否需查集群非 LLM避免循环 if pending in last_message.lower() and k8s in last_message.lower(): return k8s_api_check return retriever def call_retriever(state: State) - State: # 调用 RAG 工具 result, artifact retrieve.invoke({query: state[messages][-1].content}) state[retrieval_confidence] artifact.get(confidence, 0.0) return {messages: [AIMessage(contentresult)]} def call_k8s_api(state: State) - State: result k8s_api_check.invoke({}) return {messages: [AIMessage(contentf集群状态{result})]}Step 4组装图关键带条件边from langgraph.graph import START, END, StateGraph from langgraph.prebuilt import ToolNode graph_builder StateGraph(State) # 添加节点 graph_builder.add_node(router, route_to_tool) graph_builder.add_node(retriever, call_retriever) graph_builder.add_node(k8s_api_check, call_k8s_api) graph_builder.add_node(generate, generate) # generate 函数见前文 # 设置入口 graph_builder.set_entry_point(router) # 添加条件边router 的输出决定走向 graph_builder.add_conditional_edges( router, lambda x: x, # 直接返回字符串 { retriever: retriever, k8s_api_check: k8s_api_check, generate: generate, # 直接生成兜底 } ) # 添加普通边 graph_builder.add_edge(retriever, generate) graph_builder.add_edge(k8s_api_check, generate) graph_builder.add_edge(generate, END)Step 5启用 Checkpoint生产必需from langgraph.checkpoint.postgres import PostgresSaver # 使用专用 checkpoint 数据库 CHECKPOINT_DB_URI postgresql://postgres:passwordlocalhost:5432/langchain_checkpoint with PostgresSaver.from_conn_string(CHECKPOINT_DB_URI) as checkpointer: checkpointer.setup() # 创建必要表 app graph_builder.compile(checkpointercheckpointer)Step 6暴露为 FastAPI 接口from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI() class ChatRequest(BaseModel): message: str thread_id: str app.post(/chat) async def chat_endpoint(request: ChatRequest): try: config {configurable: {thread_id: request.thread_id}} # 流式响应提升用户体验 async for event in app.astream( {messages: [HumanMessage(contentrequest.message)]}, configconfig, stream_modevalues ): if messages in event: last_msg event[messages][-1] if isinstance(last_msg, AIMessage): yield {response: last_msg.content} except Exception as e: raise HTTPException(status_code500, detailstr(e))Step 7部署与监控最后也是最重要的一步Dockerfile基础镜像用python:3.11-slim-bookworm安装psycopg2-binary和pgvector健康检查端点GET /health检查 PostgreSQL 连接、向量库连通性、LLM 模型加载状态Prometheus 指标暴露langchain_agent_invocations_total、langchain_retriever_latency_seconds、langchain_tool_errors_total日志规范所有logger.info()必须包含thread_id和node_name便于 ELK 关联追踪。踩坑实录我们第一次上线时PostgresSaver的连接池默认只有 5高并发下大量ConnectionPoolTimeoutError。解决方案是在PostgresSaver.from_conn_string()后手动设置min_size10, max_size50。这个参数在 LangChain 文档里藏得很深但却是生产稳定性的命脉。5. LangChain 与 LangGraph 的本质区别不是新旧版本而是两种编程范式网上充斥着“LangGraph 是 LangChain 的下一代”这类误导性说法。真相是LangChain 是函数式编程范式LangGraph 是状态机编程范式。它们解决的问题域不同适用场景也不同。强行用 LangGraph 去做简单 RAG就像用 React 写计算器——过度设计而用 LangChain 去做复杂 Agent就像用 jQuery 写操作系统——力不从心。5.1 LangChain面向“单次任务”的函数式流水线LangChain 的Runnable链如prompt | llm | output_parser本质是一个纯函数管道。它的优势在于极致简洁3 行代码完成 RAG调试友好每个节点可单独invoke()测试轻量嵌入可无缝集成到现有 Flask/FastAPI 服务中不改变架构。典型适用场景内部知识库问答单轮无状态自动生成周报固定模板 数据库查询邮件内容分类调用 LLM 规则兜底。# 一个完美的 LangChain RAG 链无状态、易测试 retriever vector_store.as_retriever(search_kwargs{k: 3}) rag_chain ( {context: retriever, question: RunnablePassthrough()} | prompt | llm | StrOutputParser() ) # 单元测试只需 mock retriever 和 llm5.2 LangGraph面向“多轮协作”的状态机编排LangGraph 的StateGraph是一个有状态的有限状态机FSM。它的核心特征状态持久化checkpointer让对话跨越进程、机器、甚至数天条件分支add_conditional_edges支持基于任意逻辑的路由不仅是工具调用循环控制可定义while式循环如“直到用户确认为止”人工干预点interrupt_before可在任意节点暂停交由人工审核。典型适用场景客服对话机器人需记忆用户身份、订单号、投诉历史自动化运维 Agent需多步诊断查日志 → 重启服务 → 验证状态金融风控 Agent需调用征信接口 → 评估额度 → 生成报告 → 发送短信。# LangGraph 的核心价值状态驱动的循环 def should_continue(state: State) - Literal[continue, end]: # 检查用户是否满意回答 last_ai_msg state[messages][-1].content if 满意 in last_ai_msg or ok in last_ai_msg.lower(): return end return continue graph_builder.add_conditional_edges( generate, should_continue, {continue: router, end: END} )5.3 如何选择一张决策树告诉你面对一个新需求按此流程决策graph TD A[需求是否需要记忆多轮对话状态] --|否| B[用 LangChain Runnable 链] A --|是| C[需求是否涉及多工具协作与条件路由] C --|否| D[用 LangChain Memory] C --|是| E[必须用 LangGraph] D -- F[例如聊天机器人记住用户