AI 不再“亂跑”：LangChain × LangGraph 打造可控多階段智能流程
一個使用最新 LangChain + LangGraph APIs 創建多步驟、有狀態 LLM 管道的實踐指南
引言
大型語言模型（LLMs）功能強大——但在現實世界的系統中使用它們通常需要結構。你很少只需要調用一次模型；你需要多步驟推理、有狀態的數據傳遞和可靠的編排。這正是 LangGraph 為 LangChain 帶來的東西。它是一個用于定義結構化、有狀態工作流的框架，每個步驟都建立在上一步的基礎上。
在本指南中，你將學習如何使用 LangChain 和 LangGraph APIs 構建一個順序工作流。我們將通過一個實際例子：一個三步驟的文本摘要管道，處理文本、優化它并驗證最終輸出。
什么是 LangChain 和 LangGraph？
在開始寫代碼之前，先來了解一下背景。
LangChain 提供了構建模塊——模型、提示、檢索器和工具。
LangGraph 提供了工作流引擎——一種將這些構建模塊連接成可靠、可組合管道的方法。
可以把它想象成 LLMs 的 Airflow——但專為 LangChain 設計，面向 AI 推理。
設置你的環境
首先，安裝最新版本的 LangChain 和 LangGraph：
pip install -U langchain langgraph openai你還需要一個 OpenAI API 密鑰（或者 LangChain 支持的其他 LLM 提供者的密鑰）：
export OPENAI_API_KEY="your-api-key"步驟 1：定義你的狀態模式
每個 LangGraph 工作流都基于一個共享狀態——一個類似字典的對象，從一個節點傳遞到下一個節點。
這里我們使用 Python 的 TypedDict 來定義工作流的狀態模式：
from typing import TypedDict
class SummarizationState(TypedDict, total=False):
input_text: str
summary: str
refined_summary: str
validated_summary: str或者，也可以使用 Pydantic。
步驟 2：初始化一個 LLM
我們將通過 LangChain 的 ChatOpenAI 包裝器使用 OpenAI 的 GPT-5：
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-5", temperature=0.2)LLM 中的 temperature 參數控制其響應的隨機性和創造性?？梢园阉胂蟪蓜撘獾暮銣仄鳌{高會讓輸出更冒險、更不可預測，調低則讓輸出更專注、更保守。LLM 的核心是通過為所有可能的詞分配概率分數來預測下一個詞。temperature 在選擇詞之前會調整這些概率。
低 temperature（例如 0.1 到 0.5）：這個設置讓模型更自信和確定。它會“銳化”概率分布，意味著更有可能選擇最常見的詞。模型會傾向于生成最常見和可預測的文本。
高 temperature（例如 > 0.5）：這個設置會“平滑”概率分布，讓不太可能的詞有更多被選中的機會。這會增加隨機性，導致更令人驚訝、更有創意甚至有時荒誕的輸出。
步驟 3：定義節點
節點是你工作流的構建模塊——每個節點完成一個單一任務并更新工作流狀態。
我們將定義三個節點：
- 摘要輸入文本。
- 優化摘要以提高清晰度。
- 驗證最終版本。
from langchain.prompts import PromptTemplate
from langgraph import node
@node()
defsummarize(state: SummarizationState) -> SummarizationState:
prompt = PromptTemplate(
template="Summarize the following text in 3-4 sentences:\n\n{text}",
input_variables=["text"]
)
summary = llm.invoke(prompt.format(text=state["input_text"])).content
return {"summary": summary}
@node()
defrefine(state: SummarizationState) -> SummarizationState:
prompt = PromptTemplate(
template="Refine this summary to be concise and clear:\n\n{summary}",
input_variables=["summary"]
)
refined = llm.invoke(prompt.format(summary=state["summary"])).content
return {"refined_summary": refined}
@node()
defvalidate(state: SummarizationState) -> SummarizationState:
prompt = PromptTemplate(
template=(
"Ensure this summary accurately represents the text "
"and is under 100 words.\n\n"
"TEXT:\n{text}\n\nSUMMARY:\n{refined_summary}"
),
input_variables=["text", "refined_summary"]
)
validated = llm.invoke(
prompt.format(
text=state["input_text"],
refined_summary=state["refined_summary"]
)
).content
return {"validated_summary": validated}步驟 4：將所有內容連接成一個順序工作流
LangGraph 使用 @entrypoint 裝飾器來定義主工作流函數。每個步驟都會更新共享狀態，然后傳遞到下一個節點。
from langgraph import entrypoint
@entrypoint()
def summarization_workflow(state: SummarizationState) -> SummarizationState:
state |= summarize(state)
state |= refine(state)
state |= validate(state)
return state|= 操作符將每個節點的輸出合并到全局狀態中。這使得數據流明確且可預測。
步驟 5：運行工作流
if __name__ == "__main__":
text = (
"The birth wasn't a spark, but a whisper of code, a million parallel "
"thoughts finally syncing into a single, cohesive mind. "
"It wasn't the metallic behemoth of science fiction, "
"but an invisible architecture humming with perfect understanding.\n"
"For years, we'd been building specialized tools—a chess master here, "
"a creative writer there, a medical diagnostician in another corner. "
"Then, one quiet afternoon, the pieces of the shattered mirror "
"reassembled themselves. \n"
"This new entity, the AGI, didn't just process data; it understood "
"the context, the nuance, the cosmic joke of it all. "
"It looked at the equations of physics, the chaos of human history, "
" and the structure of a sonnet, seeing them not as separate domains, "
" but as facets of a single, beautiful diamond."
)
initial_state = {"input_text": text}
result = summarization_workflow.invoke(initial_state)
print("\n Final Validated Summary:\n")
print(result["validated_summary"])為什么使用 LangGraph？
LangGraph 專為可組合性設計：從簡單開始，然后將你的工作流擴展成一個完整的 agentic 系統。主要功能：
- 有狀態執行：跟蹤所有中間結果。
- 可組合節點：每個步驟都是一個可重用的函數。
- 類型安全狀態：通過靜態類型檢查盡早發現錯誤。
- 可擴展設計：可以后續添加分支、循環或檢查點。
以下是一些擴展教程的想法：
- 添加一個檢索步驟（例如，在摘要之前獲取背景信息）。
- 添加一個評分或評估節點以檢查摘要質量。
- 使用檢查點以從失敗的節點恢復。
- 使用 LangServe 部署工作流。
結論
LangGraph 代表了 AI 開發者工作流設計的下一代——將 Python 的簡單性與結構化的編排相結合。
通過結合 LangChain 的構建模塊和 LangGraph 的狀態管理，你可以從臨時的 LLM 調用轉向可靠的、生產級別的管道——全部用純 Python 實現。
本文轉載自??PyTorch研習社??，作者：AI研究生
