很多 Agent 項目跑不好不是模型不夠強而是流程沒有邊界。SOP 寫在文檔里模型執(zhí)行的時候經(jīng)常漏步驟、跳判斷、把不該并行的事并行掉把 SOP 畫成一張有向圖之后每一步是什么、下一步由誰決定、異常走哪條邊全部落成數(shù)據(jù)結構。Agent 不再靠“臨場發(fā)揮”決定順序而是在圖上按規(guī)則前進這就是 Graph Engineering 的核心思路。這篇文章不聊概念空轉直接拆解一套可以落地的做法把現(xiàn)有 SOP 拆成節(jié)點和條件邊用代碼構建可執(zhí)行圖跑通之后封裝成 API 服務再接入批量任務。如果已經(jīng)在用 LangGraph、自研編排引擎或者準備改造一個多步驟 Agent這篇文章可以直接收藏。1. Graph Engineering 是什么先解決流程失控問題Graph Engineering 用圖數(shù)據(jù)結構來編排 Agent 的執(zhí)行流程。圖里常見的幾個要素是Node一個具體工作步驟比如“識別意圖”“檢索知識庫”“調用訂單接口”“生成回復”。Edge節(jié)點之間的連接關系表達“上一步完成后進入下一步”。State所有節(jié)點共享的狀態(tài)對象保存用戶輸入、中間結果、最終輸出。Conditional Edge條件路由由規(guī)則或 LLM 判斷后選擇走哪條邊。Loop循環(huán)邊用來實現(xiàn)多輪修正、重試、繼續(xù)追問。Checkpoint檢查點把執(zhí)行到一半的狀態(tài)落盤異常后可以從斷點恢復。之所以說“把 SOP 畫成 Graph它就能自己跑”是因為圖結構把流程控制權從模型手里收回來了。模型只負責節(jié)點內部的計算比如文本分類、內容生成、信息抽取下一步去哪個節(jié)點由預先定義好的邊和條件路由決定。這樣既保留了 LLM 的語義判斷能力又不至于讓它把關鍵步驟帶偏。核心能力速覽能力項說明流程表達用節(jié)點和邊表達多步驟 SOP支持順序、分支、循環(huán)、并行狀態(tài)管理節(jié)點之間通過全局 State 共享數(shù)據(jù)避免口頭傳參條件路由支持規(guī)則路由和 LLM 路由模型判斷后進入不同分支可觀測性每個節(jié)點執(zhí)行前、執(zhí)行后都能打日志定位問題到具體步驟可恢復性結合 Checkpoint 可在失敗后恢復不用整條流程重跑接口化編譯后的圖對象可以封裝成 HTTP API供外部系統(tǒng)調用批量支持同一張圖可并發(fā)處理多條任務配合限流和隊列使用從這里開始后面所有內容都圍繞這個目標展開把一個普通的客服處理流程變成一張可運行、可測試、可上線的 Agent 圖。2. 適用場景與使用邊界Graph Engineering 最適合“步驟清晰但中間有判斷”的業(yè)務場景。比如客服工單處理先判斷用戶意圖再決定走訂單查詢分支、知識庫分支還是投訴升級分支。內容審核先做基礎過濾再由 LLM 判斷風險等級高風險進入人工復核。數(shù)據(jù)分析報告先取數(shù)再清洗再讓 LLM 生成結論最后交給審查節(jié)點。內部工具編排把多個內部系統(tǒng)的操作串成一條固定流程每一步都有權限校驗。文檔處理先解析文件再分塊再檢索再生成摘要最后轉成 Markdown 輸出。這類場景有一個共同點流程主干是確定的只是中途存在若干需要判斷的分岔口。Graph 正好把“確定的部分”和“判斷的部分”分開管理。Graph Engineering 不適合的場景也要說清楚。如果任務是開放閑聊、開放式探索、完全不確定的自主行為圖的節(jié)點邊界反而會變成限制。比如“幫我在互聯(lián)網(wǎng)上找資料并自主決定下一步”很難預先畫出一張完整的圖更適合用自由式 Agent Loop 或 Plan-and-Execute 配合動態(tài)規(guī)劃。還有一類場景是長時記憶高度依賴型任務對話系統(tǒng)依賴長期歷史記憶把狀態(tài)全部塞進圖 State 會讓結構非常臃腫。使用邊界方面需要特別注意圖編排本身不解決數(shù)據(jù)權限問題。節(jié)點里可能調用訂單接口、讀取用戶信息、訪問內部資料庫這些操作必須提前做好角色權限控制。涉及個人信息時要脫敏涉及外部版權內容時只能檢索和引用不能隨意生成和分發(fā)。上線前至少確認三件事誰有權觸發(fā)這個流程、流程中間能不能插人工審批、輸出內容是否需要復核。3. 環(huán)境準備與前置條件Graph Engineering 依賴不復雜核心是 Python 環(huán)境和一個圖編排框架。這里以 LangGraph 生態(tài)為例因為它和 LangChain 生態(tài)銜接好社區(qū)資料也多。前置環(huán)境清單項目建議要求操作系統(tǒng)Windows 10/11、macOS、Linux 均可Python3.9 以上建議 3.10 或 3.11包管理工具pip建議配合 virtualenv 或 condaLLM 服務需要可訪問的 OpenAI 兼容 API或本地部署模型圖編排框架langgraph按實際項目版本安裝磁盤空間純代碼編排很小程序依賴通常 2-5GB 以內顯存如果只調遠程 LLM不強制獨顯如果用本地模型按模型要求準備先創(chuàng)建虛擬環(huán)境并安裝依賴python -m venv .venv source .venv/bin/activate # Windows 使用 .venv\Scripts\activate pip install langgraph langchain-core openai fastapi uvicorn說明一下LangGraph 的 API 在不同版本之間會有調整安裝之后如果遇到函數(shù)名不一致以當前版本的官方文檔為準。下面的代碼按常見寫法演示核心邏輯不變。還需要準備 LLM 服務的訪問憑證。建議放到環(huán)境變量而不是直接寫死在代碼里export OPENAI_API_KEY你的密鑰 export OPENAI_BASE_URL你的接口地址如果不準備用任何第三方圖編排框架也可以自研一個極簡圖引擎這個方案放在第 5 節(jié)單獨說明。自研的好處是零依賴缺點是可視化、斷點恢復、并發(fā)調度都要自己寫適合學習不建議直接上生產(chǎn)。4. 把 SOP 拆成 Graph 節(jié)點一套可直接套用的方法不要一上來就寫代碼先把 SOP 畫出來。拆解過程建議按下面六步走。第一步列出完整步驟清單。把現(xiàn)有 SOP 從開始到結束的每一步寫出來不要省略人覺得“理所當然”的環(huán)節(jié)。判斷、記錄、重試都屬于步驟。第二步找出判斷點。每一步里只要出現(xiàn)“如果”“否則”“根據(jù) XX 決定”這就是一個條件路由。條件路由可以交給規(guī)則也可以交給 LLM。第三步找出可并行步驟。如果兩個節(jié)點互不依賴且執(zhí)行順序不影響結果就設計成并行分支。例如“檢索知識庫”和“查詢訂單狀態(tài)”互不依賴可以并行。第四步定義狀態(tài)字段。把每個節(jié)點的輸入輸出字段列出來。狀態(tài)字段越收斂越好不要把所有臨時變量都塞進去。第五步為每個節(jié)點寫函數(shù)簽名。每個節(jié)點函數(shù)接收 State 類型返回一個 dict更新 State 中需要變化的字段。第六步組裝圖。先用順序邊連接主干再用條件路由把分支接上最后編譯運行。用一個實際的客服工單 SOP 舉例。假設原始流程是這樣的用戶發(fā)起咨詢 → 系統(tǒng)識別用戶意圖 → 如果用戶要查訂單調用訂單系統(tǒng)查詢如果用戶是普通咨詢檢索知識庫 → 根據(jù)查詢結果生成回復 → 高風險內容進入人工復核否則直接回復。拆成圖節(jié)點后是這樣原流程步驟圖節(jié)點節(jié)點類型主要輸出字段識別用戶意圖classify_input普通節(jié)點或 LLM 節(jié)點intent檢索知識庫search_kb普通節(jié)點kb_result查詢訂單系統(tǒng)query_order工具節(jié)點order_result生成回復generate_answerLLM 節(jié)點final_answer是否進入人工復核check_risk條件路由risk_level這個例子會在下一節(jié)直接用代碼實現(xiàn)。注意一個細節(jié)訂單查詢和知識庫檢索不是必然全部執(zhí)行它們由 classify_input 的條件路由決定走哪條邊這才是圖和順序代碼的核心區(qū)別。5. 實操最小可運行的 Agent 圖這一節(jié)寫一個完整示例演示怎么把上面那張圖變成可執(zhí)行代碼。示例使用 LangGraph 的常見寫法。5.1 定義工作流狀態(tài)先定義一個 TypedDict也就是所有節(jié)點共享的狀態(tài)結構from typing import Literal, TypedDict from langgraph.graph import StateGraph, START, END class WorkflowState(TypedDict): user_input: str intent: str kb_result: str order_result: str final_answer: str字段含義很清楚user_input 是用戶原始輸入intent 是分類結果kb_result 和 order_result 分別是兩個分支產(chǎn)生的中間結果final_answer 是最終回復。5.2 寫節(jié)點函數(shù)每個節(jié)點函數(shù)只做一件事讀取需要的字段返回需要更新的字段。這里先用規(guī)則代替 LLM保證示例能脫離外部 API 運行def classify_input(state: WorkflowState) - dict: text state[user_input] if 訂單 in text or 發(fā)貨 in text or 物流 in text: return {intent: order} return {intent: consult} def search_kb(state: WorkflowState) - dict: # 實際項目里這里可以調用檢索服務或向量數(shù)據(jù)庫 return {kb_result: 知識庫檢索結果用戶咨詢的是物流標準的常見問題解答。} def query_order(state: WorkflowState) - dict: # 實際項目里這里可以調用內部訂單接口 return {order_result: 訂單 OD-2024-001 已發(fā)貨預計 2 天后到達。} def generate_answer(state: WorkflowState) - dict: if state[intent] order: answer f關于「{state[user_input]}」的查詢結果{state[order_result]} else: answer f關于「{state[user_input]}」的回復{state[kb_result]} return {final_answer: answer}節(jié)點函數(shù)返回值會覆蓋 State 中對應字段。這里故意把每個節(jié)點保持成純函數(shù)形態(tài)后續(xù)測試非常省事。5.3 建圖、加條件路由、編譯接下來構建圖結構graph StateGraph(WorkflowState) graph.add_node(classify, classify_input) graph.add_node(kb, search_kb) graph.add_node(order, query_order) graph.add_node(answer, generate_answer) graph.add_edge(START, classify) def route_by_intent(state: WorkflowState) - Literal[consult, order]: return state[intent] graph.add_conditional_edges( classify, route_by_intent, { consult: kb, order: order, }, ) graph.add_edge(kb, answer) graph.add_edge(order, answer) graph.add_edge(answer, END) app graph.compile()條件路由這一段是核心。classify 節(jié)點執(zhí)行完會調用 route_by_intent拿到返回字符串再根據(jù)映射字典決定邊走向。如果返回的字符串不在映射字典里LangGraph 會報錯所以路由函數(shù)必須保證返回值受控?，F(xiàn)在跑一次result app.invoke({user_input: 我的訂單為什么還沒發(fā)貨}) print(result[final_answer])輸出符合預期關于「我的訂單為什么還沒發(fā)貨」的查詢結果訂單 OD-2024-001 已發(fā)貨預計 2 天后到達。再跑一個咨詢類輸入result app.invoke({user_input: 一般快遞多久能送到}) print(result[final_answer])會走知識庫分支。到這里一張最小 Agent 圖已經(jīng)跑通了。5.4 把節(jié)點換成真實 LLM把 classify_input 換成 LLM 判斷核心思路是用 LLM 返回結構化結果然后寫回 State。示例使用 OpenAI 兼容接口import json import os from openai import OpenAI client OpenAI( api_keyos.environ[OPENAI_API_KEY], base_urlos.environ.get(OPENAI_BASE_URL, https://api.openai.com/v1), ) def classify_input_with_llm(state: WorkflowState) - dict: response client.chat.completions.create( modelgpt-4o-mini, messages[ { role: system, content: ( 你是一個意圖識別模塊。只從下面兩個選項中選擇一個不要輸出其他內容。 選項consult、order ), }, { role: user, content: state[user_input], }, ], temperature0, ) intent response.choices[0].message.content.strip() return {intent: intent}然后建圖時把節(jié)點替換成這個函數(shù)即可graph StateGraph(WorkflowState) graph.add_node(classify, classify_input_with_llm) # 其余邊和上面保持一致這里要注意如果 LLM 返回了預期之外的內容條件路由會直接報錯。更穩(wěn)妥的做法是在路由函數(shù)里加歸一化處理def route_by_intent(state: WorkflowState) - Literal[consult, order]: if state[intent].startswith(訂單) or order in state[intent]: return order return consult在這個階段把“模型判斷”和“流程控制”解耦開后續(xù)換模型、換提示詞都不用動圖結構。5.5 不引入框架的自研方案如果只是想先驗證思路也可以不裝 LangGraph用普通 Python 函數(shù)和 if/else 模擬同樣的流程def run_workflow(user_input: str) - dict: state {user_input: user_input} state.update(classify_input(state)) if state[intent] order: state.update(query_order(state)) else: state.update(search_kb(state)) state.update(generate_answer(state)) return state這個方案更容易理解但可擴展性差很多新增節(jié)點要改函數(shù)體并行和循環(huán)都要手寫斷點恢復也沒有現(xiàn)成機制。所以生產(chǎn)環(huán)境建議直接用圖編排框架自研版本只適合作為教學驗證。6. 封裝 API 與批量任務圖跑通之后第一步是把編譯好的 app 對象封裝成 HTTP 接口方便前端、定時任務和其他后端系統(tǒng)調用。6.1 用 FastAPI 暴露接口from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class TaskRequest(BaseModel): user_input: str class TaskResponse(BaseModel): final_answer: str app.post(/run, response_modelTaskResponse) def run_task(req: TaskRequest) - TaskResponse: result app.invoke({user_input: req.user_input}) return TaskResponse(final_answerresult[final_answer])啟動服務uvicorn server:app --host 127.0.0.1 --port 8000調用測試curl -X POST http://127.0.0.1:8000/run \ -H Content-Type: application/json \ -d {user_input: 我的訂單為什么還沒發(fā)貨}返回結果也是 JSON直接把圖執(zhí)行能力暴露成標準 API。需要注意這個示例沒有加鑒權生產(chǎn)環(huán)境要放在內網(wǎng)或加 Token 校驗不能裸奔到公網(wǎng)。6.2 批量任務與并發(fā)控制批量任務的核心是把每條輸入當成獨立的一次 graph.invoke然后控制并發(fā)。并發(fā)過高會打滿上游接口所以建議用限流器from concurrent.futures import ThreadPoolExecutor, as_completed inputs [ {user_input: 訂單怎么還沒發(fā)貨}, {user_input: 一般快遞要多久}, ] with ThreadPoolExecutor(max_workers4) as pool: futures [pool.submit(app.invoke, item) for item in inputs] for future in as_completed(futures): result future.result() print(result[final_answer])批量任務要做好三件事事項建議失敗重試使用帶退避的重試機制避免瞬時故障直接丟任務日志記錄記錄輸入 ID、節(jié)點名、開始時間、結束時間、輸出摘要任務去重批量任務要有唯一任務 ID避免重復提交造成重復處理如果單條任務運行時間很長還要考慮把任務放進隊列用 Worker 異步消費而不是在請求線程里同步等待。7. 資源占用與運行觀察先明確一個認知圖編排引擎本身占用的算力極低。它主要做節(jié)點調度、狀態(tài)合并、邊判斷不承擔模型推理。真正消耗資源的是三個地方LLM 調用次數(shù)、外部接口響應時間、State 體積。LLM 調用次數(shù)是第一個要盯的指標。一條任務如果經(jīng)過 5 個節(jié)點每個節(jié)點都調一次 LLM那就是 5 次調用。批量 1000 條任務就是 5000 次。調整路由策略把不需要 LLM 的節(jié)點改成規(guī)則判斷能直接砍掉大半成本。外部接口響應時間是第二個指標。圖流程里如果查詢訂單接口耗時 3 秒那整條鏈路就被卡住 3 秒。建議優(yōu)先做并行分支把兩個不相干的接口調用放到同一批執(zhí)行。State 體積是第三個指標。每次節(jié)點返回的 dict 都會被合并進 State。如果節(jié)點塞入大段文本、完整日志、全量 JSONState 會越來越臃腫傳輸和調試都會變慢。合理做法是只保留后續(xù)節(jié)點需要的字段中間大對象建議落到外部存儲只把引用 ID 放進 State。如果要系統(tǒng)化觀察建議在每條任務里記錄import time start time.time() result app.invoke(input_data) print(ftask complete, cost{time.time() - start:.2f}s)更細的節(jié)點級耗時可以給每個節(jié)點函數(shù)外層包一層帶日志的裝飾器把節(jié)點名、入?yún)⒄?、耗時和返回字段名打出來。這樣在線上一眼能看到瓶頸在哪個節(jié)點。顯存占用方面如果 LLM 走遠程 API本機基本不需要顯存。如果節(jié)點里跑本地模型那就要按本地模型的要求準備顯存。這部分數(shù)字因模型而異實際占用需要以本機測試為準。不要只看總量要看并發(fā)數(shù)4 條并發(fā)任務同時跑顯存占用可能接近單任務的 4 倍。8. 常見問題與排查方法圖編排看起來簡單實際踩坑點不少。下面按常見程度列一份排查清單。問題現(xiàn)象可能原因排查方式解決方案圖編譯后不執(zhí)行任何節(jié)點START 節(jié)點沒有連接打印 graph.get_graph() 看拓撲結構檢查 START 到首節(jié)點的邊條件路由報錯 KeyError返回字符串不在映射字典里打印路由函數(shù)的返回值在路由函數(shù)里做歸一化限制返回值集合輸出缺少某個字段節(jié)點函數(shù)返回 dict 拼寫錯誤檢查節(jié)點返回的字段名給 State 所有字段補默認值節(jié)點重復執(zhí)行給同一個節(jié)點加了多條進入邊查看圖結構明確關鍵節(jié)點只有一條入口多余分支被跳過條件邊和目標節(jié)點沒連接檢查 add_conditional_edges 的映射補全分支映射并行分支結果互相覆蓋多個節(jié)點寫入同一字段檢查各節(jié)點返回字段使用不同的狀態(tài)字段API 調用超時外部接口響應慢檢查節(jié)點日志耗時拆成異步節(jié)點或增加重試批量任務跑到一半崩潰單條任務異常導致整個進程退出逐個任務捕獲異常用 future.exception 收集錯誤單獨重試失敗任務LLM 返回內容混亂沒有約束輸出格式打印原始返回值提示詞要求 JSON 輸出并做解析容錯還有一個隱蔽問題反復出現(xiàn)批量并發(fā)時狀態(tài)字段互相污染。雖然每次 invoke 都是獨立狀態(tài)但如果節(jié)點函數(shù)內部用了普通模塊級變量或類屬性保存臨時數(shù)據(jù)就可能串數(shù)據(jù)。節(jié)點函數(shù)要保持無狀態(tài)只從 state 讀取只往 state 寫入。9. 最佳實踐與使用建議把 SOP 工程化成 Graph 之后維護成本才是長期問題。下面這幾條建議來自工程經(jīng)驗建議直接落到團隊規(guī)范里。9.1 狀態(tài)字段要克制State 不是業(yè)務數(shù)據(jù)庫只保存節(jié)點之間需要傳遞的數(shù)據(jù)。字段越少越好寧可多拆幾個節(jié)點也不要把“順便可能用到的信息”全塞進去。9.2 路由函數(shù)必須收斂所有條件路由的返回值最好集中在有限集合里。不要用自由文本作為跳轉標記不要直接拿 LLM 的完整輸出做路由值。先解析、再映射、再跳轉三步拆開。9.3 每個節(jié)點都要有日志和失敗策略線上排查時最怕某個節(jié)點靜默失敗。給每個節(jié)點配上統(tǒng)一的運行日志輸出節(jié)點名、輸入摘要、耗時和返回結果摘要。失敗時區(qū)分三種策略重試一次、跳到兜底節(jié)點、直接終止人工處理。9.4 把人工審批做成節(jié)點而不是事后補合規(guī)要求高的流程建議在圖中顯式加入 human_review 節(jié)點。這個節(jié)點不調用 LLM而是發(fā)送消息給審批人收到人工確認后才進入下一步?？吹綀D的時候誰在哪個環(huán)節(jié)審批、如果不通過走哪條邊一目了然。9.5 首次上線跑小流量不管本地測試多順利第一次上線都應限制并發(fā)、控制流量。先讓少量真實任務跑一段時間觀察條件路由命中率、LLM 結果質量和外部接口穩(wěn)定性再逐步放開。批量任務建議從幾十條開始跑完檢查輸出再放量。9.6 素材和數(shù)據(jù)合規(guī)必須前置如果流程涉及用戶個人數(shù)據(jù)、內部業(yè)務數(shù)據(jù)、版權素材交付前必須確認數(shù)據(jù)權限。生成類 Agent 的輸出在面向用戶前最好再過一個內容合規(guī)檢查節(jié)點。10. 總結與下一步Graph Engineering 解決的核心問題不是“讓模型更聰明”而是“讓流程更可控”。Graph 讓 SOP 變成數(shù)據(jù)結構Agent 按圖執(zhí)行每一步都知道自己在哪兒、要去哪兒、失敗了怎么處理。這個思路在客服工單、內容審核、報告生成、批量數(shù)據(jù)處理這類強流程場景里落地速度非?？臁Ｗ钪档孟茸龅氖潞芎唵握乙粋€現(xiàn)有 SOP挑出 3 到 5 個關鍵節(jié)點把步驟畫成圖再按第 5 節(jié)的代碼模板跑通一次。先不要追求復雜并行先把順序流程和條件路由跑起來這是所有復雜編排的地基。最大的坑已經(jīng)強調過幾次條件路由的返回值必須受控State 字段必須精簡節(jié)點函數(shù)必須無狀態(tài)。這三個問題如果還踩說明圖還沒真正工程化。下一步可以嘗試的方向加 Checkpoint 實現(xiàn)故障恢復把并行分支改成真正的異步執(zhí)行把路由判斷從規(guī)則升級成 LLM 決策再接一個任務隊列管理批量任務。SOP 畫成圖不是結束畫完之后有一種穩(wěn)穩(wěn)的掌控感。