from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
2.定义状态结构：
class State(TypedDict):
# 'messages' will store the chatbot conversation history.
# The 'add_messages' function ensures new messages are appended to the list.
messages: Annotated[list, add_messages]
# Create an instance of the StateGraph, passing in the State class
# Use the LLM to generate a response based on the current conversation history.
response = llm.invoke(state["messages"])
# Return the updated state with the new message appended
return {"messages": [response]}
# Add the 'chatbot' node to the graph,
graph_builder.add_node("chatbot", chatbot)
5.定义入口与出口：
# For this basic chatbot, the 'chatbot' node is both the entry and finish point
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)
6.编译：
graph = graph_builder.compile()
7.可视化：
from IPython.display import Image, display
try:
display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
# This requires some extra dependencies and is optional
pass
8.运行：
while True:
user_input = input("User: ")
if user_input.lower() in ["quit", "exit", "q"]:
print("Goodbye!")
break
# Process user input through the LangGraph
for event in graph.stream({"messages": [("user", user_input)]}):
for value in event.values():
print("Assistant:", value["messages"][-1].content)
大家修改一下，把一个可以在web上搜索信息的工具纳入进来。用的是langchain_community.tools.tavily_search的TavilySearchResults工具。当然，你得配置Tavily API key。
#pip install -U tavily-python langchain_community
from typing import Annotated
from langchain_anthropic import ChatAnthropic
from langchain_community.tools.tavily_search import TavilyearchResults
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
class State(TypedDict):
messages: Annotated[list, add_messages]
graph_builder.add_conditional_edges(
"chatbot",
tools_condition,
)
# Any time a tool is called, we return to the chatbot to decide the next step
graph_builder.add_edge("tools", "chatbot")
graph_builder.set_entry_point("chatbot")
graph = graph_builder.compile()
6. 给Chatbot添加记忆功能
记忆功能对创建可参与会话的聊天机器人十分关键，因为机器人得记住过去的互动。
LangGraph的检查点系统通过thread_id来保存状态，支撑跨会话的连续性。
实现方法：
# ... (Previous code to define State, graph_builder, nodes, and edges)
from langgraph.checkpoint.memory import MemorySaver
# Create a MemorySaver object to act as the checkpointer
memory = MemorySaver()
# Compile the graph, passing in the 'memory' object as the checkpointer
graph = graph_builder.compile(checkpointer=memory)
from langchain_anthropic import ChatAnthropic
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.messages import BaseMessage
from typing_extensions import TypedDict
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
class State(TypedDict):
messages: Annotated[list, add_messages]
memory = MemorySaver()
graph = graph_builder.compile(
checkpointer=memory,
# This is new!
interrupt_before=["tools"],
# Note: can also interrupt __after__ actions, if desired.
# interrupt_after=["tools"]
)
9
发表于 2025-3-28 10:04:54
