# 使用uv安裝LangGraph和相關依賴
uv pip install -U langgraph langchain python-dotenv typing-extensions

# .env文件內(nèi)容
DEEPSEEK_API_KEY=your_deepseek_api_key_here

from typing import Annotated

from langchain.chat_models import init_chat_model
from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages

import os
from dotenv import load_dotenv

# 加載.env文件中的環(huán)境變量
load_dotenv()

class State(TypedDict):
    messages: Annotated[list, add_messages]


graph_builder = StateGraph(State)


llm = init_chat_model(
    "deepseek-chat",  # 使用DeepSeek模型
    api_key=os.environ.get("DEEPSEEK_API_KEY")
)


def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}


# 添加節(jié)點和邊
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph = graph_builder.compile()

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)


whileTrue:
    try:
        user_input = input("User: ")
        if user_input.lower() in ["quit", "exit", "q"]:
            print("Goodbye!")
            break
        stream_graph_updates(user_input)
    except KeyboardInterrupt:
        print("\nGoodbye!")
        break

from typing import Annotated
from langchain.chat_models import init_chat_model
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
import os
from dotenv import load_dotenv

load_dotenv()

class State(TypedDict):
    messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)

llm = init_chat_model(
    "deepseek-chat",  # 使用DeepSeek模型
    api_key=os.environ.get("DEEPSEEK_API_KEY")
)

def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}

graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph = graph_builder.compile()

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 KeyboardInterrupt:
        print("\nGoodbye!")
        break

uv run 1-build-basic-chatbot.py

User: who are you?
Assistant: I’m DeepSeek Chat, your AI assistant created by **DeepSeek**! ??? I’m here to help answer your questions, provide insights, and assist with anything from general knowledge to coding, writing, and more.

Think of me as a friendly, knowledgeable companion—always ready to chat, research, or brainstorm with you! ??

What can I help you with today? ??

User: exit
Goodbye!

print(graph.get_graph().draw_mermaid())

---
config:
  flowchart:
    curve: linear
---
graph TD;
	__start__([<p>__start__</p>]):::first
	chatbot(chatbot)
	__end__([<p>__end__</p>]):::last
	__start__ --> chatbot;
	chatbot --> __end__;
	classDef default fill:#f2f0ff,line-height:1.2
	classDef first fill-opacity:0
	classDef last fill:#bfb6fc

uv pip install -U langchain-tavily httpx

DEEPSEEK_API_KEY=your_deepseek_api_key
TAVILY_API_KEY=your_tavily_api_key

from typing import Annotated

from langchain.chat_models import init_chat_model
from langchain_tavily import TavilySearch
from langchain_core.messages import BaseMessage
from typing_extensions import TypedDict

from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

import os
from dotenv import load_dotenv

# 加載.env文件中的環(huán)境變量
load_dotenv()

class State(TypedDict):
    messages: Annotated[list, add_messages]


graph_builder = StateGraph(State)

llm = init_chat_model(
    "deepseek-chat",  # 使用DeepSeek模型
    api_key=os.environ.get("DEEPSEEK_API_KEY")
)

# 初始化Tavily搜索工具
tool = TavilySearch(max_results=2)
tools = [tool]

# 將工具綁定到LLM
llm_with_tools = llm.bind_tools(tools)

def chatbot(state: State):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

# 添加聊天機器人節(jié)點
graph_builder.add_node("chatbot", chatbot)

# 添加工具節(jié)點
tool_node = ToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)

# 添加條件邊
graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,
)
# 工具調(diào)用完成后，返回到聊天機器人節(jié)點
graph_builder.add_edge("tools", "chatbot")
graph_builder.set_entry_point("chatbot")
graph = graph_builder.compile()

# 打印圖結構
print(graph.get_graph().draw_mermaid())

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)

whileTrue:
    try:
        user_input = input("User: ")
        if user_input.lower() in ["quit", "exit", "q"]:
            print("Goodbye!")
            break
        stream_graph_updates(user_input)
    except KeyboardInterrupt:
        print("\nGoodbye!")
        break

from langchain_tavily import TavilySearch
from langgraph.prebuilt import ToolNode, tools_condition

tool = TavilySearch(max_results=2)
tools = [tool]

llm_with_tools = llm.bind_tools(tools)

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")
graph = graph_builder.compile()

print(graph.get_graph().draw_mermaid())

uv run 2-add-tools.py

User: 什么是LangGraph的核心組件？

Assistant:
Assistant: {"query": "LangGraph 核心組件", "follow_up_questions": null, "answer": null, "images": [], "results": [{"url": "https://cloud.tencent.com/developer/news/2180955", "title": "Agent快速構建框架的langGraph到底是什么及案例 - 騰訊云", "content": "LangGraph是一個使用 LLM 構建有狀態(tài)、多參與者應用程序的庫，用于創(chuàng)建代理和多代理工作流。請在此處查看入門教程。\n\nLangGraph 的靈感來自Pregel和Apache Beam。公共接口靈感來自NetworkX。LangGraph 由 LangChain 的創(chuàng)建者 LangChain Inc 構建，但可以在沒有 LangChain 的情況下使用。\n\nLangGraph核心組件\n\nLangGraph的核心組件包括Graphs、State、Nodes、Edges、Send、Checkpointer。LangGraph有哪些優(yōu)點：可控性、持久性、Human-in-the-loop、ReactAgent。\n\nstate\n\nfrom langgraph.graph import StateGraph\n\nfrom typing import TypedDict, List, Annotated\n\nimport Operator\n\nclass State(TypedDict):\n\n  input: str", "score": 0.9403148, "raw_content": null}, {"url": "https://blog.csdn.net/TaloyerG/article/details/142546057", "title": "Langgraph 的Planning Agent 深度解析原創(chuàng) - CSDN博客", "content": "Langgraph 的核心組件包括語言理解模塊、知識庫和規(guī)劃模塊。 1.2.1 語言理解模塊. 語言理解模塊負責解析用戶的自然語言輸入，將其轉化為可供計算機處理的", "score": 0.9348309, "raw_content": null}], "response_time": 1.87}
Assistant: LangGraph的核心組件主要包括以下幾個方面：

1. **Graphs（圖）**：用于構建和管理工作流的結構。
2. **State（狀態(tài)）**：用于跟蹤和管理應用程序的狀態(tài)。
3. **Nodes（節(jié)點）**：表示工作流中的單個步驟或操作。
4. **Edges（邊）**：定義節(jié)點之間的關系和流程控制。
5. **Send（發(fā)送）**：用于在節(jié)點之間傳遞數(shù)據(jù)或消息。
6. **Checkpointer（檢查點）**：用于持久化和恢復工作流的狀態(tài)。

此外，LangGraph還支持可控性、持久性、Human-in-the-loop（人工干預）和ReactAgent（反應式代理）等功能特性。

這些組件共同構成了LangGraph的基礎，使其能夠高效地構建和管理有狀態(tài)的多參與者應用程序或代理工作流。

User: exit
Goodbye!

"""LangGraph 教程: 添加記憶功能的聊天機器人

本示例展示了如何使用 LangGraph 的檢查點功能為聊天機器人添加記憶功能，
使其能夠記住對話歷史并在多輪對話中保持上下文。
"""

from typing import Annotated

from langchain.chat_models import init_chat_model
from langchain_tavily import TavilySearch
from langchain_core.messages import BaseMessage
from typing_extensions import TypedDict

# 導入 MemorySaver 用于實現(xiàn)記憶功能
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

import os
from dotenv import load_dotenv

# 加載.env文件中的環(huán)境變量
load_dotenv()

# 定義狀態(tài)類型，使用 add_messages 注解來自動合并消息列表
class State(TypedDict):
    messages: Annotated[list, add_messages]  # 消息列表將使用 add_messages reducer 自動合并


# 初始化 DeepSeek 聊天模型
llm = init_chat_model(
    "deepseek-chat",  # 使用DeepSeek模型
    api_key=os.environ.get("DEEPSEEK_API_KEY")  # 從環(huán)境變量中獲取API密鑰
)

# 創(chuàng)建狀態(tài)圖構建器
graph_builder = StateGraph(State)

# 初始化Tavily搜索工具
print("\n初始化Tavily搜索工具...")
tool = TavilySearch(max_results=2)  # 設置最多返回2個搜索結果
tools = [tool]

# 將工具綁定到LLM
llm_with_tools = llm.bind_tools(tools)

# 定義聊天機器人節(jié)點函數(shù)
def chatbot(state: State):
    """LLM節(jié)點函數(shù)，處理用戶輸入并生成響應"""
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

# 添加聊天機器人節(jié)點
graph_builder.add_node("chatbot", chatbot)

# 添加工具節(jié)點
tool_node = ToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)

# 添加條件邊
graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,
)
# 工具調(diào)用完成后，返回到聊天機器人節(jié)點
graph_builder.add_edge("tools", "chatbot")
graph_builder.set_entry_point("chatbot")

print("\n構建圖并添加記憶功能...")

# 創(chuàng)建內(nèi)存保存器
print("\n創(chuàng)建 MemorySaver 實例作為檢查點保存器...")
memory = MemorySaver()  # 在內(nèi)存中保存狀態(tài)，適用于開發(fā)和測試

# 使用內(nèi)存保存器編譯圖
print("使用檢查點保存器編譯圖...")
graph = graph_builder.compile(checkpointer=memory)  # 將內(nèi)存保存器傳遞給圖

# 打印圖結構
print("\n圖結構如下：")
print(graph.get_graph().draw_mermaid())

# 定義對話線程ID
print("\n設置對話線程 ID = '1'...")
config = {"configurable": {"thread_id": "1"}}  # 使用線程ID來標識和區(qū)分不同的對話

# 示例 1: 第一次對話
print("\n示例 1: 第一次對話 - 用戶介紹自己")
user_input = "Hi there! My name is Will."
print(f"\n用戶輸入: '{user_input}'")

# 注意: config 是 stream() 函數(shù)的第二個參數(shù)!
print("使用線程 ID '1' 調(diào)用圖...")
events = graph.stream(
    {"messages": [{"role": "user", "content": user_input}]},
    config,  # 傳遞包含 thread_id 的配置
    stream_mode="values",
)

print("\n助理回應:")
for event in events:
    event["messages"][-1].pretty_print()  # 打印助理的回應

# 示例 2: 測試記憶功能
print("\n\n示例 2: 第二次對話 - 測試記憶功能")
user_input = "Remember my name?"
print(f"\n用戶輸入: '{user_input}'")

# 使用相同的線程ID再次調(diào)用圖
print("使用相同的線程 ID '1' 再次調(diào)用圖...")
events = graph.stream(
    {"messages": [{"role": "user", "content": user_input}]},
    config,  # 使用相同的配置，圖將加載之前保存的狀態(tài)
    stream_mode="values",
)

print("\n助理回應 (應該記得用戶名字):")
for event in events:
    event["messages"][-1].pretty_print()

# 示例 3: 新對話線程
print("\n\n示例 3: 新對話線程 - 測試線程隔離")
print("創(chuàng)建新的線程 ID = '2'...")

# 使用不同的線程ID
print("使用新的線程 ID '2' 調(diào)用圖...")
events = graph.stream(
    {"messages": [{"role": "user", "content": user_input}]},
    {"configurable": {"thread_id": "2"}},  # 使用新的線程ID
    stream_mode="values",
)

print("\n助理回應 (不應該記得用戶名字):")
for event in events:
    event["messages"][-1].pretty_print()

# 示例 4: 返回第一個線程
print("\n\n示例 4: 返回第一個線程 - 驗證記憶持久性")
print(f"\n用戶輸入: '{user_input}'")

# 再次使用第一個線程ID
print("再次使用線程 ID '1' 調(diào)用圖...")
events = graph.stream(
    {"messages": [{"role": "user", "content": user_input}]},
    config,  # 使用原始線程ID
    stream_mode="values",
)

print("\n助理回應 (應該仍然記得用戶名字):")
for event in events:
    event["messages"][-1].pretty_print()

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "1"}}

def stream_graph_updates(user_input: str):
    # 注意：config是stream()的第二個位置參數(shù)
    for event in graph.stream(
        {"messages": [{"role": "user", "content": user_input}]},
        config,
        stream_mode="values",
    ):
        for value in event.values():
            print("Assistant:", value["messages"][-1].content)

uv run 3-add-memory.py

設置對話線程 ID = '1'...

示例 1: 第一次對話 - 用戶介紹自己

用戶輸入: 'Hi there! My name is Will.'
使用線程 ID '1' 調(diào)用圖...

助理回應:
================================ Human Message =================================

Hi there! My name is Will.
================================== Ai Message ==================================

Hi Will! It's great to meet you. How can I assist you today?


示例 2: 第二次對話 - 測試記憶功能

用戶輸入: 'Remember my name?'
使用相同的線程 ID '1' 再次調(diào)用圖...

助理回應 (應該記得用戶名字):
================================ Human Message =================================

Remember my name?
================================== Ai Message ==================================

Of course, Will! I’ll remember your name for the rest of our conversation. How can I help you today? ??


示例 3: 新對話線程 - 測試線程隔離
創(chuàng)建新的線程 ID = '2'...
使用新的線程 ID '2' 調(diào)用圖...

助理回應 (不應該記得用戶名字):
================================ Human Message =================================

Remember my name?
================================== Ai Message ==================================

I don’t have the ability to remember personal details like names between interactions. However, you can tell me your name, and I’ll happily use it for the rest of our current conversation! What should I call you?


示例 4: 返回第一個線程 - 驗證記憶持久性

用戶輸入: 'Remember my name?'
再次使用線程 ID '1' 調(diào)用圖...

助理回應 (應該仍然記得用戶名字):
================================ Human Message =================================

Remember my name?
================================== Ai Message ==================================

Got it, Will! I’ll keep your name in mind while we chat. Let me know how I can assist you! ??