企业官网建设流程全解析

LangGraph实战：用状态图构建多步Agent工作流-腾讯云开发者社区-腾讯云

用StateGraph实现分支、循环与人工审批

LangGraph的StateGraph是如何用一张有向图把"分支判断、循环重试、人工审批、断点续传"这四件Chain做不了的事，一次解决的。

企业级Agent的本质不是"调用LLM"，是"编排一个可靠的工作流"。而工作流本质上是一个状态机。
StateGraph核心四件套

LangGraph的核心是StateGraph——一个有向图状态机。

① State（状态）：一个共享的TypedDict，所有节点都读它、写它。是"图的血液"。

② Node（节点）：一个函数，输入State，输出State的部分更新。是"图的背肌"。

③ Edge（边）：节点之间的连接。可以是硬边（A一定到B）。

④ Conditional Edge（条件边）：一个路由函数，读当前State，决定下一个节点是谁。这是Chain跟Graph拉开差距的结构。

# 第一步：定义状态 - 这是「血液」 from typing import TypedDict from langgraph.graph import StateGraph, ENDclass DocState(TypedDict): raw_text: str extracted: dict confidence: float needs_review: bool final_status: str

# 第二步：定义节点函数 def extract_node(state: DocState): result = llm.invoke(state["raw_text"]) return { "extracted": result.data, "confidence": result.score } def check_node(state: DocState): threshold = 0.85 return {"needs_review":state["confidence"] < threshold}

Conditional Edge让文档走不同路径：

# 第三步：路由函数（关键！） def route_after_check(state: DocState): if state["needs_review"]: return "human_review" return "write_to_db" # 第四步：装配图 builder = StateGraph(DocState) builder.add_node("extract", extract_node) builder.add_node("check", check_node) builder.add_node("human_review", hitl_node) builder.add_node("write_to_db", db_node) builder.set_entry_point("extract") builder.add_edge("extract", "check") builder.add_conditional_edges("check", route_after_check) builder.add_edge("write_to_db", END) graph = builder.compile()

add_conditional_edges接收的路由函数返回的是节点名字，不是节点本身。这让整个图的路径变成运行时决定的，而不是定义时硬编码的。意味着LLM本身可以决定走哪里——这才是Agent，而不是工作流。

文档入库 → LLM抽取 → 质量检查 → 人工审批 → 写入向量库

文档处理工作流拓扑

START

↓

ingest（文档入库）

↓

llm_extract（LLM抽取）

↓

quality_check（质量检查）

↓ Conditional Edge

human_review（人工审批）

|

write_vector_db（写入向量库）

↓

END

低置信文档自动走人工审批分支，高置信直接写库

第四章 Checkpointing：让Agent"断点续跑"

每次节点执行结束，自动把State快照写到持久化存储。流程中断后，从最近一个checkpoint恢复，跳过已执行的节点。这就把Agent的"重启=重跑"变成了"重启=接上"。

从开发到生产，Checkpoint有三档配置：

▸ 开发期：InMemorySaver— 内存里存，进程退出就丢，单元测试用。

▸ 单机生产：SqliteSaver— 本地SQLite文件，进程重启不丢数据。

▸ 集群生产：PostgresSaver— 多进程共享，支持高并发，企业首选。

# 生产配置：Postgres持久化 from langgraph.checkpoint.postgres \ import PostgresSaverDB_URI = "postgresql://user:pwd@"localhost:5432/agent_state" checkpointer = PostgresSaver.from_conn_string(DB_URI) checkpointer.setup() # 创建 schemagraph = builder.compile(checkpointer=checkpointer,interrupt_before=["human_review"]) # 用thread_id标识一次会话 config = {"configurable": {"thread_id": "doc-2026-001"}} # 第一次运行：跑到human_review暂停 result = graph.invoke({"raw_text": doc}, config) # 进程崩溃也没关系，下次： # 用同一个thread_id继续 result = graph.invoke(None, config) # LangGraph自动从checkpoint恢复

这里的interrupt_before=["human_review"]就是HITL（Human-in-the-Loop）的入口：图执行到human_review节点前会暂停，把当前State持久化，等待人工接入。这是企业Agent上线的硬门槛——没有HITL就别谈生产环境。

还有一个被低估的特性：时间旅行调试。你可以遍历某个thread_id的所有checkpoint，回到任意历史状态重跑。

加上Postgres Checkpoint后，最常见的两类问题——"进程被k8s重启导致执行中断"和"用户中途修改输入导致需要回滚"——都从架构层面消失了。

第五章 Human-in-the-Loop：AI暂停等人

第四章的interrupt_before是HITL的"入门款"。LangGraph还有更细粒度的interrupt()函数，让节点内部主动暂停。这一点对企业级场景至关重要——很多审批不是"前置审批"，而是"过程中决策点"。

from langgraph.types import interrupt from langgraph.types import Commanddef human_review_node( state: DocPipelineState): # 把决策上下文交给人 decision = interrupt({ "doc_id": state["doc_id"], "meta": state["extracted_meta"], "confidence": state["confidence"], "prompt": "approve / reject / edit" }) return {"review_decision": decision}# 前端拿到interrupt后渲染审批面板 # 用户决定后用Command恢复 graph.invoke(Command(resume="approved"),config)

interrupt()抛出的不是异常，是一个"暂停信号"：当前State自动持久化，前端拿到interrupt数据，渲染审批UI；用户决定后通过Command(resume=...)把结果送回，节点函数继续执行——就好像那个interrupt调用刚返回一样。

四种HITL经典模式，企业级Agent里基本都见过：

模式一 · 审批确认：低置信度结果暂停等审批，approve则继续，reject则回炉。

模式二 · 内容编辑：把LLM抽取结果交给人编辑，编辑后的内容回写到State继续往下走。

模式三 · 多选决策：LLM给出几个候选方案，让人选一个。最常见于工具选择、动作规划。

模式四 · 异常上报：遇到无法处理的情况主动停下来，附上上下文给人，避免Agent硬跑出错。

第六章 SubGraph与选型对比

SubGraph是LangGraph的"模块化机制"——当工作流变大时，把"文档解析""质量校验""索引写入"等子任务封装成可复用的子图，主图只关心编排逻辑。

# 子图：文档解析 parse_subgraph = build_parse_graph() # 子图：质量校验 validate_subgraph = build_validate_graph() # 主图：直接把子图作为节点 main = StateGraph(MainState) main.add_node("parse", parse_subgraph) main.add_node("validate", validate_subgraph) main.add_edge("parse", "validate")

Agent工程的三层抽象：

第一层 · 调用：Chain时代，"我能让LLM跑起来"。

第二层 · 编排：StateGraph时代，"我能让多个LLM按图协作"。

第三层 · 治理：Checkpoint+HITL时代，"我能让生产Agent可中断、可审批、可观测"。

混合检索RAG：多路召回+Reranker重排模型实战-腾讯云开发者社区-腾讯云

向量检索和关键词检索用的是两套完全不同的评分体系，向量检索给的是余弦相似度，取值在 0～1 之间；ES 的 BM25 分数理论上没有上界，可能是 3.2，也可能是 27.8。这两个分数直接放一起比大小，就跟拿身高的"厘米数"跟体重的"公斤数"去排序一样，压根不是一个量纲，谁排在前面全看运气。

单路召回为什么不够用

向量检索的盲区：语义漂移

向量检索擅长"语义相近"，但对精确匹配特别不敏感。

用户问"CVE-2024-3094 影响哪些版本"，向量检索会召回一堆讲"漏洞影响范围""安全补丁"的相关文档，但很可能漏掉那篇标题里精确写着这个 CVE 编号、内容却是纯表格没什么语义描述的公告文档——因为向量模型对编号、型号这类字符串的语义表达能力天生就弱，编号本身在向量空间里几乎是"噪声"。专有名词、编号、精确术语，向量检索天生吃力。

用户问"服务突然挂了怎么排查"，文档里写的是"进程异常退出后的故障定位方法"，两句话意思一样，但共同出现的关键词几乎为零，BM25 直接抓瞎。纯关键词检索在"口语化提问"场景下的 Top-5 召回率只有向量检索的 60% 左右，这个差距在客服场景尤其致命，因为用户很少会用文档里的"官方措辞"来提问。

向量检索管"意思相近"，关键词检索管"字面精确"，两者互补而不是互相替代。

多路召回架构：谁跟谁并行查

最终采用的是三路并行召回

并行分发

路径A →Milvus向量检索 Top 30，管语义相近

路径B →ES BM25检索 Top 30，管字面精确

路径C →知识图谱检索 Top 10，管实体关系

三路结果汇总

RRF融合排序

Reranker精排

Top 5 送入LLM

前面提到分数量纲不一致的问题，业界的标准解法是RRF（Reciprocal Rank Fusion，倒数排名融合）。它的核心思路特别聪明：不看原始分数，只看排名（rank），排名是天然可比的，第 1 名不管在哪一路都是"最好"，不存在量纲问题。

公式很简单：

RRF_score(d) = Σ 1 / (k + rank_i(d))

对文档 d，把它在每一路召回结果里的排名 rank_i 取倒数再累加，k 是个平滑常数（通常取 60），防止排名靠前的文档权重过大导致分数差距失真。排名越靠前，1/(k+rank) 越大，贡献越高；一篇文档如果同时在向量检索和关键词检索里都排前几名，它的融合分数会明显高于只在一路里靠前的文档——这正是我们想要的效果：两路都认可的结果，可信度更高。

混合检索RAG实战：多路召回+Reranker重排模型_多路检索后还需要rerank吗,为什么rerank后反而把标准答案排到了后面-CSDN博客

def rrf_fusion(vector_results: list[str],bm25_results: list[str],k: int = 60) -> dict[str, float]: """对两路召回结果做RRF融合""" scores: dict[str, float] = {} for rank, doc_id in enumerate( vector_results, start=1 ): scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank)for rank, doc_id in enumerate(bm25_results, start=1): scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank) return dict(sorted(scores.items(),key=lambda x: x[1],reverse=True))

单路向量检索 Top-5 命中率 76%，单纯拼接两路结果（不做融合，直接各取一半）是 79%，用 RRF 融合之后到了 87%。

业界经验值 k=60 是从信息检索领域的大量实验里得出的，没有特殊场景不建议改。

为什么还需要 Reranker

Reranker（重排模型）解决的正是这个问题：它是一个专门训练用来判断"query 和某段文档到底有多相关"的模型，直接吃 query 和候选文档的原始文本，输出一个精确的相关性分数

Bi-Encoder vs Cross-Encoder：为什么 Rerank 不能用向量检索代替

向量检索用的 Embedding 模型和 Reranker 模型，都是"判断相关性"，为什么不能只用向量检索？答案在于两者的模型架构完全不同。

Embedding 模型是Bi-Encoder架构：query 和文档分别独立编码成向量，之后算个余弦相似度。这种架构的好处是可以提前把文档向量算好存库里，检索时只需要编码 query，速度极快，能支撑百万级候选集的检索。模型只能靠"两个独立向量的几何距离"去近似相关性，精度天然有损失。

Reranker 用的是Cross-Encoder架构：query 和文档拼接成一个整体输入模型，让模型在自注意力层里充分交互，逐词判断相关性，精度明显更高。

代价是没法预计算，每个候选文档都要跟 query 现场拼接一次做推理，计算量比 Bi-Encoder 高一个量级，这也是为什么 Reranker 只能用在"粗筛之后的少量候选"上，不可能拿去做百万级的初筛。

Bi-Encoder（Embedding）

query和doc独立编码，算余弦距离， 速度快，可预计算， 适合百万级初筛

Cross-Encoder（Reranker）

query和doc拼接后一起编码， 自注意力充分交互，精度更高， 速度慢，只能用于精排少量候选

Rerank 这一步比 Embedding 更值得投入微调成本。

mbedding 微调是给"通用向量空间"做局部调整，收益有天花板；而 Reranker 本身就是"专门判断相关性"的模型，用业务真实的 query-doc 对做微调，相当于直接教它认识你的业务语言，收益立竿见影。

BGE-Reranker-v2-m3 打底，拿线上积累的用户反馈数据（点赞/点踩+人工复核）持续微调

数据敏感、有一定 ML 工程能力，选 BGE-Reranker 私有化部署+持续微调，长期收益最大；

批量推理是唯一解

批量推理（Batching），把 20 个 query-doc 对拼成一个 batch 一次性丢进模型：

from FlagEmbedding import FlagRerankerreranker = FlagReranker("BAAI/bge-reranker-v2-m3",use_fp16=True) def rerank(query: str, docs: list[str]) -> list[float]: """一次批量推理，而非逐个调用""" pairs = [[query, d] for d in docs] scores = reranker.compute_score(pairs, batch_size=32) return scores

线上有大量高频重复问题（客服场景尤其明显，"怎么退款"这类问题一天能问几百遍）。对这种情况，我们加了一层基于 query 语义相似度的缓存——不是精确字符串匹配缓存（用户措辞千变万化，命中率极低），而是拿 query 的 Embedding 向量做近似去重，语义高度相似的 query 直接复用之前的 Rerank 结果：

async def rerank_with_cache(query: str, docs: list[str]): q_vec = await embed_query(query) # 查缓存：语义相似度>0.95视为同一query cached = await semantic_cache.get(q_vec, threshold=0.95) if cached: return cachedscores = await rerank(query, docs) await semantic_cache.set(q_vec, scores, ttl=3600) return scores

上线之后统计缓存命中率大概在 35% 左右，也就是三分之一的 Rerank 请求直接省掉了，GPU 负载明显降下来了。这里有个细节要提醒：相似度阈值设 0.95 是我们业务场景（客服问答，措辞相对集中）跑出来的经验值，如果你的场景 query 多样性很高，这个阈值可能需要调得更宽松，否则缓存命中率会很低，白白多一层查询开销。

从单路到"双路+RRF+Reranker"，Recall@5 提升了 20 个百分点，MRR（衡量正确答案排名靠前程度的指标）提升了近 40%，代价是延迟从 45 毫秒涨到 155 毫秒。