最近在嘗試構(gòu)建復雜的多智能體應用時你是否也遇到過這樣的困境多個Agent之間狀態(tài)流轉(zhuǎn)混亂、協(xié)作邏輯難以編排、循環(huán)和分支控制全靠硬編碼網(wǎng)上資料要么是簡單的LangChain調(diào)用要么是晦澀的學術(shù)論文真正能落地、可復現(xiàn)的LangGraph實戰(zhàn)教程少之又少。本文正是為了解決這個痛點。我將為你系統(tǒng)拆解LangGraph的核心架構(gòu)手把手帶你從零構(gòu)建一個功能完整的多智能體協(xié)作系統(tǒng)。內(nèi)容涵蓋從核心概念、環(huán)境搭建、State設(shè)計、Graph構(gòu)建到實現(xiàn)具備工具調(diào)用、條件路由、循環(huán)迭代等高級特性的完整項目。無論你是想快速上手LangGraph的新手還是希望將現(xiàn)有LangChain單體Agent升級為復雜工作流的開發(fā)者這篇文章都能提供一套閉環(huán)的實操方案。跟著做下來你不僅能理解多智能體架構(gòu)的精髓更能獲得一套可直接復用于自己項目的代碼模板。1. LangGraph與多智能體為什么是現(xiàn)在在深入代碼之前我們有必要厘清幾個核心概念理解LangGraph出現(xiàn)的背景及其要解決的根本問題。1.1 從LangChain到LangGraph架構(gòu)的演進LangChain是一個強大的框架它將大語言模型LLM與各種工具、數(shù)據(jù)源連接起來構(gòu)建出所謂的“鏈”Chains。你可以把它想象成一個流水線數(shù)據(jù)從一端輸入經(jīng)過一系列預設(shè)的、線性的處理節(jié)點LLM調(diào)用、工具執(zhí)行等最終從另一端輸出結(jié)果。這種模式非常適合確定性的、順序執(zhí)行的任務(wù)比如“讀取文件 - 總結(jié)內(nèi)容 - 翻譯成英文”。然而現(xiàn)實世界中的許多任務(wù)并非線性。它們更像一個流程圖或狀態(tài)機需要根據(jù)中間結(jié)果動態(tài)決定下一步走向甚至需要在多個“智能體”Agent之間進行循環(huán)、分支和協(xié)作。例如一個客服機器人先理解用戶意圖如果是查詢訂單則調(diào)用訂單查詢工具如果是投訴則轉(zhuǎn)接給人工坐席Agent并等待坐席反饋后再回復用戶。一個數(shù)據(jù)分析Agent先讓“規(guī)劃Agent”制定分析步驟再由“執(zhí)行Agent”調(diào)用Python工具進行計算最后讓“報告Agent”生成可視化圖表整個過程可能需要多次循環(huán)校驗。這就是LangGraph誕生的原因。它不是要取代LangChain而是構(gòu)建在LangChain之上的一個編排層。LangGraph將執(zhí)行過程抽象為一個有向圖其中節(jié)點代表執(zhí)行步驟可以是LLM、工具或函數(shù)邊代表控制流。它引入了關(guān)鍵的State概念來管理整個圖執(zhí)行過程中的共享數(shù)據(jù)從而優(yōu)雅地支持了循環(huán)、分支、并行和多個智能體之間的協(xié)作。簡單對比LangChain Chain 線性管道。輸入 - 節(jié)點A - 節(jié)點B - 輸出。LangGraph Graph 可循環(huán)、可分支的圖。開始 - 節(jié)點A - (條件判斷) - 分支B 或 分支C - (可能回到A) - 結(jié)束。1.2 多智能體系統(tǒng)的核心價值多智能體系統(tǒng)Multi-Agent System由多個具備一定自主性的智能體組成它們通過協(xié)作來解決單個智能體難以完成的復雜問題。其核心優(yōu)勢在于專業(yè)化分工不同Agent專注不同領(lǐng)域如搜索、計算、寫作效率更高。魯棒性一個Agent失敗其他Agent可以補位或調(diào)整策略?？蓴U展性可以方便地增加或修改特定功能的Agent而不影響整體架構(gòu)。LangGraph為實現(xiàn)這樣的系統(tǒng)提供了理想的底層框架。它讓每個Agent成為圖中的一個或多個節(jié)點并通過清晰定義的State和路由邏輯來管理它們之間的交互。1.3 核心概念速覽在開始實戰(zhàn)前快速理解以下LangGraph核心術(shù)語State 一個貫穿整個圖執(zhí)行過程的共享數(shù)據(jù)存儲通常是TypedDict或Pydantic BaseModel。它定義了圖中所有節(jié)點都能讀寫的數(shù)據(jù)結(jié)構(gòu)。Node 圖中的一個執(zhí)行單元。它是一個函數(shù)接收當前的State執(zhí)行操作如調(diào)用LLM、運行工具并返回一個更新后的State。Edge 連接節(jié)點的邊決定執(zhí)行流程的方向。分為普通邊固定流向和條件邊根據(jù)State內(nèi)容動態(tài)決定下一個節(jié)點。Graph 由節(jié)點和邊組成的整體結(jié)構(gòu)。Compilation 將定義好的Graph對象編譯成一個可執(zhí)行的、類似鏈Chain的對象。2. 環(huán)境準備與項目初始化我們將使用Python進行開發(fā)。請確保你的環(huán)境已準備就緒。2.1 環(huán)境與依賴操作系統(tǒng) Windows 10/11, macOS 或 Linux (Ubuntu 20.04) 均可。Python版本 建議使用 Python 3.10 或 3.11兼容性最好。首先創(chuàng)建一個新的項目目錄并安裝必要的包。我們主要需要langgraph,langchain以及OpenAI的SDK這里以O(shè)penAI為例你也可以使用其他兼容的LLM如Anthropic、Ollama本地模型等。# 創(chuàng)建項目目錄 mkdir langgraph-multi-agent-tutorial cd langgraph-multi-agent-tutorial # 創(chuàng)建虛擬環(huán)境可選但推薦 python -m venv venv # Windows激活: venv\Scripts\activate # Mac/Linux激活: source venv/bin/activate # 安裝核心依賴 pip install langgraph langchain langchain-openai # 安裝可能用到的額外工具包用于示例中的工具調(diào)用 pip install wikipedia requests duckduckgo-search重要版本說明langgraph和langchain生態(tài)迭代較快本文代碼基于langgraph0.0.40,langchain0.1.0版本編寫。如果遇到API變更請參考官方文檔調(diào)整。本文重點在于傳授架構(gòu)思想和實現(xiàn)模式具體API調(diào)用方式可隨版本更新。2.2 設(shè)置API密鑰為了調(diào)用OpenAI的模型你需要設(shè)置API密鑰。切勿將密鑰硬編碼在代碼中提交到版本庫。# 在命令行中設(shè)置環(huán)境變量臨時 # Windows (cmd): setx OPENAI_API_KEY your-api-key-here # Windows (PowerShell): $env:OPENAI_API_KEYyour-api-key-here # Mac/Linux: export OPENAI_API_KEYyour-api-key-here或者在代碼中通過os.environ設(shè)置僅用于開發(fā)測試import os os.environ[OPENAI_API_KEY] your-api-key-here3. LangGraph核心組件深度解析理解State、Node、Edge是構(gòu)建任何LangGraph應用的基礎(chǔ)。讓我們逐一拆解。3.1 State設(shè)計系統(tǒng)的共享內(nèi)存State是整個圖的“中樞神經(jīng)系統(tǒng)”所有節(jié)點都通過讀寫State來通信。設(shè)計一個好的State是成功的第一步。State通常是一個Pydantic模型它強制定義了數(shù)據(jù)類型有利于開發(fā)和調(diào)試。from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages from typing_extensions import TypedDict import operator # 方法一使用TypedDict簡單場景 class AgentState(TypedDict): # 消息歷史使用langgraph的注解實現(xiàn)自動累加 messages: Annotated[List[str], add_messages] # 當前查詢或任務(wù) query: str # 由哪個Agent處理 current_agent: str # 工具執(zhí)行的結(jié)果 tool_output: str # 是否需要繼續(xù)循環(huán) should_continue: bool # 方法二使用Pydantic BaseModel推薦功能更強大 from pydantic import BaseModel, Field from typing import Optional class PydanticAgentState(BaseModel): 使用Pydantic定義的State支持更復雜的驗證和默認值。 messages: Annotated[List[dict], add_messages] Field(default_factorylist) user_query: str current_agent: str planner tool_output: Optional[str] None should_continue: bool True # 可以添加任意其他字段 intermediate_steps: List[tuple] Field(default_factorylist) class Config: arbitrary_types_allowed True關(guān)鍵點Annotated[List[...], add_messages] 這是一個神奇的注解。它告訴LangGraph當多個節(jié)點返回的State中都包含messages字段時不要覆蓋而是自動追加append。這對于維護對話歷史至關(guān)重要。其他字段如query,current_agent默認是覆蓋語義。你可以根據(jù)業(yè)務(wù)需求自由添加任何字段。3.2 Node智能體與工具的執(zhí)行單元Node是一個普通的Python函數(shù)或可調(diào)用對象它接收一個State字典返回一個更新后的State字典。from langchain_openai import ChatOpenAI # 初始化LLM llm ChatOpenAI(modelgpt-3.5-turbo) def planner_node(state: PydanticAgentState) - dict: 規(guī)劃節(jié)點分析用戶查詢決定由哪個Agent處理。 print(f[Planner Node] 正在規(guī)劃查詢: {state.user_query}) # 構(gòu)建系統(tǒng)提示詞 system_prompt 你是一個任務(wù)規(guī)劃員。請分析用戶請求并決定由哪個專家處理 - 如果問題是關(guān)于事實、定義或最新事件的交給 researcher。 - 如果問題是需要計算、推理或分析的交給 calculator。 - 如果問題是創(chuàng)意寫作、總結(jié)或翻譯交給 writer。 - 如果問題無法歸類或需要綜合處理交給 generalist。 只輸出Agent的名字不要有其他內(nèi)容。 # 調(diào)用LLM response llm.invoke([ {role: system, content: system_prompt}, {role: user, content: state.user_query} ]) next_agent response.content.strip().lower() print(f[Planner Node] 決定交由 {next_agent} 處理。) # 更新State并返回 return {current_agent: next_agent, messages: [{role: assistant, content: f任務(wù)已分配給 {next_agent}。}]} def researcher_node(state: PydanticAgentState) - dict: 研究節(jié)點使用搜索工具查找信息。 print(f[Researcher Node] 正在研究: {state.user_query}) # 這里可以集成真實的搜索工具如SerpAPI、DuckDuckGo # 為示例我們模擬一個結(jié)果 tool_result f根據(jù)網(wǎng)絡(luò)搜索關(guān)于{state.user_query}的信息是LangGraph是一個用于構(gòu)建多智能體應用的庫。 return { tool_output: tool_result, messages: [{role: assistant, content: f研究完成{tool_result}}] }3.3 Edge與路由控制流程的走向Edge決定了執(zhí)行完一個Node后下一步該去哪個Node。LangGraph提供了兩種主要的路由方式3.3.1 固定路由最簡單的方式從一個節(jié)點固定地指向下一個節(jié)點。from langgraph.graph import StateGraph, END # 創(chuàng)建圖 builder StateGraph(PydanticAgentState) # 添加節(jié)點 builder.add_node(planner, planner_node) builder.add_node(researcher, researcher_node) builder.add_node(calculator, calculator_node) # 假設(shè)已定義 builder.add_node(writer, writer_node) # 假設(shè)已定義 # 設(shè)置入口點 builder.set_entry_point(planner) # 添加固定邊從planner出來后根據(jù)它決定的agent路由到不同節(jié)點 # 注意這里還不是條件邊只是演示固定連接 builder.add_edge(planner, researcher) # 這只是個例子實際需要條件邏輯3.3.2 條件路由Conditional Edge這是實現(xiàn)分支和循環(huán)的關(guān)鍵。你需要定義一個路由函數(shù)根據(jù)State的內(nèi)容返回下一個節(jié)點的名稱。def route_after_planner(state: PydanticAgentState) - str: 規(guī)劃節(jié)點之后的路由根據(jù)current_agent字段決定下一個節(jié)點。 next_agent state.current_agent if next_agent in [researcher, calculator, writer, generalist]: return next_agent # 返回節(jié)點名稱 else: # 如果無法識別默認結(jié)束 return end def should_continue(state: PydanticAgentState) - str: 判斷是否繼續(xù)循環(huán)。 if state.should_continue: return continue_loop else: return END # END是LangGraph預定義的終止標識在構(gòu)建圖時使用add_conditional_edges來添加條件邊。# 添加條件邊從planner節(jié)點出發(fā)根據(jù)route_after_planner函數(shù)的返回值決定下一個節(jié)點 builder.add_conditional_edges( planner, route_after_planner, # 路由函數(shù) # 可選映射路由函數(shù)返回值到節(jié)點名如果函數(shù)直接返回節(jié)點名則不需要 # 但我們的函數(shù)已直接返回節(jié)點名所以這里不需要映射 )4. 完整實戰(zhàn)構(gòu)建一個多智能體協(xié)作系統(tǒng)現(xiàn)在我們將把所有概念組合起來構(gòu)建一個具備完整工作流的系統(tǒng)。這個系統(tǒng)包含規(guī)劃員(Planner)、研究員(Researcher)、作家(Writer)和一個通用助手(Generalist)它們協(xié)作回答用戶問題。4.1 項目結(jié)構(gòu)langgraph-multi-agent-tutorial/ ├── agents/ # 智能體模塊 │ ├── __init__.py │ ├── planner.py │ ├── researcher.py │ ├── writer.py │ └── generalist.py ├── tools/ # 工具模塊 │ ├── __init__.py │ └── search_tool.py ├── state.py # State定義 ├── graph_builder.py # 圖構(gòu)建邏輯 ├── app.py # 主應用入口 └── requirements.txt4.2 定義State與工具state.py:from typing import List, Optional, Annotated from pydantic import BaseModel, Field from langgraph.graph.message import add_messages class MultiAgentState(BaseModel): 多智能體系統(tǒng)的共享狀態(tài)。 # 對話消息歷史自動追加 messages: Annotated[List[dict], add_messages] Field(default_factorylist) # 原始用戶輸入 user_input: str # 當前活躍的智能體名稱 current_agent: str planner # 上一個工具調(diào)用的輸出 last_tool_output: Optional[str] None # 控制流程的標志 needs_research: bool False needs_writing: bool False # 最終答案 final_answer: Optional[str] None # 迭代次數(shù)防止無限循環(huán) iteration_count: int 0 class Config: arbitrary_types_allowed Truetools/search_tool.py:import requests from duckduckgo_search import DDGS class SearchTool: 一個簡單的搜索工具示例。 staticmethod def search_web(query: str, max_results: int 3) - str: 使用DuckDuckGo搜索網(wǎng)絡(luò)。 try: with DDGS() as ddgs: results [] for r in ddgs.text(query, max_resultsmax_results): results.append(f- {r[title]}: {r[body]}) if results: return \n.join(results) else: return 未找到相關(guān)信息。 except Exception as e: return f搜索過程中出錯{e} staticmethod def mock_search(query: str) - str: 模擬搜索用于離線測試。 # 在實際項目中替換為真實的API調(diào)用 mock_data { LangGraph: LangGraph是LangChain生態(tài)系統(tǒng)中的一個庫用于構(gòu)建具有循環(huán)、分支和多智能體協(xié)作的復雜應用。它使用圖結(jié)構(gòu)來編排狀態(tài)和流程。, 多智能體: 多智能體系統(tǒng)由多個自治或半自治的智能體組成通過協(xié)作解決超出單個智能體能力范圍的復雜問題。, Python教程: Python是一種高級、解釋型的通用編程語言以其清晰的語法和強大的庫生態(tài)系統(tǒng)而聞名。 } return mock_data.get(query, f關(guān)于{query}找到一些相關(guān)信息這是一個熱門技術(shù)話題。)4.3 實現(xiàn)各個智能體節(jié)點agents/planner.py:from langchain_openai import ChatOpenAI from ..state import MultiAgentState llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) def planner_node(state: MultiAgentState) - dict: 規(guī)劃智能體分析用戶輸入制定處理計劃。 print(f\n 規(guī)劃員開始工作 ) prompt f 你是一個資深任務(wù)規(guī)劃員。請分析以下用戶問題并制定處理計劃 用戶問題{state.user_input} 請判斷 1. 這個問題是否需要查詢最新信息或事實如果需要設(shè)置 needs_research: true。 2. 這個問題是否需要優(yōu)美的文字總結(jié)、潤色或創(chuàng)作如果需要設(shè)置 needs_writing: true。 3. 根據(jù)以上判斷決定下一個執(zhí)行的智能體 - 如果需要研究交給 researcher。 - 如果需要寫作且不需要研究或研究已完成交給 writer。 - 如果既不需要研究也不需要寫作或者問題很簡單交給 generalist。 請用以下JSON格式回復僅返回JSON {{ needs_research: true/false, needs_writing: true/false, next_agent: researcher or writer or generalist, reasoning: 簡要推理過程 }} response llm.invoke([{role: user, content: prompt}]) import json try: plan json.loads(response.content) print(f規(guī)劃結(jié)果{plan}) return { needs_research: plan[needs_research], needs_writing: plan[needs_writing], current_agent: plan[next_agent], messages: [{role: assistant, content: f規(guī)劃員分析{plan[reasoning]}}] } except json.JSONDecodeError: print(規(guī)劃員返回了非JSON格式使用默認路由。) return { current_agent: generalist, messages: [{role: assistant, content: 規(guī)劃完成交由通用助手處理。}] }agents/researcher.py:from langchain_openai import ChatOpenAI from ..tools.search_tool import SearchTool from ..state import MultiAgentState llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) def researcher_node(state: MultiAgentState) - dict: 研究智能體執(zhí)行搜索并提煉信息。 print(f\n 研究員開始工作 ) # 使用工具進行搜索 search_query state.user_input # 在實際應用中可以讓LLM生成更優(yōu)的搜索關(guān)鍵詞 # raw_results SearchTool.search_web(search_query) raw_results SearchTool.mock_search(search_query) # 使用模擬工具 # 提煉搜索結(jié)果 refine_prompt f 以下是根據(jù)查詢“{search_query}”搜索到的原始信息 {raw_results} 請?zhí)釤挸雠c用戶問題最相關(guān)、最準確的信息形成一段簡潔、客觀的總結(jié)。 用戶原始問題是{state.user_input} 直接輸出總結(jié)內(nèi)容不要提及“根據(jù)搜索”等前綴。 refined_info llm.invoke([{role: user, content: refine_prompt}]).content print(f研究完成提煉信息{refined_info[:100]}...) return { last_tool_output: refined_info, needs_research: False, # 研究完成標記為False messages: [{role: assistant, content: f研究員已獲取信息{refined_info[:150]}...}] }agents/writer.py:from langchain_openai import ChatOpenAI from ..state import MultiAgentState llm ChatOpenAI(modelgpt-4, temperature0.7) # 寫作可以使用更有創(chuàng)造力的模型 def writer_node(state: MultiAgentState) - dict: 寫作智能體潤色、總結(jié)或創(chuàng)作文本。 print(f\n 作家開始工作 ) # 獲取上下文可能是原始輸入也可能是研究員的結(jié)果 source_material state.last_tool_output if state.last_tool_output else state.user_input prompt f 你是一位專業(yè)的文本寫作助手。請根據(jù)以下材料生成一個對用戶友好、結(jié)構(gòu)清晰、語言優(yōu)美的回答。 用戶原始問題{state.user_input} 參考材料 {source_material} 請生成最終答案。如果材料不足請基于你的知識進行補充并注明哪些是已知信息。 final_answer llm.invoke([{role: user, content: prompt}]).content print(f作家完成創(chuàng)作。) return { final_answer: final_answer, needs_writing: False, messages: [{role: assistant, content: f作家已生成回答。}] }agents/generalist.py:from langchain_openai import ChatOpenAI from ..state import MultiAgentState llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.3) def generalist_node(state: MultiAgentState) - dict: 通用智能體處理簡單或直接的問題。 print(f\n 通用助手開始工作 ) prompt f 請直接回答用戶的問題。如果問題需要事實核查而你不確定請說明這一點。 用戶問題{state.user_input} 請?zhí)峁蚀_、有幫助的回答。 answer llm.invoke([{role: user, content: prompt}]).content return { final_answer: answer, messages: [{role: assistant, content: f通用助手生成回答。}] }4.4 構(gòu)建圖與路由邏輯graph_builder.py:from langgraph.graph import StateGraph, END from .state import MultiAgentState from .agents.planner import planner_node from .agents.researcher import researcher_node from .agents.writer import writer_node from .agents.generalist import generalist_node def route_after_planner(state: MultiAgentState) - str: 規(guī)劃員之后的路由邏輯。 # 防止無限循環(huán) if state.iteration_count 5: return end next_agent state.current_agent print(f路由決策下一個智能體是 {next_agent}) return next_agent def route_after_researcher(state: MultiAgentState) - str: 研究員之后的路由邏輯。 if state.needs_writing: return writer else: # 研究完成不需要寫作直接生成最終答案由通用助手整理 return generalist def route_after_writer(state: MultiAgentState) - str: 作家之后的路由邏輯任務(wù)完成。 return end def route_after_generalist(state: MultiAgentState) - str: 通用助手之后的路由邏輯任務(wù)完成。 return end def build_multi_agent_graph() - StateGraph: 構(gòu)建并返回多智能體圖。 # 1. 創(chuàng)建圖構(gòu)建器指定State類型 builder StateGraph(MultiAgentState) # 2. 添加所有節(jié)點 builder.add_node(planner, planner_node) builder.add_node(researcher, researcher_node) builder.add_node(writer, writer_node) builder.add_node(generalist, generalist_node) # 3. 設(shè)置入口點 builder.set_entry_point(planner) # 4. 添加條件邊實現(xiàn)分支邏輯 # 從planner出發(fā)根據(jù)它設(shè)定的current_agent路由 builder.add_conditional_edges( planner, route_after_planner, # route_after_planner 函數(shù)返回的字符串直接對應節(jié)點名 ) # 從researcher出發(fā)根據(jù)needs_writing標志路由 builder.add_conditional_edges( researcher, route_after_researcher, ) # 從writer出發(fā)固定結(jié)束 builder.add_edge(writer, END) # 從generalist出發(fā)固定結(jié)束 builder.add_edge(generalist, END) # 5. 編譯圖 graph builder.compile() # 可選保存圖的可視化 try: from IPython.display import Image, display # 這需要graphviz僅用于演示 # display(Image(graph.get_graph().draw_mermaid_png())) except: pass print(多智能體圖構(gòu)建完成) return graph4.5 主程序與運行示例app.py:from graph_builder import build_multi_agent_graph from state import MultiAgentState def main(): # 1. 構(gòu)建圖 print(正在初始化多智能體系統(tǒng)...) graph build_multi_agent_graph() # 2. 定義初始狀態(tài) user_query input(請輸入您的問題或按回車使用默認問題).strip() if not user_query: user_query 請用通俗易懂的方式解釋什么是LangGraph和多智能體系統(tǒng) print(f使用默認問題{user_query}) initial_state MultiAgentState( user_inputuser_query, messages[{role: user, content: user_query}], iteration_count0 ) # 3. 執(zhí)行圖 print(\n *50) print(開始執(zhí)行多智能體協(xié)作流程...) print(*50) final_state None # 我們也可以使用stream模式來觀察每一步這里使用invoke一次性執(zhí)行 for step in graph.stream(initial_state, stream_modevalues): node_name list(step.keys())[0] state step[node_name] print(f\n[步驟] 節(jié)點 {node_name} 執(zhí)行完畢。) print(f 當前負責Agent: {state.get(current_agent, N/A)}) print(f 迭代計數(shù): {state.get(iteration_count, 0)}) if state.get(final_answer): print(f 最終答案已生成) final_state state # 4. 輸出結(jié)果 print(\n *50) print(執(zhí)行完成) print(*50) if final_state and final_state.get(final_answer): print(\n【最終答案】) print(final_state[final_answer]) else: print(\n未生成最終答案。) if final_state: print(最后的狀態(tài), final_state) # 5. 打印執(zhí)行軌跡消息歷史 print(\n【執(zhí)行軌跡】) for msg in final_state.get(messages, [])[-5:]: # 顯示最后幾條消息 role msg.get(role, unknown).upper() content_preview msg.get(content, )[:80] print(f {role}: {content_preview}...) if __name__ __main__: main()4.6 運行與驗證在項目根目錄下運行python app.py示例交互輸出正在初始化多智能體系統(tǒng)... 多智能體圖構(gòu)建完成 請輸入您的問題或按回車使用默認問題什么是Transformer架構(gòu) 開始執(zhí)行多智能體協(xié)作流程... 規(guī)劃員開始工作 規(guī)劃結(jié)果{needs_research: True, needs_writing: True, next_agent: researcher, reasoning: 用戶詢問的是“Transformer架構(gòu)”這是一個具體的、基于事實的技術(shù)概念可能需要查詢最新或最準確的解釋。同時為了讓解釋通俗易懂可能需要寫作潤色。} [步驟] 節(jié)點 planner 執(zhí)行完畢。 當前負責Agent: researcher 迭代計數(shù): 0 研究員開始工作 研究完成提煉信息Transformer架構(gòu)是一種用于自然語言處理任務(wù)的深度學習模型架構(gòu)... [步驟] 節(jié)點 researcher 執(zhí)行完畢。 當前負責Agent: researcher 迭代計數(shù): 0 作家開始工作 作家完成創(chuàng)作。 [步驟] 節(jié)點 writer 執(zhí)行完畢。 當前負責Agent: writer 迭代計數(shù): 0 執(zhí)行完成 【最終答案】 Transformer架構(gòu)是谷歌在2017年提出的、徹底改變自然語言處理領(lǐng)域的深度學習模型核心設(shè)計... 【執(zhí)行軌跡】 ASSISTANT: 規(guī)劃員分析用戶詢問的是“Transformer架構(gòu)”... ASSISTANT: 研究員已獲取信息Transformer架構(gòu)是一種用于自然語言處理任務(wù)的深度學習模型架構(gòu)... ASSISTANT: 作家已生成回答。5. 常見問題與排查思路在開發(fā)LangGraph多智能體應用時你可能會遇到以下典型問題。問題現(xiàn)象可能原因排查思路與解決方案KeyError或 State字段不存在1. State的Pydantic模型字段名與節(jié)點返回的字典鍵不匹配。2. 使用了未在State中定義的字段。1. 仔細檢查State類的字段定義和每個node函數(shù)返回的字典鍵。2. 確保所有節(jié)點讀寫State的字段都已聲明。使用Pydantic能提前發(fā)現(xiàn)很多此類錯誤。圖陷入無限循環(huán)1. 條件路由邏輯有誤導致節(jié)點間形成閉環(huán)。2. 沒有設(shè)置循環(huán)終止條件。1. 在State中添加iteration_count字段并在路由函數(shù)中檢查超過閾值則導向END。2. 使用langgraph的checkpointer或interrupt機制來手動控制循環(huán)。打印每個節(jié)點的執(zhí)行日志觀察循環(huán)路徑。add_messages注解不工作1. 未正確導入Annotated和add_messages。2.messages字段類型不是List[dict]或List[BaseMessage]。1. 確保導入from typing import Annotated和from langgraph.graph.message import add_messages。2. 確保字段定義為messages: Annotated[List[dict], add_messages]。消息格式應為{role: ..., content: ...}。條件路由函數(shù)返回的節(jié)點名未找到路由函數(shù)返回的字符串在圖中的節(jié)點集合里不存在。1. 使用builder.add_node()添加節(jié)點時使用的名稱必須與路由函數(shù)返回的名稱完全一致區(qū)分大小寫。2. 在路由函數(shù)中添加print語句或使用調(diào)試器檢查其返回值。確保返回值是END或已添加的節(jié)點名。LLM調(diào)用超時或報錯1. API密鑰未設(shè)置或錯誤。2. 網(wǎng)絡(luò)問題。3. 模型名稱錯誤或額度不足。1. 檢查OPENAI_API_KEY環(huán)境變量。2. 添加超時和重試邏輯。例如ChatOpenAI(..., request_timeout30, max_retries2)。3. 確認模型名稱如gpt-3.5-turbo可用。工具調(diào)用失敗1. 工具函數(shù)本身有bug。2. 工具返回的結(jié)果格式不符合下游LLM期望。1. 單獨測試工具函數(shù)。2. 在工具調(diào)用前后添加日志打印輸入和輸出。3. 確保工具返回的是字符串或可序列化為字符串的類型。編譯圖時報類型錯誤State的Pydantic模型與節(jié)點函數(shù)簽名不兼容。1. 確保所有節(jié)點函數(shù)都接受State類型或?qū)淖值渥鳛閰?shù)并返回一個字典。2. 如果使用Pydantic節(jié)點函數(shù)參數(shù)類型注解應為(state: YourStateModel) - dict。6. 最佳實踐與進階建議掌握了基礎(chǔ)構(gòu)建后遵循以下最佳實踐能讓你的多智能體系統(tǒng)更健壯、更易維護。6.1 State設(shè)計原則最小化與清晰化State只存儲流程中真正需要共享的數(shù)據(jù)。避免將臨時變量或大量無關(guān)數(shù)據(jù)放入State。使用Pydantic強烈推薦使用Pydantic的BaseModel來定義State。它能提供類型檢查、數(shù)據(jù)驗證、自動文檔和更好的IDE支持。區(qū)分可變與不可變利用Annotated和add_messages來處理需要追加的列表如消息歷史。對于需要覆蓋的簡單字段使用普通類型。包含流程控制標志如should_continue,needs_approval等這些是驅(qū)動條件路由的關(guān)鍵。6.2 節(jié)點設(shè)計原則單一職責每個節(jié)點應只做一件事并把它做好。例如一個節(jié)點負責調(diào)用LLM另一個節(jié)點負責處理工具結(jié)果。冪等性盡可能讓節(jié)點函數(shù)是冪等的相同輸入產(chǎn)生相同輸出。這有助于調(diào)試和保證流程確定性。充分的日志在每個節(jié)點的開始和結(jié)束處打印關(guān)鍵信息如print(f[NodeName] Processing...)這對于跟蹤復雜的圖執(zhí)行流程至關(guān)重要。錯誤處理在節(jié)點內(nèi)部使用try...except捕獲預期錯誤并更新State以反映錯誤狀態(tài)讓路由邏輯能處理失敗情況。6.3 圖結(jié)構(gòu)與路由優(yōu)化可視化你的圖在開發(fā)初期使用graph.get_graph().draw_mermaid_png()需安裝graphviz生成流程圖。這能幫你直觀理解控制流發(fā)現(xiàn)設(shè)計缺陷。簡化路由邏輯路由函數(shù)應盡可能簡單只做判斷。復雜的決策邏輯最好放在前驅(qū)節(jié)點中將結(jié)果存入State再由路由函數(shù)讀取。使用預編譯的子圖對于復雜的、可復用的流程模塊可以將其構(gòu)建為獨立的子圖StateGraph然后通過add_node將其作為一個大節(jié)點加入主圖。這能極大提升代碼的模塊化和可讀性。實現(xiàn)持久化與檢查點對于長時運行的工作流利用LangGraph的Checkpointer功能保存執(zhí)行狀態(tài)。這樣在應用重啟后可以從中斷處繼續(xù)這對生產(chǎn)環(huán)境至關(guān)重要。6.4 生產(chǎn)環(huán)境部署考量配置管理將LLM模型名稱、API密鑰、溫度等參數(shù)外置到配置文件如.env或config.yaml中不要硬編碼。異步支持如果節(jié)點涉及大量I/O操作如網(wǎng)絡(luò)請求、數(shù)據(jù)庫查詢考慮使用異步節(jié)點async def和AsyncGraph來提升并發(fā)性能。監(jiān)控與可觀測性為你的圖集成監(jiān)控?？梢杂涗浢總€節(jié)點的執(zhí)行時間、輸入輸出注意脫敏、以及State的演變過程。這對于排查問題和優(yōu)化性能很有幫助。版本控制圖的定義節(jié)點、邊、State應納入代碼版本控制。當業(yè)務(wù)邏輯變更時可以通過版本回滾。6.5 擴展方向集成外部工具將更多的真實工具數(shù)據(jù)庫查詢API、內(nèi)部系統(tǒng)API、代碼執(zhí)行器封裝成LangChain Tool供智能體調(diào)用。實現(xiàn)人工審核節(jié)點在關(guān)鍵節(jié)點如最終答案發(fā)布前引入human-in-the-loop等待人工確認后再繼續(xù)。動態(tài)圖構(gòu)建根據(jù)運行時的State動態(tài)地添加或移除圖中的節(jié)點實現(xiàn)更靈活的工作流。多圖協(xié)作構(gòu)建多個專門的圖如“研究圖”、“寫作圖”、“審核圖”并通過一個主圖來協(xié)調(diào)它們實現(xiàn)更清晰的系統(tǒng)架構(gòu)。多智能體系統(tǒng)的構(gòu)建是一個迭代過程。從一個小而精的圖開始驗證核心流程然后逐步增加節(jié)點、優(yōu)化路由、完善State。利用好LangGraph提供的可視化、調(diào)試和持久化工具能讓你在開發(fā)復雜工作流時事半功倍。