LangGraph 是一個用於構建可控 Agent 的低級編排框架。LangChain 提供整合和可組合元件來簡化LLM應用程式開發，而 LangGraph 庫則支援代理編排，提供可定製的架構、長期記憶體和人機協同功能，以可靠地處理複雜任務。 https://datasciencedojo.com/blog/langgraph-tutorial/ 快速建立Agent 首先，安裝 LangGraph： pip install -U langgraph 使用預構建的元件： 快速可靠地構建代理系統，而無需從頭開始實施編排、記憶體或人工反饋處理。 要使用特定參數（例如溫度）進行配置LLM，請使用 init_chat_model： #建構一個基於 Claude的簡單Agent案例#bashpip install -U langgraph "langchain[anthropic]"from langgraph.prebuilt import create_react_agent#先定義一個模擬的天氣資料抓取工具給Agent抓def get_weather(city: str) -> str:      """Get weather for a given city."""    return f"It's always sunny in {city}!"#Agent根據指定模型, 可抓工具, 你的指令去做事agent = create_react_agent(    model="anthropic:claude-3-7-sonnet-latest",      tools=[get_weather],      prompt="You are a helpful assistant"  )# Run the agentagent.invoke(    {"messages": [{"role": "user", "content": "what is the weather in sf"}]}) from langchain.chat_models import init_chat_modelfrom langgraph.prebuilt import create_react_agentmodel = init_chat_model(    "anthropic:claude-3-7-sonnet-latest",    temperature=0)agent = create_react_agent(    model=model,    tools=[get_weather],) 提示指示LLM如何作。新增以下類型的提示之一： Static：字串被解釋為系統消息。 Dynamic：在運行時根據輸入或配置生成的消息清單。 #靜態from langgraph.prebuilt import create_react_agentagent = create_react_agent(    model="anthropic:claude-3-7-sonnet-latest",    tools=[get_weather],    # A static prompt that never changes    prompt="Never answer questions about the weather.")agent.invoke(    {"messages": [{"role": "user", "content": "what is the weather in sf"}]}) #動態from langchain_core.messages import AnyMessagefrom langchain_core.runnables import RunnableConfigfrom langgraph.prebuilt.chat_agent_executor import AgentStatefrom langgraph.prebuilt import create_react_agentdef prompt(state: AgentState, config: RunnableConfig) -> list[AnyMessage]:      user_name = config["configurable"].get("user_name")    system_msg = f"You are a helpful assistant. Address the user as {user_name}."    return [{"role": "system", "content": system_msg}] + state["messages"]agent = create_react_agent(    model="anthropic:claude-3-7-sonnet-latest",    tools=[get_weather],    prompt=prompt)agent.invoke(    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},    config={"configurable": {"user_name": "John Smith"}}) 要允許與代理進行多輪次對話，您需要在創建代理時提供檢查點程式來啟用持久性。在運行時，您需要提供一個包含 thread_id 的配置——對話（會話）的唯一標識碼： from langgraph.prebuilt import create_react_agentfrom langgraph.checkpoint.memory import InMemorySavercheckpointer = InMemorySaver()agent = create_react_agent(    model="anthropic:claude-3-7-sonnet-latest",    tools=[get_weather],    checkpointer=checkpointer  )# Run the agentconfig = {"configurable": {"thread_id": "1"}}sf_response = agent.invoke(    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},    config  )ny_response = agent.invoke(    {"messages": [{"role": "user", "content": "what about new york?"}]},    config) 事實上，記憶非常重要。非常建議把記憶部分全部參透。 要生成符合架構的結構化回應，請使用 response_format 參數。可以使用 Pydantic 模型或 TypedDict 定義模式。結果可通過 structured_response 欄位訪問。 from pydantic import BaseModelfrom langgraph.prebuilt import create_react_agentclass WeatherResponse(BaseModel):    conditions: stragent = create_react_agent(    model="anthropic:claude-3-7-sonnet-latest",    tools=[get_weather],    response_format=WeatherResponse  )response = agent.invoke(    {"messages": [{"role": "user", "content": "what is the weather in sf"}]})response["structured_response"] 使用 LangGraph： 自定義您的架構，使用長期記憶體，並實施人機協同功能，以可靠地處理複雜任務。 LangGraph 提供位於任何工作流或代理下的低級支援基礎設施。它不抽象提示或架構，並提供三個主要好處： 1.持久化 記憶體：LangGraph 持久化應用程式狀態的任意方面，支援使用者交互內部和之間的對話和其他更新的記憶體; 人機協同：由於狀態是檢查點，因此可以中斷和恢復執行，從而允許通過人工輸入進行決策、驗證和更正。 2.Streaming 2.流媒體 LangGraph 在執行過程中為使用者（或開發人員）提供對工作流/代理狀態的支援。LangGraph 支援對事件（例如來自工具調用的反饋）和來自應用程式中嵌入的調用的LLM令牌進行流式處理。 3.Debugging and deployment調試和部署3.調試和部署調試和部署 LangGraph 為 透過 LangGraph 平台測試、調試和部署應用程式提供了簡單的入口。這包括 Studio，這是一個支援工作流或代理的可視化、交互和調試的 IDE。這還包括許多部署選項。 LangGraph 的生態系統 LangSmith — 有助於代理評估和可觀察性。調試性能LLM不佳的應用程式運行，評估代理軌跡，獲得生產中的可見性，並隨著時間的推移提高性能。 LangGraph 平臺 — 使用專門構建的部署平台輕鬆部署和擴展代理，以實現長時間運行的有狀態工作流。在 LangGraph Studio 中發現、重用、配置和共用代理，並通過 LangGraph Studio 中的可視化原型快速反覆運算。 Overview 概述 可靠性和可控性。通過審核檢查和人機協同審批來指導座席行動。LangGraph 為長時間運行的工作流保留上下文，讓您的代理保持正常。 低級且可擴展。使用完全描述性的低級基元構建自定義代理，這些基元不受限制自定義的殭化抽象的影響。設計可擴展的多代理系統，每個代理都充當根據您的使用案例量身定製的特定角色。 一流的流媒體支援。通過逐個令牌流式處理和中間步驟流式處理，LangGraph 使用戶能夠清楚地瞭解代理的推理和作，因為它們實時展開。 要熟悉 LangGraph 的關鍵概念和功能，請完成以下 LangGraph 基礎教程系列： Build a basic chatbot 構建基本聊天機器人 Add tools 添加工具: Add tools 新增工具： pip install -U langchain-tavily_set_env("TAVILY_API_KEY") from langchain_tavily import TavilySearchtool = TavilySearch(max_results=2)tools = [tool]tool.invoke("What's a 'node' in LangGraph?") {'query': "What's a 'node' in LangGraph?",'follow_up_questions': None,'answer': None,'images': [],'results': [{'title': "Introduction to LangGraph: A Beginner's Guide - Medium",'url': 'https://medium.com/@cplog/introduction-to-langgraph-a-beginners-guide-14f9be027141','content': 'Stateful Graph: LangGraph revolves around the concept of a stateful graph, where each node in the graph represents a step in your computation, and the graph maintains a state that is passed around and updated as the computation progresses. LangGraph supports conditional edges, allowing you to dynamically determine the next node to execute based on the current state of the graph. We define nodes for classifying the input, handling greetings, and handling search queries. def classify_input_node(state): LangGraph is a versatile tool for building complex, stateful applications with LLMs. By understanding its core concepts and working through simple examples, beginners can start to leverage its power for their projects. Remember to pay attention to state management, conditional edges, and ensuring there are no dead-end nodes in your graph.','score': 0.7065353,'raw_content': None},{'title': 'LangGraph Tutorial: What Is LangGraph and How to Use It?','url': 'https://www.datacamp.com/tutorial/langgraph-tutorial','content': 'LangGraph is a library within the LangChain ecosystem that provides a framework for defining, coordinating, and executing multiple LLM agents (or chains) in a structured and efficient manner. By managing the flow of data and the sequence of operations, LangGraph allows developers to focus on the high-level logic of their applications rather than the intricacies of agent coordination. Whether you need a chatbot that can handle various types of user requests or a multi-agent system that performs complex tasks, LangGraph provides the tools to build exactly what you need. LangGraph significantly simplifies the development of complex LLM applications by providing a structured framework for managing state and coordinating agent interactions.','score': 0.5008063,'raw_content': None}],'response_time': 1.38} 選擇LLM: pip install -U "langchain[openai]"import osfrom langchain.chat_models import init_chat_modelos.environ["OPENAI_API_KEY"] = "sk-..."llm = init_chat_model("openai:gpt-4.1")#將其合併到StateGraph中from typing import Annotatedfrom typing_extensions import TypedDictfrom langgraph.graph import StateGraph, START, ENDfrom langgraph.graph.message import add_messagesclass State(TypedDict):    messages: Annotated[list, add_messages]graph_builder = StateGraph(State)# Modification: tell the LLM which tools it can call# highlight-next-linellm_with_tools = llm.bind_tools(tools)def chatbot(state: State):    return {"messages": [llm_with_tools.invoke(state["messages"])]}graph_builder.add_node("chatbot", chatbot) 現在，創建一個函數，以便在調用工具時運行這些工具。為此，請將工具添加到名為BasicToolNode 的新節點，該節點檢查狀態中的最新消息，並在消息包含tool_calls時調用 tools。它依賴於 LLM的 tool_calling支援，該支援可在 Anthropic、OpenAI、Google Gemini 和許多其他LLM供應商中使用。 import jsonfrom langchain_core.messages import ToolMessageclass BasicToolNode:    """A node that runs the tools requested in the last AIMessage."""    def __init__(self, tools: list) -> None:        self.tools_by_name = {tool.name: tool for tool in tools}    def __call__(self, inputs: dict):        if messages := inputs.get("messages", []):            message = messages[-1]        else:            raise ValueError("No message found in input")        outputs = []        for tool_call in message.tool_calls:            tool_result = self.tools_by_name[tool_call["name"]].invoke(                tool_call["args"]            )            outputs.append(                ToolMessage(                    content=json.dumps(tool_result),                    name=tool_call["name"],                    tool_call_id=tool_call["id"],                )            )        return {"messages": outputs}tool_node = BasicToolNode(tools=[tool])graph_builder.add_node("tools", tool_node) 添加工具節點后，現在可以定義conditional_edges。 Edge 將控制流從一個節點路由到下一個節點。條件邊從單個節點開始，通常包含 「if」 語句，根據當前圖形狀態路由到不同的節點。這些函數接收當前圖形狀態並返回一個字串或字串清單，指示接下來要調用哪個節點。 接下來，定義一個名為 route_tools 的路由器函數，用於檢查聊天機器人輸出中的tool_calls。通過調用 add_conditional_edges 將此功能提供給圖形，這會告訴圖形，每當聊天機器人節點完成時，都會檢查此函數以查看下一步要去哪裡。 def route_tools(    state: State,):    """    Use in the conditional_edge to route to the ToolNode if the last message    has tool calls. Otherwise, route to the end.    """    if isinstance(state, list):        ai_message = state[-1]    elif messages := state.get("messages", []):        ai_message = messages[-1]    else:        raise ValueError(f"No messages found in input state to tool_edge: {state}")    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:        return "tools"    return END# The `tools_condition` function returns "tools" if the chatbot asks to use a tool, and "END" if# it is fine directly responding. This conditional routing defines the main agent loop.graph_builder.add_conditional_edges(    "chatbot",    route_tools,    # The following dictionary lets you tell the graph to interpret the condition's outputs as a specific node    # It defaults to the identity function, but if you    # want to use a node named something else apart from "tools",    # You can update the value of the dictionary to something else    # e.g., "tools": "my_tools"    {"tools": "tools", END: END},)# Any time a tool is called, we return to the chatbot to decide the next stepgraph_builder.add_edge("tools", "chatbot")graph_builder.add_edge(START, "chatbot")graph = graph_builder.compile() 視覺化 from IPython.display import Image, displaytry:    display(Image(graph.get_graph().draw_mermaid_png()))except Exception:    # This requires some extra dependencies and is optional    pass def stream_graph_updates(user_input: str):    for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):        for value in event.values():            print("Assistant:", value["messages"][-1].content)while True:    try:        user_input = input("User: ")        if user_input.lower() in ["quit", "exit", "q"]:            print("Goodbye!")            break        stream_graph_updates(user_input)    except:        # fallback if input() is not available        user_input = "What do you know about LangGraph?"        print("User: " + user_input)        stream_graph_updates(user_input)        break 3. Add memory 添加記憶體 同步版本 — SQLite Checkpointer #加入記憶，利用thread_id分割記憶流不同id不同短期記憶 -> 利用checkpointer產生snapshot儲存記憶流下面的每一個狀態from typing import Annotatedfrom langchain.chat_models import init_chat_modelfrom langchain_tavily import TavilySearchfrom langchain_core.messages import BaseMessagefrom typing_extensions import TypedDictfrom langgraph.checkpoint.memory import MemorySaverfrom langgraph.graph import StateGraphfrom langgraph.graph.message import add_messagesfrom langgraph.prebuilt import ToolNode, tools_conditionclass State(TypedDict):    messages: Annotated[list, add_messages]graph_builder = StateGraph(State)tool = TavilySearch(max_results=2)tools = [tool]llm_with_tools = llm.bind_tools(tools)def chatbot(state: State):    return {"messages": [llm_with_tools.invoke(state["messages"])]}graph_builder.add_node("chatbot", chatbot)tool_node = ToolNode(tools=[tool])graph_builder.add_node("tools", tool_node)graph_builder.add_conditional_edges(    "chatbot",    tools_condition,)graph_builder.add_edge("tools", "chatbot")graph_builder.set_entry_point("chatbot")memory = MemorySaver()graph = graph_builder.compile(checkpointer=memory) 非同步版本 — AsyncSqliteSaver import asynciofrom langgraph.checkpoint.sqlite.aio import AsyncSqliteSaverfrom langgraph.graph import StateGraphasync def main():    # 使用實際的資料庫文件    db_path = "./production_chatbot_async.db"        async with AsyncSqliteSaver.from_conn_string(db_path) as checkpointer:        # 建立同樣的圖表結構        graph_builder = StateGraph(State)        graph_builder.add_node("chatbot", chatbot)        graph_builder.set_entry_point("chatbot")        graph_builder.set_finish_point("chatbot")                # 編譯圖表        graph = graph_builder.compile(checkpointer=checkpointer)                # 配置        config = {"configurable": {"thread_id": "user_456", "checkpoint_ns": ""}}                # 非同步調用        result1 = await graph.ainvoke(            {"messages": [{"role": "user", "content": "Hi, I'm Bob"}]},            config=config        )        print(result1["messages"][-1].content)                # 第二次對話        result2 = await graph.ainvoke(            {"messages": [{"role": "user", "content": "Do you remember my name?"}]},            config=config        )        print(result2["messages"][-1].content)                # 列出所有 checkpoints        checkpoints = []        async for checkpoint in checkpointer.alist(config):            checkpoints.append(checkpoint)        print(f"Total checkpoints: {len(checkpoints)}")# 運行非同步主函數if __name__ == "__main__":    asyncio.run(main()) 自定義SQLite存checlpoint from typing import Annotatedfrom langchain.chat_models import init_chat_modelfrom langchain_tavily import TavilySearchfrom langchain_core.messages import BaseMessagefrom typing_extensions import TypedDictfrom langgraph.checkpoint.memory import MemorySaverfrom langgraph.graph import StateGraphfrom langgraph.graph.message import add_messagesfrom langgraph.prebuilt import ToolNode, tools_conditionclass State(TypedDict):    messages: Annotated[list, add_messages]graph_builder = StateGraph(State)tool = TavilySearch(max_results=2)tools = [tool]llm_with_tools = llm.bind_tools(tools)def chatbot(state: State):    return {"messages": [llm_with_tools.invoke(state["messages"])]}graph_builder.add_node("chatbot", chatbot)tool_node = ToolNode(tools=[tool])graph_builder.add_node("tools", tool_node)graph_builder.add_conditional_edges(    "chatbot",    tools_condition,)graph_builder.add_edge("tools", "chatbot")graph_builder.set_entry_point("chatbot")memory = MemorySaver()graph = graph_builder.compile(checkpointer=memory) 4. Add human-in-the-loop controls4. 添加人機協同控制添加人機回圈控件 這裡我會一步步說明如何使用 LangGraph 實現人機回饋機制（human-in-the-loop）。 在聊天機器人中加入人機回饋控制 有時候，AI 代理可能不夠可靠，需要人類的輸入才能成功完成任務。同樣地，對於某些重要操作，你可能會想要在執行前獲得人類的批准，以確保一切按預期運行。 LangGraph 的持久層支援人機交互工作流，允許根據用戶反饋暫停和恢復執行。主要介面是 interrupt 函數，在節點內調用它會暫停執行，然後透過傳入 Command 來恢復執行，並接收來自人類的新輸入。 1. 安裝必要套件並設定環境 # 安裝所需套件!pip install -U langgraph langchain langchain-openai langchain-tavily # 設定環境變數import osfrom langchain.chat_models import init_chat_model # 設定你的 OpenAI API keyos.environ["OPENAI_API_KEY"] = "sk-..."# 初始化聊天模型llm = init_chat_model("openai:gpt-4") 2. 導入必要的模組 from typing import Annotatedfrom langchain_tavily import TavilySearchfrom langchain_core.tools import toolfrom typing_extensions import TypedDict from langgraph.checkpoint.memory import MemorySaverfrom langgraph.graph import StateGraph, START, ENDfrom langgraph.graph.message import add_messagesfrom langgraph.prebuilt import ToolNode, tools_conditionfrom langgraph.types import Command, interrupt 3. 定義狀態類型 class State(TypedDict):    messages: Annotated[list, add_messages] 4. 創建人機交互工具 這個工具使用 interrupt 函數來暫停執行並等待人類輸入： @tooldef human_assistance(query: str) -> str:    """Request assistance from a human."""    human_response = interrupt({"query": query})    return human_response["data"] 5. 設定工具和模型 # 創建搜尋工具tool = TavilySearch(max_results=2) # 將人機交互工具和搜尋工具組合tools = [tool, human_assistance]# 將工具綁定到語言模型llm_with_tools = llm.bind_tools(tools) 6. 定義聊天機器人節點 def chatbot(state: State):    message = llm_with_tools.invoke(state["messages"])    # 因為我們將在工具執行期間中斷，    # 我們禁用並行工具調用以避免在恢復時重複任何工具調用    assert len(message.tool_calls) <= 1    return {"messages": [message]} 7. 構建圖形 # 初始化圖形構建器graph_builder = StateGraph(State) # 添加聊天機器人節點graph_builder.add_node("chatbot", chatbot)# 添加工具節點tool_node = ToolNode(tools=tools)graph_builder.add_node("tools", tool_node)# 添加條件邊和正常邊graph_builder.add_conditional_edges(    "chatbot",    tools_condition,)graph_builder.add_edge("tools", "chatbot")graph_builder.add_edge(START, "chatbot") 8. 編譯圖形 # 創建記憶體保存器memory = MemorySaver() # 編譯圖形graph = graph_builder.compile(checkpointer=memory) 9. 可視化圖形（可選） from IPython.display import Image, display try:    display(Image(graph.get_graph().draw_mermaid_png()))except Exception:    # 這需要一些額外的依賴並且是可選的    pass 10. 測試人機交互功能 首先，向聊天機器人發送一個需要人類協助的請求： user_input = "I need some expert guidance for building an AI agent. Could you request assistance for me?"config = {"configurable": {"thread_id": "1"}} # 流式執行圖形events = graph.stream(    {"messages": [{"role": "user", "content": user_input}]},    config,    stream_mode="values",)# 顯示輸出for event in events:    if "messages" in event:        event["messages"][-1].pretty_print() 11. 檢查執行狀態 當執行暫停時，檢查圖形的當前狀態： # 獲取當前狀態snapshot = graph.get_state(config)print("Next node to execute:", snapshot.next) 12. 恢復執行並提供人類回應 當準備好恢復執行時，創建一個 Command 物件並傳入人類的回應： # 準備人類的回應human_response = (    "We, the experts are here to help! We'd recommend you check out LangGraph to build your agent."    " It's much more reliable and extensible than simple autonomous agents.") # 創建恢復命令human_command = Command(resume={"data": human_response})# 恢復執行events = graph.stream(human_command, config, stream_mode="values")# 顯示輸出for event in events:    if "messages" in event:        event["messages"][-1].pretty_print() 13. 驗證完整對話 # 檢查完整的對話歷史final_state = graph.get_state(config)print(f"總共有 {len(final_state.values['messages'])} 條訊息") # 顯示所有訊息for msg in final_state.values['messages']:    print(f"{msg.get('role', 'unknown')}: {msg.get('content', msg)[:100]}...") 這個實作展示了如何在 LangGraph 中加入人機回饋機制。當 AI 需要人類協助時，執行會暫停並等待人類輸入，然後根據人類的回應繼續執行。這種模式對於需要人類監督和干預的關鍵任務非常有用。 from typing import Annotatedfrom langchain_tavily import TavilySearchfrom langchain_core.messages import ToolMessagefrom langchain_core.tools import InjectedToolCallId, toolfrom typing_extensions import TypedDictfrom langgraph.checkpoint.memory import MemorySaverfrom langgraph.graph import StateGraph, START, ENDfrom langgraph.graph.message import add_messagesfrom langgraph.prebuilt import ToolNode, tools_conditionfrom langgraph.types import Command, interruptfrom langchain.chat_models import init_chat_modelimport os# 設定環境os.environ["OPENAI_API_KEY"] = ""# 定義包含自訂欄位的狀態class State(TypedDict):    messages: Annotated[list, add_messages]    name: str  # 新增的欄位    birthday: str  # 新增的欄位# 人工協助工具，可以更新狀態@tooldef human_assistance(    name: str, birthday: str, tool_call_id: Annotated[str, InjectedToolCallId]) -> str:    """Request assistance from a human."""    # 暫停執行並等待人類回應    human_response = interrupt(        {            "question": "Is this correct?",            "name": name,            "birthday": birthday,        },    )        # 根據人類回應決定是否更新資訊    if human_response.get("correct", "").lower().startswith("y"):        verified_name = name        verified_birthday = birthday        response = "Correct"    else:        verified_name = human_response.get("name", name)        verified_birthday = human_response.get("birthday", birthday)        response = f"Made a correction: {human_response}"    # 使用 Command 物件更新狀態    state_update = {        "name": verified_name,        "birthday": verified_birthday,        "messages": [ToolMessage(response, tool_call_id=tool_call_id)],    }    return Command(update=state_update)# 設定工具和模型tool = TavilySearch(max_results=2)tools = [tool, human_assistance]llm = init_chat_model("openai:gpt-4")llm_with_tools = llm.bind_tools(tools)# 定義聊天機器人節點def chatbot(state: State):    message = llm_with_tools.invoke(state["messages"])    assert(len(message.tool_calls) <= 1)    return {"messages": [message]}# 構建圖形graph_builder = StateGraph(State)graph_builder.add_node("chatbot", chatbot)tool_node = ToolNode(tools=tools)graph_builder.add_node("tools", tool_node)graph_builder.add_conditional_edges("chatbot", tools_condition)graph_builder.add_edge("tools", "chatbot")graph_builder.add_edge(START, "chatbot")# 編譯圖形memory = MemorySaver()graph = graph_builder.compile(checkpointer=memory)# 使用範例config = {"configurable": {"thread_id": "1"}}# 開始對話，確保初始狀態包含所有必要的欄位initial_state = {    "messages": [{        "role": "user",         "content": "Can you look up when LangGraph was released? When you have the answer, use the human_assistance tool for review."    }],    "name": "",  # 初始化空值    "birthday": ""  # 初始化空值}print("=== 開始對話 ===")events = graph.stream(    initial_state,    config,    stream_mode="values",)# 顯示第一部分的對話for event in events:    if "messages" in event:        print(f"訊息數量: {len(event['messages'])}")        event["messages"][-1].pretty_print()# 模擬人類回應print("\n=== 人類介入回應 ===")human_command = Command(    resume={        "correct": "no",        "name": "LangGraph",        "birthday": "Jan 17, 2024",    },)# 繼續執行events = graph.stream(human_command, config, stream_mode="values")for event in events:    if "messages" in event:        event["messages"][-1].pretty_print()# 查看更新後的狀態print("\n=== 檢查狀態 ===")snapshot = graph.get_state(config)# 安全地檢查狀態中的值if "name" in snapshot.values:    print(f"Name: {snapshot.values['name']}")else:    print("Name 未設置")if "birthday" in snapshot.values:    print(f"Birthday: {snapshot.values['birthday']}")else:    print("Birthday 未設置")# 手動更新狀態print("\n=== 手動更新狀態 ===")update_result = graph.update_state(config, {"name": "LangGraph (library)", "birthday": "Jan 17, 2024"})print(f"更新結果: {update_result}")# 再次查看狀態print("\n=== 更新後的狀態 ===")final_snapshot = graph.get_state(config)print(f"Name: {final_snapshot.values.get('name', '未設置')}")print(f"Birthday: {final_snapshot.values.get('birthday', '未設置')}")# 顯示所有訊息print("\n=== 完整對話歷史 ===")for i, msg in enumerate(final_snapshot.values.get('messages', [])):    role = msg.__class__.__name__    content = str(msg.content)[:100] + "..." if len(str(msg.content)) > 100 else str(msg.content)    print(f"{i+1}. {role}: {content}") 5. Customize state 自訂狀態 自訂狀態欄位 這個教程展示如何在 LangGraph 的狀態中添加自定義欄位（name 和 birthday），讓聊天機器人能夠查找特定資訊並提交給人類審查。透過在工具內使用 Command 物件來更新狀態，實現更複雜的工作流程。 6. Time travel 時間旅行 時間旅行功能 LangGraph 的時間旅行功能讓你可以回到對話的任何之前狀態，從那個時間點繼續執行。這對於探索不同的對話路徑、修正錯誤或嘗試不同策略非常有用。 from typing import Annotatedfrom langchain_tavily import TavilySearchfrom langchain_core.messages import BaseMessagefrom typing_extensions import TypedDictfrom langgraph.checkpoint.memory import MemorySaverfrom langgraph.graph import StateGraph, START, ENDfrom langgraph.graph.message import add_messagesfrom langgraph.prebuilt import ToolNode, tools_conditionfrom langchain.chat_models import init_chat_modelimport os# 設定環境os.environ["OPENAI_API_KEY"] = "你的API金鑰"llm = init_chat_model("openai:gpt-4")# 定義狀態class State(TypedDict):    messages: Annotated[list, add_messages]# 建立圖形graph_builder = StateGraph(State)# 設定工具tool = TavilySearch(max_results=2)tools = [tool]llm_with_tools = llm.bind_tools(tools)# 定義聊天機器人節點def chatbot(state: State):    return {"messages": [llm_with_tools.invoke(state["messages"])]}# 構建圖形graph_builder.add_node("chatbot", chatbot)tool_node = ToolNode(tools=[tool])graph_builder.add_node("tools", tool_node)graph_builder.add_conditional_edges("chatbot", tools_condition)graph_builder.add_edge("tools", "chatbot")graph_builder.add_edge(START, "chatbot")# 編譯圖形（必須使用 checkpointer）memory = MemorySaver()graph = graph_builder.compile(checkpointer=memory)# 開始對話config = {"configurable": {"thread_id": "1"}}# 第一次交互events = graph.stream(    {        "messages": [            {                "role": "user",                "content": "I'm learning LangGraph. Could you do some research on it for me?",            },        ],    },    config,    stream_mode="values",)for event in events:    if "messages" in event:        print(f"訊息數量: {len(event['messages'])}")# 第二次交互events = graph.stream(    {        "messages": [            {                "role": "user",                "content": "Ya that's helpful. Maybe I'll build an autonomous agent with it!",            },        ],    },    config,    stream_mode="values",)for event in events:    if "messages" in event:        print(f"訊息數量: {len(event['messages'])}")# 查看完整的狀態歷史print("\n=== 狀態歷史 ===")to_replay = Nonefor state in graph.get_state_history(config):    print(f"訊息數量: {len(state.values['messages'])}, 下一個節點: {state.next}")    print(f"檢查點ID: {state.config.get('configurable', {}).get('checkpoint_id')}")    print("-" * 80)        # 選擇一個特定的狀態來重播（例如選擇有6條訊息的狀態）    if len(state.values["messages"]) == 6:        to_replay = state# 從選定的檢查點恢復執行if to_replay:    print(f"\n=== 從檢查點恢復 ===")    print(f"要重播的下一個節點: {to_replay.next}")    print(f"檢查點配置: {to_replay.config}")        # 從特定時間點恢復執行    events = graph.stream(None, to_replay.config, stream_mode="values")        for event in events:        if "messages" in event:            last_msg = event["messages"][-1]            print(f"\n最後一條訊息類型: {last_msg.__class__.__name__}")            if hasattr(last_msg, 'content'):                print(f"內容: {str(last_msg.content)[:200]}...")# 也可以直接使用檢查點ID來加載特定狀態specific_config = {    "configurable": {        "thread_id": "1",        "checkpoint_id": to_replay.config["configurable"]["checkpoint_id"] if to_replay else None    }}# 獲取特定時間點的狀態if to_replay:    specific_state = graph.get_state(specific_config)    print(f"\n=== 特定時間點的狀態 ===")    print(f"訊息數量: {len(specific_state.values['messages'])}")    print(f"下一個節點: {specific_state.next}")# 示範：從中間狀態開始，改變對話方向print("\n=== 改變對話方向 ===")if to_replay:    # 從之前的狀態開始，但發送不同的訊息    new_events = graph.stream(        {            "messages": [                {                    "role": "user",                    "content": "Actually, let me search for something else. What about LangSmith?",                }            ]        },        to_replay.config,        stream_mode="values",    )        for event in new_events:        if "messages" in event:            print(f"新路徑 - 訊息數量: {len(event['messages'])}") 關鍵功能： get_state_history() - 獲取完整的狀態歷史 每個狀態都有唯一的 checkpoint_id 可以使用特定的 checkpoint_id 來恢復執行 從任何檢查點恢復後，可以改變對話方向 所有歷史狀態都被保存，可以任意回溯