前言

这个blog用来记录学习langchain框架的过程

langgraph

langgraph有3要素：状态变量，节点，边

流程如下：

1. 定义状态变量

# 定义状态
class ChatState(TypedDict):
    messages: Annotated[list, add_messages]

3. 创建llm节点

# 创建 LLM 节点
def chatbot(state: ChatState):
    llm = ChatOpenAI(
        model='qwen-plus',
        base_url='https://dashscope.aliyuncs.com/compatible-mode/v1',
        api_key=dotenv.get_key('.env', 'DASHSCOPE_API_KEY'),
    )
    
    response = llm.invoke(state["messages"])
    
    # 注意：这里只返回新消息，add_messages 会自动将它追加到状态中
    return {"messages": [response]}

5. 构建图

# 构建图
workflow = StateGraph(ChatState)
workflow.add_node("chatbot", chatbot)
workflow.add_edge(START, "chatbot")
workflow.add_edge("chatbot", END)

# 创建检查点存储
checkpointer = InMemorySaver()

# 编译图时指定检查点
graph = workflow.compile(checkpointer=checkpointer)

# 创建线程ID用于保存对话历史
thread_id = str(uuid.uuid4())

6. 生成选择和循环的边

### 选择 ###



workflow.add_conditional_edges(
    "router",              # 源节点：从哪个节点出发
    route_decision,         # 决策函数：根据状态决定走哪条路
    {                       # 路由映射：决策结果 → 目标节点
        "normal": "chatbot",                    # 如果返回"normal" → 去chatbot节点
        "complex_task": "complex_handler",      # 如果返回"complex_task" → 去complex_handler节点
        "confirmation": "confirmation_handler"  # 如果返回"confirmation" → 去confirmation_handler节点
    }
)
### 循环和选择差不多，就是把去的节点写到源节点就可以了

7. 创建工具

### 先定义工具函数 ###
@tool
def write_to_file(file_path: str, content: str, mode: str = "a") -> str:
    """将内容写入文件。支持追加模式（'a'）和覆盖模式（'w'）。
    
    Args:
        file_path: 文件路径
        content: 要写入的内容
        mode: 写入模式，'a'表示追加，'w'表示覆盖，默认为'a'
    
    Returns:
        操作结果信息
    """
    try:
        with open(file_path, mode, encoding='utf-8') as f:
            f.write(content)
        return f"✅ 成功写入文件: {file_path} (模式: {mode})"
    except Exception as e:
        return f"❌ 写入文件失败: {str(e)}"
### 将工具绑定到 LLM ###
    llm = ChatOpenAI(
        model='qwen-plus',
        base_url='https://dashscope.aliyuncs.com/compatible-mode/v1',
        api_key=dotenv.get_key('.env', 'DASHSCOPE_API_KEY'),
    ).bind_tools(tools)

注：为了保证对话纪录被永久保存，这里需要引入两个变量，checkpoint和uuid，一个用来存档，保存对话，是langgraph的状态持久化机制，一个用来标记是哪一段对话，因为用户可能会开启两段与ai的对话

工具

ai聊天机器人的完整代码如下：

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from typing import Annotated, TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
import dotenv
from langgraph.checkpoint.memory import InMemorySaver
import uuid


# 定义状态
class ChatState(TypedDict):
    messages: Annotated[list, add_messages]


# 创建 LLM 节点
def chatbot(state: ChatState):
    llm = ChatOpenAI(
        model='qwen-plus',
        base_url='https://dashscope.aliyuncs.com/compatible-mode/v1',
        api_key=dotenv.get_key('.env', 'DASHSCOPE_API_KEY'),
    )
    
    response = llm.invoke(state["messages"])
    
    # 注意：这里只返回新消息，add_messages 会自动将它追加到状态中
    return {"messages": [response]}


# 构建图
workflow = StateGraph(ChatState)
workflow.add_node("chatbot", chatbot)
workflow.add_edge(START, "chatbot")
workflow.add_edge("chatbot", END)

# 创建检查点存储
checkpointer = InMemorySaver()

# 编译图时指定检查点
graph = workflow.compile(checkpointer=checkpointer)

# 创建线程ID用于保存对话历史
thread_id = str(uuid.uuid4())

print("=" * 60)
print("聊天机器人已启动！输入 'quit' 或 'exit' 退出")
print("=" * 60)
print()

# 显示初始欢迎消息
initial_input = {"messages": [HumanMessage(content="你好, 我的名字是大冉")]}
result = graph.invoke(initial_input, config={"configurable": {"thread_id": thread_id}})

# 显示AI的欢迎回复
print("🤖 机器人:", result["messages"][-1].content)
print()

# 交互式聊天循环
while True:
    # 获取用户输入
    user_input = input("👤 你: ").strip()
    
    # 检查退出条件
    if user_input.lower() in ['quit', 'exit', '退出', 'q']:
        print("\n再见！")
        break
    
    if not user_input:
        print("请输入有效内容！")
        continue
    
    # 调用图处理用户消息
    try:
        result = graph.invoke(
            {"messages": [HumanMessage(content=user_input)]},
            config={"configurable": {"thread_id": thread_id}}
        )
        
        # 显示AI回复
        ai_response = result["messages"][-1].content
        print(f"🤖 机器人: {ai_response}")
        print()
        
    except Exception as e:
        print(f"❌ 错误: {e}")
        print()

参考资料

https://zhuanlan.zhihu.com/p/1945401093786940263