Command簡述

????????在 LangGraph 中，Command 是一個極具實用性的功能，它能夠將控制流（邊）和狀態更新（節點）巧妙地結合起來。這意味著開發者可以在同一個節點中，既執行狀態更新操作，又決定下一個要前往的節點，為工作流的構建帶來了極大的靈活性。

Command基本用法

Command?允許你在單個節點函數中完成兩件關鍵任務：

1. 更新圖狀態 (update): 修改共享的?State?對象。

2. 指定下一節點 (goto): 顯式決定工作流下一步執行哪個節點。

def my_node(state: State) -> Command[Literal["my_other_node"]]:?return Command(?# 狀態更新?update={"foo": "bar"},?# 控制流?goto="my_other_node"?)

????????在這個示例中，my_node 函數返回一個 Command 對象，

• update 參數用于指定狀態的更新內容，將狀態中的 "foo" 鍵值設為 "bar"；

• goto 參數則決定了下一個要前往的節點是 "my_other_node"。

關鍵約束

????????在節點函數中返回 Command 時，必須添加返回類型注釋，注明該節點可路由到的節點名稱列表，如 Command[Literal["my_other_node"]]。這一點至關重要，它不僅是 graph 渲染所必需的，還能明確告知 LangGraph 該節點可以導航到 "my_other_node"。

Command 核心應用場景

動態控制流（替代條件邊）

????????直接在節點邏輯中根據狀態判斷跳轉路徑，代碼更內聚：

def check_threshold(state: State) -> Command[Literal["proceed", "halt"]]:if state["value"] > state["threshold"]:return Command(goto="halt")  # 無需更新狀態時update可省略return Command(goto="proceed")

Command 與條件邊的決策

• Command當您需要同時更新圖形狀態和路由到其他節點時使用。例如，在實現多代理切換時，需要路由到其他代理并向該代理傳遞一些信息。

• 當僅需根據當前狀態選擇分支且不涉及狀態修改（如循環判斷、簡單路由），使用條件邊更清晰。

跨層級節點跳轉（子圖 → 父圖）

????????實現跨子圖的工作流跳轉，需指定?graph=Command.PARENT，代碼示例如下：

from typing import Literal, Optional, Annotated
from operator import addfrom langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from pydantic import BaseModel# 定義子圖狀態
class SubgraphState(BaseModel):"""子圖狀態定義"""issue: strescalated: bool = False  # 與父圖共享的狀態字段# 定義父圖狀態 reducer（處理子圖與父圖的狀態沖突）
def escalated_reducer(left: bool, right: bool) -> bool:"""共享字段escalated的合并策略：右側值（子圖更新）優先"""return right# 定義父圖狀態
class ParentGraphState(BaseModel):"""父圖狀態定義"""issue: strescalated: Annotated[bool, escalated_reducer] = False  # 使用reducer注解resolution: Optional[str] = None# 子圖節點
def subgraph_worker(state: SubgraphState) -> Command[Literal["manager_approval", END]]:"""子圖中的工作節點，決定是否升級到父圖"""if "critical" in state.issue.lower():return Command(update={"escalated": True},  # 更新共享狀態goto="manager_approval",  # 父圖中的目標節點graph=Command.PARENT  # 關鍵：指定跳轉到父圖)return Command(goto=END)  # 不升級則直接結束子圖# 構建子圖
subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node("worker", subgraph_worker)
subgraph_builder.add_edge(START, "worker")
subgraph = subgraph_builder.compile()# 父圖節點
def manager_approval_node(state: ParentGraphState) -> ParentGraphState:"""父圖中的經理審批節點"""return ParentGraphState(**state.model_dump(), resolution="經理已審批處理")# 構建父圖（包含子圖）
parent_builder = StateGraph(ParentGraphState)
parent_builder.add_node("subgraph", subgraph)  # 嵌入子圖
parent_builder.add_node("manager_approval", manager_approval_node)# 定義父圖邊
parent_builder.add_edge(START, "subgraph")
parent_builder.add_edge("manager_approval", END)# 編譯父圖
parent_graph = parent_builder.compile(checkpointer=MemorySaver()
)# 運行示例
if __name__ == "__main__":print("=== 跨層級跳轉測試 ===")# 測試會觸發升級的情況result1 = parent_graph.invoke({"issue": "Critical error: system down"})print(f"測試1 - 觸發升級: {result1}")# 應輸出包含escalated=True和經理審批結果的狀態# 測試不會觸發升級的情況result2 = parent_graph.invoke({"issue": "Minor issue: slow response"})print(f"測試2 - 不觸發升級: {result2}")# 應輸出escalated=False且無經理審批結果的狀態

• 狀態同步注意：若父子圖共享狀態字段，需在父圖狀態中為該字段定義?reducer?函數處理沖突。

工具調用與狀態注入

????????一個常見的工具調用場景是從工具內部更新圖譜狀態。例如，在客戶支持應用中，您可能希望在對話開始時根據客戶的賬號或 ID 查找客戶信息。要從工具中更新圖譜狀態，您可以從工具中返回?Command(update={"my_custom_key": "foo", "messages": [...]})。代碼示例如下：

from typing import Annotated, Dict, Any, List
from pydantic import BaseModel
from operator import addfrom langgraph.graph import StateGraph, START, END, Command
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, ToolMessage# 定義消息列表的reducer
def add_messages(left: List[Any], right: List[Any]) -> List[Any]:"""消息列表的合并策略"""return left + right# 1. 定義狀態模型
class SupportState(BaseModel):user_id: str  # 客戶IDuser_info: Annotated[Dict[str, Any], add] = {}  # 工具返回的用戶信息messages: Annotated[List[Any], add_messages] = []  # 消息歷史（必須包含工具調用記錄）query_status: str = "pending"  # 流程狀態標記# 2. 定義工具函數
@tool
def lookup_user_info(tool_call_id: Annotated[str, "tool_call_id"]) -> Command:"""根據用戶ID查詢客戶信息（內部工具）"""# 假設從線程配置中獲取user_id（實際中可從config獲取）# 注意：此處簡化處理，實際需注入configuser_id = "CUST-789"  # 模擬獲取print(f"正在查詢用戶信息: {user_id}")# 模擬實際查詢邏輯（可替換為數據庫/API調用）user_data = {"user_id": user_id,"name": "陳明亮","membership": "白金會員","contact": "chen@example.com","recent_tickets": "訂單延遲問題"}print(f"查詢結果: {user_data}")# 返回Command更新狀態return Command(update={"user_info": user_data,"messages": [ToolMessage(content=f"已獲取用戶 {user_id} 的詳細信息",tool_call_id=tool_call_id)],"query_status": "user_info_fetched"})# 3. 定義觸發工具調用的節點
def call_tool(state: SupportState) -> SupportState:"""創建工具調用消息的節點"""print(f"\n===== 觸發工具調用 =====")tool_call_message = AIMessage(content="",tool_calls=[{"name": "lookup_user_info","args": {},  # 無額外參數"id": "call_123"}])return SupportState(messages=state.messages + [tool_call_message])# 4. 定義后續處理節點
def handle_support_request(state: SupportState) -> SupportState:"""處理客戶請求的節點（依賴工具返回的用戶信息）"""print(f"\n===== 開始處理客戶請求 =====")print(f"客戶ID: {state.user_id}")print(f"客戶信息: {state.user_info}")print(f"當前狀態: {state.query_status}")print(f"消息記錄: {state.messages}")# 基于用戶信息進行業務處理if state.user_info.get("membership") == "白金會員":priority_msg = "優先處理白金會員請求"else:priority_msg = "標準流程處理請求"# 更新最終狀態return SupportState(user_id=state.user_id,user_info=state.user_info,messages=state.messages + [priority_msg, "客戶請求處理完成"],query_status="completed")# 5. 構建圖譜（使用ToolNode處理Command）
def build_support_graph():builder = StateGraph(SupportState)# 創建ToolNodetool_node = ToolNode([lookup_user_info])# 添加節點builder.add_node("call_tool", call_tool)builder.add_node("tools", tool_node)builder.add_node("process_request", handle_support_request)# 定義邊builder.add_edge(START, "call_tool")builder.add_edge("call_tool", "tools")builder.add_edge("tools", "process_request")builder.add_edge("process_request", END)return builder.compile(checkpointer=MemorySaver())# 6. 運行示例
if __name__ == "__main__":support_graph = build_support_graph()print("=== 客戶支持流程開始 ===")result = support_graph.invoke({"user_id": "CUST-789"})print("\n===== 流程結束 =====")print(f"最終用戶信息: {result['user_info']['name']} ({result['user_info']['membership']})")print(f"流程完成狀態: {result['query_status']}")print(f"完整消息歷史: {result['messages']}")

人機協同工作流（Human-in-the-Loop）

? ? ? ? Command還可以與?interrupt()?結合實現人工審核中斷與恢復：

from typing import TypedDict
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph
from langgraph.types import Command, interrupt# 新增模型相關導入
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
load_dotenv()# 定義狀態類型
class State(TypedDict):some_text: str# 初始化大模型（需要設置OPENAI_API_KEY環境變量）
model = ChatOpenAI(model="gpt-4o-mini")# 初始化檢查點存儲
checkpointer = MemorySaver()# 定義人類介入節點
def human_node(state: State):value = interrupt({"text_to_revise": state["some_text"],"instructions": "請修改以下文本："})return {"some_text": value}# 修改后的自動處理節點（調用大模型）
def process_text(state: State):# 構造模型請求message = model.invoke([HumanMessage(content=f"請處理以下請求：{state['some_text']}。保持回答簡潔。")])# 返回模型生成的文本return {"some_text": message.content}# 構建工作流
graph_builder = StateGraph(State)# 添加節點
graph_builder.add_node("human_review", human_node)
graph_builder.add_node("auto_process", process_text)# 設置流程
graph_builder.set_entry_point("auto_process")
graph_builder.add_edge("auto_process", "human_review")# 編譯圖表
graph = graph_builder.compile(checkpointer=checkpointer,interrupt_before=["human_review"]
)# 使用示例（保持不變）
if __name__ == "__main__":thread_id = "thread_123"thread_config = {"configurable": {"thread_id": thread_id}}initial_state = {"some_text": "輸出一個五五乘法表"}result = graph.invoke(initial_state, config=thread_config)print("自動處理結果:", result["some_text"])human_input = input("請輸入人類輸入：")resume_result = graph.invoke(Command(resume=human_input),config=thread_config)print("最終結果:", resume_result["some_text"])

最佳實踐與注意事項

1. 類型標注不可少：-> Command[Literal["node_a", "node_b"]]?是 LangGraph 靜態檢查和繪圖的基礎。

2. 跨圖狀態設計：子圖跳轉父圖更新共享狀態時，父圖需定義?reducer?處理沖突（如?lambda current, update: update）。

3. 命名一致性：goto?指定的節點名必須與圖中注冊的節點名完全一致。

4. 工具安全調用：在工具中使用?Command?時，確保狀態更新不會破壞圖的一致性。

參考文獻

Overview