LangGraph實(shí)戰(zhàn)用狀態(tài)圖構(gòu)建多步Agent工作流-騰訊云開(kāi)發(fā)者社區(qū)-騰訊云用StateGraph實(shí)現(xiàn)分支、循環(huán)與人工審批LangGraph的StateGraph是如何用一張有向圖把分支判斷、循環(huán)重試、人工審批、斷點(diǎn)續(xù)傳這四件Chain做不了的事一次解決的。企業(yè)級(jí)Agent的本質(zhì)不是調(diào)用LLM是編排一個(gè)可靠的工作流。而工作流本質(zhì)上是一個(gè)狀態(tài)機(jī)。StateGraph核心四件套LangGraph的核心是StateGraph——一個(gè)有向圖狀態(tài)機(jī)。① State狀態(tài)一個(gè)共享的TypedDict所有節(jié)點(diǎn)都讀它、寫(xiě)它。是圖的血液。② Node節(jié)點(diǎn)一個(gè)函數(shù)輸入State輸出State的部分更新。是圖的背肌。③ Edge邊節(jié)點(diǎn)之間的連接?？梢允怯策匒一定到B。④ Conditional Edge條件邊一個(gè)路由函數(shù)讀當(dāng)前State決定下一個(gè)節(jié)點(diǎn)是誰(shuí)。這是Chain跟Graph拉開(kāi)差距的結(jié)構(gòu)。# 第一步定義狀態(tài) - 這是「血液」 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# 第二步定義節(jié)點(diǎn)函數(shù) 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讓文檔走不同路徑# 第三步路由函數(shù)關(guān)鍵 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接收的路由函數(shù)返回的是節(jié)點(diǎn)名字不是節(jié)點(diǎn)本身。這讓整個(gè)圖的路徑變成運(yùn)行時(shí)決定的而不是定義時(shí)硬編碼的。意味著LLM本身可以決定走哪里——這才是Agent而不是工作流。文檔入庫(kù) → LLM抽取 → 質(zhì)量檢查 → 人工審批 → 寫(xiě)入向量庫(kù)文檔處理工作流拓?fù)銼TART↓ingest文檔入庫(kù)↓llm_extractLLM抽取↓quality_check質(zhì)量檢查↓ Conditional Edgehuman_review人工審批|write_vector_db寫(xiě)入向量庫(kù)↓END低置信文檔自動(dòng)走人工審批分支高置信直接寫(xiě)庫(kù)第四章 Checkpointing讓Agent斷點(diǎn)續(xù)跑每次節(jié)點(diǎn)執(zhí)行結(jié)束自動(dòng)把State快照寫(xiě)到持久化存儲(chǔ)。流程中斷后從最近一個(gè)checkpoint恢復(fù)跳過(guò)已執(zhí)行的節(jié)點(diǎn)。這就把Agent的重啟重跑變成了重啟接上。從開(kāi)發(fā)到生產(chǎn)Checkpoint有三檔配置? 開(kāi)發(fā)期InMemorySaver— 內(nèi)存里存進(jìn)程退出就丟單元測(cè)試用。? 單機(jī)生產(chǎn)SqliteSaver— 本地SQLite文件進(jìn)程重啟不丟數(shù)據(jù)。? 集群生產(chǎn)PostgresSaver— 多進(jìn)程共享支持高并發(fā)企業(yè)首選。# 生產(chǎn)配置Postgres持久化 from langgraph.checkpoint.postgres \ import PostgresSaverDB_URI postgresql://user:pwdlocalhost:5432/agent_state checkpointer PostgresSaver.from_conn_string(DB_URI) checkpointer.setup() # 創(chuàng)建 schemagraph builder.compile(checkpointercheckpointer,interrupt_before[human_review]) # 用thread_id標(biāo)識(shí)一次會(huì)話(huà) config {configurable: {thread_id: doc-2026-001}} # 第一次運(yùn)行跑到human_review暫停 result graph.invoke({raw_text: doc}, config) # 進(jìn)程崩潰也沒(méi)關(guān)系下次 # 用同一個(gè)thread_id繼續(xù) result graph.invoke(None, config) # LangGraph自動(dòng)從checkpoint恢復(fù)這里的interrupt_before[human_review]就是HITLHuman-in-the-Loop的入口圖執(zhí)行到human_review節(jié)點(diǎn)前會(huì)暫停把當(dāng)前State持久化等待人工接入。這是企業(yè)Agent上線(xiàn)的硬門(mén)檻——沒(méi)有HITL就別談生產(chǎn)環(huán)境。還有一個(gè)被低估的特性時(shí)間旅行調(diào)試。你可以遍歷某個(gè)thread_id的所有checkpoint回到任意歷史狀態(tài)重跑。加上Postgres Checkpoint后最常見(jiàn)的兩類(lèi)問(wèn)題——進(jìn)程被k8s重啟導(dǎo)致執(zhí)行中斷和用戶(hù)中途修改輸入導(dǎo)致需要回滾——都從架構(gòu)層面消失了。第五章 Human-in-the-LoopAI暫停等人第四章的interrupt_before是HITL的入門(mén)款。LangGraph還有更細(xì)粒度的interrupt()函數(shù)讓節(jié)點(diǎn)內(nèi)部主動(dòng)暫停。這一點(diǎn)對(duì)企業(yè)級(jí)場(chǎng)景至關(guān)重要——很多審批不是前置審批而是過(guò)程中決策點(diǎn)。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后渲染審批面板 # 用戶(hù)決定后用Command恢復(fù) graph.invoke(Command(resumeapproved),config)interrupt()拋出的不是異常是一個(gè)暫停信號(hào)當(dāng)前State自動(dòng)持久化前端拿到interrupt數(shù)據(jù)渲染審批UI用戶(hù)決定后通過(guò)Command(resume...)把結(jié)果送回節(jié)點(diǎn)函數(shù)繼續(xù)執(zhí)行——就好像那個(gè)interrupt調(diào)用剛返回一樣。四種HITL經(jīng)典模式企業(yè)級(jí)Agent里基本都見(jiàn)過(guò)模式一 · 審批確認(rèn)低置信度結(jié)果暫停等審批approve則繼續(xù)reject則回爐。模式二 · 內(nèi)容編輯把LLM抽取結(jié)果交給人編輯編輯后的內(nèi)容回寫(xiě)到State繼續(xù)往下走。模式三 · 多選決策LLM給出幾個(gè)候選方案讓人選一個(gè)。最常見(jiàn)于工具選擇、動(dòng)作規(guī)劃。模式四 · 異常上報(bào)遇到無(wú)法處理的情況主動(dòng)停下來(lái)附上上下文給人避免Agent硬跑出錯(cuò)。第六章 SubGraph與選型對(duì)比SubGraph是LangGraph的模塊化機(jī)制——當(dāng)工作流變大時(shí)把文檔解析質(zhì)量校驗(yàn)索引寫(xiě)入等子任務(wù)封裝成可復(fù)用的子圖主圖只關(guān)心編排邏輯。# 子圖文檔解析 parse_subgraph build_parse_graph() # 子圖質(zhì)量校驗(yàn) validate_subgraph build_validate_graph() # 主圖直接把子圖作為節(jié)點(diǎn) main StateGraph(MainState) main.add_node(parse, parse_subgraph) main.add_node(validate, validate_subgraph) main.add_edge(parse, validate)框架核心抽象最佳場(chǎng)景不適合LangGraph有向狀態(tài)圖可控、可中斷、可觀(guān)測(cè)的企業(yè)工作流完全開(kāi)放式的多智能體涌現(xiàn)CrewAI角色任務(wù)角色化協(xié)作產(chǎn)品經(jīng)理工程師QA需要嚴(yán)格狀態(tài)機(jī)控制的流程AutoGen / AG2多Agent對(duì)話(huà)研究、頭腦風(fēng)暴、群聊式協(xié)作需要確定性輸出的生產(chǎn)場(chǎng)景Claude Agent SDK原生工具循環(huán)Anthropic生態(tài)深度集成多模型、跨Provider場(chǎng)景OpenAI Agents SDKHandoff交接OpenAI模型為主的輕量場(chǎng)景需要細(xì)粒度狀態(tài)持久化Agent工程的三層抽象第一層 · 調(diào)用Chain時(shí)代我能讓LLM跑起來(lái)。第二層 · 編排StateGraph時(shí)代我能讓多個(gè)LLM按圖協(xié)作。第三層 · 治理CheckpointHITL時(shí)代我能讓生產(chǎn)Agent可中斷、可審批、可觀(guān)測(cè)。混合檢索RAG多路召回Reranker重排模型實(shí)戰(zhàn)-騰訊云開(kāi)發(fā)者社區(qū)-騰訊云向量檢索和關(guān)鍵詞檢索用的是兩套完全不同的評(píng)分體系向量檢索給的是余弦相似度取值在 01 之間ES 的 BM25 分?jǐn)?shù)理論上沒(méi)有上界可能是 3.2也可能是 27.8。這兩個(gè)分?jǐn)?shù)直接放一起比大小就跟拿身高的厘米數(shù)跟體重的公斤數(shù)去排序一樣壓根不是一個(gè)量綱誰(shuí)排在前面全看運(yùn)氣。單路召回為什么不夠用向量檢索的盲區(qū)語(yǔ)義漂移向量檢索擅長(zhǎng)語(yǔ)義相近但對(duì)精確匹配特別不敏感。用戶(hù)問(wèn)CVE-2024-3094 影響哪些版本向量檢索會(huì)召回一堆講漏洞影響范圍安全補(bǔ)丁的相關(guān)文檔但很可能漏掉那篇標(biāo)題里精確寫(xiě)著這個(gè) CVE 編號(hào)、內(nèi)容卻是純表格沒(méi)什么語(yǔ)義描述的公告文檔——因?yàn)橄蛄磕Ｐ蛯?duì)編號(hào)、型號(hào)這類(lèi)字符串的語(yǔ)義表達(dá)能力天生就弱編號(hào)本身在向量空間里幾乎是噪聲。專(zhuān)有名詞、編號(hào)、精確術(shù)語(yǔ)向量檢索天生吃力。用戶(hù)問(wèn)服務(wù)突然掛了怎么排查文檔里寫(xiě)的是進(jìn)程異常退出后的故障定位方法兩句話(huà)意思一樣但共同出現(xiàn)的關(guān)鍵詞幾乎為零BM25 直接抓瞎。純關(guān)鍵詞檢索在口語(yǔ)化提問(wèn)場(chǎng)景下的 Top-5 召回率只有向量檢索的 60% 左右這個(gè)差距在客服場(chǎng)景尤其致命因?yàn)橛脩?hù)很少會(huì)用文檔里的官方措辭來(lái)提問(wèn)。向量檢索管意思相近關(guān)鍵詞檢索管字面精確兩者互補(bǔ)而不是互相替代。多路召回架構(gòu)誰(shuí)跟誰(shuí)并行查最終采用的是三路并行召回并行分發(fā)路徑A →Milvus向量檢索 Top 30管語(yǔ)義相近路徑B →ES BM25檢索 Top 30管字面精確路徑C →知識(shí)圖譜檢索 Top 10管實(shí)體關(guān)系三路結(jié)果匯總RRF融合排序Reranker精排Top 5 送入LLM前面提到分?jǐn)?shù)量綱不一致的問(wèn)題業(yè)界的標(biāo)準(zhǔn)解法是RRFReciprocal Rank Fusion倒數(shù)排名融合。它的核心思路特別聰明不看原始分?jǐn)?shù)只看排名rank排名是天然可比的第 1 名不管在哪一路都是最好不存在量綱問(wèn)題。公式很簡(jiǎn)單RRF_score(d) Σ 1 / (k rank_i(d))對(duì)文檔 d把它在每一路召回結(jié)果里的排名 rank_i 取倒數(shù)再累加k 是個(gè)平滑常數(shù)通常取 60防止排名靠前的文檔權(quán)重過(guò)大導(dǎo)致分?jǐn)?shù)差距失真。排名越靠前1/(krank) 越大貢獻(xiàn)越高一篇文檔如果同時(shí)在向量檢索和關(guān)鍵詞檢索里都排前幾名它的融合分?jǐn)?shù)會(huì)明顯高于只在一路里靠前的文檔——這正是我們想要的效果兩路都認(rèn)可的結(jié)果可信度更高?；旌蠙z索RAG實(shí)戰(zhàn)多路召回Reranker重排模型_多路檢索后還需要rerank嗎,為什么rerank后反而把標(biāo)準(zhǔn)答案排到了后面-CSDN博客def rrf_fusion(vector_results: list[str],bm25_results: list[str],k: int 60) - dict[str, float]: 對(duì)兩路召回結(jié)果做RRF融合 scores: dict[str, float] {} for rank, doc_id in enumerate( vector_results, start1 ): scores[doc_id] scores.get(doc_id, 0) 1 / (k rank)for rank, doc_id in enumerate(bm25_results, start1): scores[doc_id] scores.get(doc_id, 0) 1 / (k rank) return dict(sorted(scores.items(),keylambda x: x[1],reverseTrue))單路向量檢索 Top-5 命中率 76%單純拼接兩路結(jié)果不做融合直接各取一半是 79%用 RRF 融合之后到了 87%。業(yè)界經(jīng)驗(yàn)值 k60 是從信息檢索領(lǐng)域的大量實(shí)驗(yàn)里得出的沒(méi)有特殊場(chǎng)景不建議改。為什么還需要 RerankerReranker重排模型解決的正是這個(gè)問(wèn)題它是一個(gè)專(zhuān)門(mén)訓(xùn)練用來(lái)判斷query 和某段文檔到底有多相關(guān)的模型直接吃 query 和候選文檔的原始文本輸出一個(gè)精確的相關(guān)性分?jǐn)?shù)Bi-Encoder vs Cross-Encoder為什么 Rerank 不能用向量檢索代替向量檢索用的 Embedding 模型和 Reranker 模型都是判斷相關(guān)性為什么不能只用向量檢索答案在于兩者的模型架構(gòu)完全不同。Embedding 模型是Bi-Encoder架構(gòu)query 和文檔分別獨(dú)立編碼成向量之后算個(gè)余弦相似度。這種架構(gòu)的好處是可以提前把文檔向量算好存庫(kù)里檢索時(shí)只需要編碼 query速度極快能支撐百萬(wàn)級(jí)候選集的檢索。模型只能靠?jī)蓚€(gè)獨(dú)立向量的幾何距離去近似相關(guān)性精度天然有損失。Reranker 用的是Cross-Encoder架構(gòu)query 和文檔拼接成一個(gè)整體輸入模型讓模型在自注意力層里充分交互逐詞判斷相關(guān)性精度明顯更高。代價(jià)是沒(méi)法預(yù)計(jì)算每個(gè)候選文檔都要跟 query 現(xiàn)場(chǎng)拼接一次做推理計(jì)算量比 Bi-Encoder 高一個(gè)量級(jí)這也是為什么 Reranker 只能用在粗篩之后的少量候選上不可能拿去做百萬(wàn)級(jí)的初篩。Bi-EncoderEmbeddingquery和doc獨(dú)立編碼算余弦距離 速度快可預(yù)計(jì)算 適合百萬(wàn)級(jí)初篩Cross-EncoderRerankerquery和doc拼接后一起編碼 自注意力充分交互精度更高 速度慢只能用于精排少量候選模型Top-5準(zhǔn)確率單次延遲(20候選)部署方式僅RRF無(wú)Rerank87.0%--BGE-Reranker-v2-m393.4%約80ms私有化部署Cohere Rerank 394.1%約200ms含網(wǎng)絡(luò)API調(diào)用業(yè)務(wù)數(shù)據(jù)微調(diào)交叉編碼器96.7%約60ms私有化部署Rerank 這一步比 Embedding 更值得投入微調(diào)成本。mbedding 微調(diào)是給通用向量空間做局部調(diào)整收益有天花板而 Reranker 本身就是專(zhuān)門(mén)判斷相關(guān)性的模型用業(yè)務(wù)真實(shí)的 query-doc 對(duì)做微調(diào)相當(dāng)于直接教它認(rèn)識(shí)你的業(yè)務(wù)語(yǔ)言收益立竿見(jiàn)影。BGE-Reranker-v2-m3 打底拿線(xiàn)上積累的用戶(hù)反饋數(shù)據(jù)點(diǎn)贊/點(diǎn)踩人工復(fù)核持續(xù)微調(diào)數(shù)據(jù)敏感、有一定 ML 工程能力選 BGE-Reranker 私有化部署持續(xù)微調(diào)長(zhǎng)期收益最大批量推理是唯一解批量推理Batching把 20 個(gè) query-doc 對(duì)拼成一個(gè) batch 一次性丟進(jìn)模型from FlagEmbedding import FlagRerankerreranker FlagReranker(BAAI/bge-reranker-v2-m3,use_fp16True) def rerank(query: str, docs: list[str]) - list[float]: 一次批量推理而非逐個(gè)調(diào)用 pairs [[query, d] for d in docs] scores reranker.compute_score(pairs, batch_size32) return scores線(xiàn)上有大量高頻重復(fù)問(wèn)題客服場(chǎng)景尤其明顯怎么退款這類(lèi)問(wèn)題一天能問(wèn)幾百遍。對(duì)這種情況我們加了一層基于 query 語(yǔ)義相似度的緩存——不是精確字符串匹配緩存用戶(hù)措辭千變?nèi)f化命中率極低而是拿 query 的 Embedding 向量做近似去重語(yǔ)義高度相似的 query 直接復(fù)用之前的 Rerank 結(jié)果async def rerank_with_cache(query: str, docs: list[str]): q_vec await embed_query(query) # 查緩存語(yǔ)義相似度0.95視為同一query cached await semantic_cache.get(q_vec, threshold0.95) if cached: return cachedscores await rerank(query, docs) await semantic_cache.set(q_vec, scores, ttl3600) return scores上線(xiàn)之后統(tǒng)計(jì)緩存命中率大概在 35% 左右也就是三分之一的 Rerank 請(qǐng)求直接省掉了GPU 負(fù)載明顯降下來(lái)了。這里有個(gè)細(xì)節(jié)要提醒相似度閾值設(shè) 0.95 是我們業(yè)務(wù)場(chǎng)景客服問(wèn)答措辭相對(duì)集中跑出來(lái)的經(jīng)驗(yàn)值如果你的場(chǎng)景 query 多樣性很高這個(gè)閾值可能需要調(diào)得更寬松否則緩存命中率會(huì)很低白白多一層查詢(xún)開(kāi)銷(xiāo)。方案Recall5MRRP99延遲純向量檢索76.2%0.6845ms純BM25檢索71.5%0.6320ms雙路RRF融合87.0%0.7970ms雙路RRFReranker96.7%0.94155ms從單路到雙路RRFRerankerRecall5 提升了 20 個(gè)百分點(diǎn)MRR衡量正確答案排名靠前程度的指標(biāo)提升了近 40%代價(jià)是延遲從 45 毫秒漲到 155 毫秒。