如何使用 LangGraph 构建应用程序：分步指南

标签#LangGraph#LangChain#AI#代理#Python#LLM#状态管理#工作流

大家好，我想向大家介绍一下我正在研究和开发的工具 LangGraph。

基本上，传统的人工智能应用程序在处理复杂的工作流和动态状态时经常面临挑战。LangGraph提供了一个强大的解决方案，可以创建有状态的代理，管理复杂的对话，做出基于上下文的决策，并执行复杂的工作流。

本文提供了使用LangGraph 构建应用程序的分步指南，LangGraph 是一个用于创建具有状态图的多步骤代理的框架。

实施步骤：

1. 设置环境和安装依赖项
2. 定义应用程序状态
3. 创建图节点
4. 配置状态图
5. 运行代理

1.设置环境并安装依赖项

第一步是设置 Python 环境并安装必要的库：

pip install langgraph langchain langchain-openai

配置 API 凭据：

import os
from langchain_openai import ChatOpenAI

# Configure your API Key
os.environ["OPENAI_API_KEY"] = "your-api-key-here"

# Initialize the model
llm = ChatOpenAI(model="gpt-4", temperature=0)

2.定义应用程序状态

LangGraph 使用TypedDict定义图节点之间共享的状态：

from typing import TypedDict, Annotated
from operator import add

class AgentState(TypedDict):
    """State shared between graph nodes"""
    messages: Annotated[list, add]
    user_input: str
    response: str
    next_step: str

该状态存储

• 消息：交换信息的历史
• user_input：当前用户输入
• 响应：由代理生成的响应
• next_step：下一个要执行的操作

3.创建图形节点

3.1 - 输入处理节点

该节点处理用户输入并准备上下文：

def process_input(state: AgentState) -> AgentState:
    """Processes user input"""
    user_message = state["user_input"]
    
    # Add message to history
    state["messages"].append({
        "role": "user",
        "content": user_message
    })
    
    # Define next step
    state["next_step"] = "analyze"
    
    return state

3.2 - 分析和决策节点

该节点使用 LLM 分析输入并决定下一步操作：

from langchain.prompts import ChatPromptTemplate

def analyze_request(state: AgentState) -> AgentState:
    """Analyzes the request and decides the next action"""
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are an intelligent assistant. Analyze the user's request and determine the best way to respond."),
        ("user", "{input}")
    ])
    
    chain = prompt | llm
    
    result = chain.invoke({
        "input": state["user_input"]
    })
    
    state["response"] = result.content
    state["next_step"] = "respond"
    
    return state

3.3 - 响应节点

该节点格式化并返回最终响应：

def generate_response(state: AgentState) -> AgentState:
    """Generates the final response"""
    
    # Add response to history
    state["messages"].append({
        "role": "assistant",
        "content": state["response"]
    })
    
    state["next_step"] = "END"
    
    return state

4.配置状态图

4.1 - 创建图表

现在，让我们将所有节点连接到状态图中：

from langgraph.graph import StateGraph, END

# Create the graph
workflow = StateGraph(AgentState)

# Add nodes
workflow.add_node("process_input", process_input)
workflow.add_node("analyze", analyze_request)
workflow.add_node("respond", generate_response)

# Define entry point
workflow.set_entry_point("process_input")

# Add transitions (edges)
workflow.add_edge("process_input", "analyze")
workflow.add_edge("analyze", "respond")
workflow.add_edge("respond", END)

# Compile the graph
app = workflow.compile()

4.2 - 图表可视化

LangGraph 允许您将图结构可视化：

from IPython.display import Image, display

try:
    display(Image(app.get_graph().draw_mermaid_png()))
except Exception:
    print("Graph visualization requires additional dependencies")

5.运行代理

5.1 - 执行简单查询

def run_agent(user_input: str):
    """Runs the agent with user input"""
    
    # Initial state
    initial_state = {
        "messages": [],
        "user_input": user_input,
        "response": "",
        "next_step": ""
    }
    
    # Execute the graph
    result = app.invoke(initial_state)
    
    return result["response"]

# Test the agent
response = run_agent("What is the capital of France?")
print(f"Response: {response}")

预期输出：

Response: The capital of France is Paris.

5.2 - 使用流式执行

对于交互式应用程序，可以使用流：

async def run_agent_stream(user_input: str):
    """Runs the agent with streaming"""
    
    initial_state = {
        "messages": [],
        "user_input": user_input,
        "response": "",
        "next_step": ""
    }
    
    async for event in app.astream(initial_state):
        for node_name, node_state in event.items():
            print(f"\n--- {node_name} ---")
            if "response" in node_state and node_state["response"]:
                print(f"Partial response: {node_state['response']}")

高级功能

检查点和持久性

LangGraph 支持检查点来保存状态：

from langgraph.checkpoint.memory import MemorySaver

# Add memory to the graph
memory = MemorySaver()
app_with_memory = workflow.compile(checkpointer=memory)

# Execute with persistence
config = {"configurable": {"thread_id": "user-123"}}
result = app_with_memory.invoke(initial_state, config)

条件和动态路由

您可以为路由添加条件逻辑：

def router(state: AgentState) -> str:
    """Determines the next node based on state"""
    
    if "urgent" in state["user_input"].lower():
        return "priority_handler"
    else:
        return "normal_handler"

# Add conditional routing
workflow.add_conditional_edges(
    "analyze",
    router,
    {
        "priority_handler": "priority_node",
        "normal_handler": "normal_node"
    }
)

使用案例

LangGraph 适用于

1. 复杂的聊天机器人：利用上下文管理多轮对话
2. 自主代理：创建基于状态决策的代理
3. 处理工作流：协调数据处理管道
4. 多代理系统：协调多个专业代理

实际应用

欲了解更多详情和实际案例，请访问

• LangGraph 官方文档
• GitHub 示例库

结论

LangGraph 为构建有状态的人工智能应用程序提供了一种强大而灵活的方法。通过将状态图与 LLM 相结合，您可以创建复杂的系统来管理复杂的对话、根据上下文做出决策并执行动态工作流。

LangGraph 的模块化结构允许您从简单的聊天机器人扩展到复杂的多代理系统，同时保持代码的有序性和可维护性。

谢谢