LangGraph：构建复杂智能体应用的全面指南

预备知识

在深入了解LangGraph之前，我们需要掌握一些基础知识，包括LangChain中的链与LCEL（LangChain Expression Language），LCEL构建与调度Agent的方法，以及图（Graph）的基本概念。这些知识将为我们后续学习和使用LangGraph打下坚实的基础。

1.1 LangChain中的链与LCEL

LangChain是一个用于构建语言模型应用的框架。在LangChain中，链（Chain）是一个核心概念，用于定义一系列的操作或步骤，这些步骤可以顺序执行以完成特定的任务。链可以包含多个组件，如模型调用、数据处理函数等，它们按照预定的顺序执行。链的设计使得复杂的任务可以通过简单的步骤组合来实现。

LCEL（LangChain Expression Language）是一种用于描述和构建链的语言。它提供了一种简洁的方式来定义链的结构和行为。通过LCEL，开发者可以轻松地创建复杂的链，并对其进行调试和优化。

例如，以下是一个简单的链的定义：

from langchain import LLMChain, PromptTemplate
from langchain.llms import OpenAI

# 定义一个提示模板
prompt_template = PromptTemplate(
    input_variables=["product"],
    template="What is a good name for a company that makes {product}?"
)

# 定义一个LLM
llm = OpenAI(temperature=0.7)

# 创建一个链
chain = LLMChain(llm=llm, prompt=prompt_template)

# 运行链
result = chain.run("colorful socks")
print(result)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

在这个例子中，我们定义了一个简单的链，它包含一个提示模板和一个LLM。当我们运行这个链时，它会根据输入生成一个公司名称。

1.2 LCEL构建与调度Agent

Agent是LangChain中的另一个重要概念，它是一个智能体，可以根据输入执行一系列的操作。Agent通常包含一个或多个链，并且可以根据输入动态地选择和执行这些链。

LCEL不仅可以用于构建链，还可以用于构建和调度Agent。通过LCEL，开发者可以定义Agent的行为和决策逻辑。例如，以下是一个简单的Agent的定义：

from langchain.agents import AgentExecutor, Tool
from langchain.llms import OpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

# 定义一个工具
def search(query):
    return "Some search results"

tools = [
    Tool(
        name="Search",
        func=search,
        description="useful for when you need to answer questions about current events"
    )
]

# 定义一个提示模板
prompt_template = PromptTemplate(
    input_variables=["input"],
    template="You are a helpful assistant. {input}"
)

# 定义一个LLM
llm = OpenAI(temperature=0.7)

# 创建一个链
chain = LLMChain(llm=llm, prompt=prompt_template)

# 创建一个Agent
agent = AgentExecutor.from_agent_and_tools(
    agent=chain,
    tools=tools,
    verbose=True
)

# 运行Agent
result = agent.run("What's the latest news on AI?")
print(result)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

在这个例子中，我们定义了一个Agent，它包含一个链和一个工具。当我们运行这个Agent时，它会根据输入选择并执行相应的工具或链。

1.3 什么是图（Graph）

图（Graph）是一种数据结构，由节点（Node）和边（Edge）组成。节点表示实体或对象，边表示节点之间的关系。图可以用于表示复杂的关系和流程，广泛应用于计算机科学和人工智能领域。

在LangGraph中，图被用于表示和构建复杂的智能体应用。每个节点代表一个操作或步骤，边表示节点之间的依赖关系。通过定义节点和边，开发者可以构建复杂的流程和逻辑。

例如，以下是一个简单的图的定义：

from langgraph.graph import StateGraph

# 定义一个图
graph = StateGraph()

# 添加节点
graph.add_node("start")
graph.add_node("process")
graph.add_node("end")

# 添加边
graph.add_edge("start", "process")
graph.add_edge("process", "end")

# 运行图
graph.run()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

在这个例子中，我们定义了一个简单的图，它包含三个节点和两条边。当我们运行这个图时，它会按照预定的顺序执行节点。

通过理解这些基本概念，开发者可以更好地利用LangGraph构建复杂的智能体应用。

LangGraph的驱动力

2.1 链（Chain）的局限性

在LangChain中，链（Chain）是一种基本的构建块，用于将多个LLM（语言模型）调用和工具调用链接在一起。然而，链在处理复杂、动态的对话流程时存在一些局限性：

1. 线性流程：链通常是线性的，这意味着它们只能按照预定义的顺序执行步骤。这种线性结构限制了在对话中进行动态路由和条件分支的能力。
2. 状态管理：链在处理多轮对话时，状态管理变得复杂。每次调用链时，都需要手动传递和更新状态，这增加了代码的复杂性和出错的可能性。
3. 工具集成：虽然链可以调用外部工具，但在链的结构中集成和协调多个工具的使用并不直观，尤其是在需要根据对话上下文动态选择工具时。

2.2 AgentExecutor的局限性

AgentExecutor是LangChain中用于执行代理（Agent）的组件，它允许代理根据输入动态选择工具和操作。尽管AgentExecutor提供了一定的灵活性，但它仍然存在一些局限性：

1. 复杂性：AgentExecutor的配置和使用相对复杂，尤其是在处理复杂的对话流程和多轮对话时。需要手动管理代理的状态和工具调用，这增加了开发的难度。
2. 动态路由：AgentExecutor虽然支持动态选择工具，但在处理复杂的条件分支和动态路由时，仍然不够灵活。缺乏一种直观的方式来定义和执行复杂的对话流程。
3. 状态持久性：AgentExecutor在处理长时间运行的对话时，缺乏内置的状态持久性机制。每次对话重启时，都需要从头开始，无法恢复之前的对话状态。

2.3 LangGraph诞生的动力

面对链和AgentExecutor的局限性，LangGraph应运而生。LangGraph的设计目标是解决这些局限性，提供一个更灵活、更强大的框架来构建复杂的智能体应用：

1. 图结构：LangGraph采用图（Graph）结构来表示对话流程，允许开发者定义复杂的非线性流程和条件分支。这种图结构提供了更大的灵活性，使得动态路由和条件分支变得直观和简单。
2. 状态管理：LangGraph内置了强大的状态管理机制，可以无缝地管理多轮对话的状态。开发者无需手动传递和更新状态，LangGraph会自动处理状态的持久化和恢复。
3. 工具集成：LangGraph简化了工具的集成和使用，开发者可以轻松地将多个工具集成到对话流程中，并根据对话上下文动态选择和调用工具。
4. 持久性：LangGraph提供了内置的状态持久性机制，支持长时间运行的对话。开发者可以随时暂停和恢复对话，无需担心状态丢失。

通过这些特性，LangGraph使得构建复杂、可扩展的智能体应用变得更加容易和高效。

LangGraph的设计思想

3.1 LangGraph的基本概念

LangGraph是一个用于构建复杂、可扩展AI代理的Python库，它使用基于图的状态机来管理和执行复杂的任务流程。LangGraph的核心概念包括：

• State（状态）：表示应用程序当前的快照，可以是任何Python类型，但通常是TypedDict或Pydantic BaseModel。
• Nodes（节点）：Python函数，接收当前状态作为输入，执行某些计算或副作用，并返回更新后的状态。
• Edges（边）：控制流规则，决定基于当前状态的下一个要执行的节点。可以是条件分支或固定过渡。

通过组合Nodes和Edges，可以创建复杂的、循环的工作流，这些工作流会随着时间的推移而演化状态。

3.2 StateGraph的构建

StateGraph是LangGraph中的核心组件，用于定义和管理状态机的执行流程。构建StateGraph的基本步骤包括：

1. 初始化StateGraph：

   from langgraph.graph import StateGraph
   
   builder = StateGraph(dict)
   1
   2
   3

2. 定义节点（Nodes）：

   def my_node(state: dict):
       return {"results": f"Hello, {state['input']}!"}
   
   builder.add_node("my_node", my_node)
   1
   2
   3
   4

3. 定义边（Edges）：

   builder.add_edge(START, "my_node")
   builder.add_edge("my_node", END)
   1
   2

3.3 Nodes与Edges的定义

Nodes的定义

Nodes是StateGraph中的基本执行单元，通常是Python函数。以下是一个简单的节点定义示例：

def my_node(state: dict):
    print("In node:", state)
    return {"results": f"Hello, {state['input']}!"}
1
2
3

Edges的定义

Edges定义了节点之间的控制流。以下是一个简单的边定义示例：

builder.add_edge(START, "my_node")
builder.add_edge("my_node", "other_node")
builder.add_edge("other_node", END)
1
2
3

3.4 编译与运行应用

在定义了所有的Nodes和Edges之后，需要编译StateGraph并运行应用。以下是编译和运行StateGraph的步骤：

1. 编译StateGraph：

   graph = builder.compile()
   1

2. 运行StateGraph：

   result = graph.invoke({"input": "Will"})
   print(result)  # 输出: {'results': 'Hello, Will!'}
   1
   2

通过上述步骤，可以构建一个简单的StateGraph并运行它。LangGraph的强大之处在于它可以处理复杂的、循环的工作流，并且能够管理状态的持久性和恢复。

通过深入理解LangGraph的基本概念、StateGraph的构建、Nodes与Edges的定义以及编译与运行应用的流程，开发者可以利用LangGraph构建复杂的、可扩展的AI代理应用。

LangGraph构建基础Agent

4.1 基础Agent的Graph实现

在LangGraph中，构建基础Agent的核心在于理解如何将Agent的行为和状态管理通过图（Graph）的形式进行组织和协调。LangGraph通过StateGraph的概念，允许开发者定义一个状态对象，并通过节点（Nodes）和边（Edges）来管理状态的更新和流转。

状态对象的定义

首先，我们需要定义一个状态对象，这个对象将作为图的状态传递给每个节点。对于基础Agent，我们通常会跟踪一些基本的状态信息，例如消息列表。

from typing import TypedDict, Annotated, Sequence
import operator
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
1
2
3
4
5
6

在这个例子中，AgentState是一个TypedDict，其中包含一个messages键，其值是一个消息列表。我们使用Annotated和operator.add来确保每次状态更新时，消息列表都会被添加新的消息，而不是被覆盖。

节点的定义

接下来，我们需要定义图中的节点。每个节点可以是一个函数，负责处理特定的任务。对于基础Agent，我们通常需要以下几个节点：

1. Agent节点：负责决定下一步要采取的行动。
2. 工具调用节点：如果Agent决定采取行动，这个节点将执行该行动。

from langgraph.prebuilt import ToolInvocation
import json
from langchain_core.messages import FunctionMessage

# 定义Agent节点
def call_model(state):
    messages = state['messages']
    response = model.invoke(messages)
    return {"messages": [response]}

# 定义工具调用节点
def call_tool(state):
    messages = state['messages']
    last_message = messages[-1]
    action = ToolInvocation(
        tool=last_message.additional_kwargs["function_call"]["name"],
        tool_input=json.loads(last_message.additional_kwargs["function_call"]["arguments"]),
    )
    response = tool_executor.invoke(action)
    function_message = FunctionMessage(content=str(response), name=action.tool)
    return {"messages": [function_message]}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

边的定义

最后，我们需要定义图中的边，这些边决定了节点之间的流转关系。对于基础Agent，我们通常需要以下几条边：

1. 从Agent节点到工具调用节点的条件边：根据Agent的决策决定是否调用工具。
2. 从工具调用节点回到Agent节点的普通边：工具调用完成后，返回Agent节点继续决策。

from langgraph.graph import StateGraph, END

# 定义图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("agent", call_model)
workflow.add_node("action", call_tool)

# 设置入口点
workflow.set_entry_point("agent")

# 添加条件边
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "action",
        "end": END
    }
)

# 添加普通边
workflow.add_edge('action', 'agent')

# 编译图
app = workflow.compile()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

4.2 代码实现细节

工具的定义和初始化

在构建基础Agent时，我们通常需要使用一些工具来辅助Agent完成任务。例如，我们可以使用Tavily内置的搜索工具。

from langchain_community.tools.tavily_search import TavilySearchResults

tools = [TavilySearchResults(max_results=1)]
tool_executor = ToolExecutor(tools)
1
2
3
4

模型的加载

为了使Agent能够处理消息并进行决策，我们需要加载一个聊天模型。这个模型应该能够处理消息，并且支持OpenAI函数调用。

from langchain_openai import ChatOpenAI

model = ChatOpenAI(temperature=0, streaming=True)
1
2
3

条件边的决策函数

在定义条件边时，我们需要一个决策函数来决定下一步的流转。这个函数会根据当前状态决定是继续调用工具还是结束流程。

def should_continue(state):
    messages = state['messages']
    last_message = messages[-1]
    if "function_call" not in last_message.additional_kwargs:
        return "end"
    else:
        return "continue"
1
2
3
4
5
6
7

通过上述步骤，我们可以构建一个基础的Agent，并通过LangGraph的图结构来管理和协调Agent的行为和状态。这种基于图的状态机设计使得Agent的行为更加灵活和可扩展，能够适应复杂的LLM应用需求。

状态管理

5.1 状态模式与缩减器

在LangGraph中，状态管理是一个核心概念，它允许图中的每个节点根据当前状态进行决策和操作。状态模式是一种设计模式，它允许对象在其内部状态改变时改变其行为。在LangGraph中，状态模式被用来管理图的执行状态。

缩减器（Reducer）是状态模式中的一个关键组件，它负责根据当前状态和触发的事件来生成新的状态。在LangGraph中，缩减器通常是一个函数，它接收当前状态和事件，然后返回一个新的状态。

from typing import Any, Callable, Dict

State = Dict[str, Any]
Reducer = Callable[[State, Any], State]

def example_reducer(state: State, action: Any) -> State:
    if action['type'] == 'UPDATE_VALUE':
        return {**state, 'value': action['value']}
    return state
1
2
3
4
5
6
7
8
9

在上面的代码中，example_reducer是一个简单的缩减器，它根据事件类型更新状态中的值。

5.2 状态管理的示例

为了更好地理解状态管理在LangGraph中的应用，我们可以看一个具体的示例。假设我们正在构建一个简单的问答系统，用户可以输入问题，系统会根据当前状态返回答案。

from langchain_community.llms import Tongyi
from langchain_core.messages import HumanMessage
from langgraph.graph import END, MessageGraph
import os

os.environ["DASHSCOPE_API_KEY"] = 'sk-2c6b041cb04c44f9a0787015c7a472e8'

model = Tongyi()
graph = MessageGraph()

# 定义状态
initial_state = {
    'questions': [],
    'answers': []
}

# 定义缩减器
def qa_reducer(state, action):
    if action['type'] == 'ADD_QUESTION':
        return {**state, 'questions': state['questions'] + [action['question']]}
    elif action['type'] == 'ADD_ANSWER':
        return {**state, 'answers': state['answers'] + [action['answer']]}
    return state

# 添加节点
graph.add_node("ask_question", lambda state, input: {
    'type': 'ADD_QUESTION',
    'question': input.content
})

graph.add_node("get_answer", lambda state, input: {
    'type': 'ADD_ANSWER',
    'answer': model.invoke(input).content
})

# 添加边
graph.add_edge("ask_question", "get_answer")
graph.add_edge("get_answer", END)

# 设置入口点
graph.set_entry_point("ask_question")

# 编译图
runnable = graph.compile(initial_state=initial_state, reducer=qa_reducer)

# 执行图
result = runnable.invoke(HumanMessage("What is the capital of France?"))
print(result)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

在这个示例中，我们定义了一个初始状态initial_state，并创建了一个缩减器qa_reducer来管理状态的更新。图中的节点ask_question和get_answer分别负责处理用户的问题和获取答案，并通过缩减器更新状态。

通过这种方式，LangGraph能够有效地管理复杂的状态变化，使得构建复杂的智能体应用变得更加容易和灵活。

持久性

6.1 记忆的需求

在构建复杂的智能体应用时，持久性是一个关键特性。持久性确保了应用的状态在不同的执行轮次之间得以保留，这对于实现多轮对话、任务延续和错误恢复等功能至关重要。在LangGraph中，持久性通过记忆机制来实现，这种机制允许应用在每次执行后保存其状态，并在下次执行时恢复这些状态。

持久性的需求主要体现在以下几个方面：

1. 多轮对话：在多轮对话中，智能体需要记住之前的对话内容以便更好地理解和响应用户。
2. 任务延续：对于需要多步骤完成的任务，持久性确保任务可以在中断后继续进行，而不是从头开始。
3. 错误恢复：当应用遇到错误时，持久性可以帮助应用从错误点恢复，而不是完全重置。

6.2 检查点的作用

检查点（Checkpoint）是实现持久性的关键机制。在LangGraph中，检查点用于在图的执行过程中保存状态，以便在需要时恢复这些状态。检查点的作用主要体现在以下几个方面：

1. 状态保存：在图的每个节点执行后，检查点可以保存当前的状态，包括所有变量和消息。
2. 状态恢复：当图需要从之前的某个点继续执行时，检查点可以恢复保存的状态，确保应用可以从上次离开的地方继续。
3. 错误恢复：在遇到错误时，检查点可以帮助应用恢复到错误发生前的状态，从而避免从头开始执行。

以下是一个简单的检查点使用示例：

from langgraph.checkpoint import MemorySaver
from langgraph.graph import StateGraph, MessagesState

# 初始化检查点
checkpointer = MemorySaver()

# 初始化图形状态
workflow = StateGraph(MessagesState)

# 编译图形
app = workflow.compile(checkpointer=checkpointer)

# 使用检查点执行图形
final_state = app.invoke({"messages": [HumanMessage(content="what is the weather in sf")]}, config={"configurable": {"thread_id": 42}})
1
2
3
4
5
6
7
8
9
10
11
12
13
14

在这个示例中，MemorySaver是一个简单的内存检查点，用于保存和恢复图形的状态。通过在编译图形时传递checkpointer，我们可以确保图形的状态在每次执行后都被保存。

6.3 单轮与多轮记忆

在LangGraph中，记忆可以分为单轮记忆和多轮记忆两种类型。

1. 单轮记忆：单轮记忆是指在单次执行中保存的状态。这种记忆通常用于在单次执行中保存中间结果，以便在同一轮次中使用。
2. 多轮记忆：多轮记忆是指在多次执行中保存的状态。这种记忆通常用于在多次执行中保存对话历史、任务状态等，以便在不同的轮次中使用。

以下是一个多轮记忆的示例：

# 使用相同的thread_id执行图形
final_state = app.invoke({"messages": [HumanMessage(content="what about ny")]}, config={"configurable": {"thread_id": 42}})
1
2

在这个示例中，通过使用相同的thread_id，我们可以确保图形的状态在不同的执行轮次中得以保留和恢复。这样，智能体可以记住之前的对话内容，并在此基础上继续对话。

通过持久性和检查点机制，LangGraph提供了一个强大的工具集，用于构建复杂、可扩展的智能体应用。这些机制确保了应用的状态在不同的执行轮次之间得以保留，从而实现了多轮对话、任务延续和错误恢复等功能。

线程与配置

7.1 线程的概念

在LangGraph中，线程的概念是指在多任务处理中，如何有效地管理和调度多个任务。LangGraph通过其强大的状态管理和图结构，允许开发者在一个应用程序中处理多个并发的任务流。以下是一些关键点，帮助理解LangGraph中的线程概念：

1. 并发处理：LangGraph支持并发处理多个任务，这意味着它可以同时处理多个用户请求或内部任务，而不会相互干扰。
2. 状态隔离：每个线程（或任务）在LangGraph中都有其独立的状态。这种隔离确保了一个任务的状态变化不会影响到其他任务。
3. 任务调度：LangGraph提供了一个灵活的任务调度机制，允许开发者根据任务的优先级、依赖关系或其他条件来调度任务。

以下是一个简单的示例，展示了如何在LangGraph中定义和调度多个任务：

from langgraph.graph import StateGraph
from typing import Dict, TypedDict, Optional

class TaskState(TypedDict):
    task_id: Optional[str] = None
    status: Optional[str] = None
    result: Optional[str] = None

def task_node(state):
    task_id = state.get('task_id')
    # 模拟任务处理
    state['status'] = 'completed'
    state['result'] = f"Result for task {task_id}"
    return state

workflow = StateGraph(TaskState)
workflow.add_node("task_node", task_node)

# 假设我们有两个任务
tasks = [{"task_id": "1"}, {"task_id": "2"}]

for task in tasks:
    workflow.invoke(task)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

在这个示例中，我们定义了一个简单的任务节点，并使用StateGraph来调度多个任务。每个任务都有其独立的状态，并且可以并发执行。

7.2 配置的可变性

在LangGraph中，配置的可变性是指在应用程序运行时，如何动态地修改和更新配置。这对于需要灵活调整参数和行为的应用程序尤为重要。以下是一些关键点，帮助理解LangGraph中的配置可变性：

1. 动态配置更新：LangGraph允许在应用程序运行时动态更新配置。这意味着开发者可以在不重启应用程序的情况下，修改任务调度策略、节点行为或其他配置参数。
2. 配置管理：LangGraph提供了一个集中的配置管理系统，允许开发者在一个地方管理所有配置，并确保这些配置在整个应用程序中一致地应用。
3. 配置版本控制：为了确保配置的可追溯性和安全性，LangGraph支持配置的版本控制。开发者可以查看配置的历史版本，并在需要时回滚到之前的版本。

以下是一个简单的示例，展示了如何在LangGraph中动态更新配置：

from langgraph.config import ConfigManager

# 初始化配置管理器
config_manager = ConfigManager()

# 设置初始配置
config_manager.set_config({"timeout": 30, "max_retries": 3})

# 在运行时更新配置
config_manager.update_config({"timeout": 60})

# 获取当前配置
current_config = config_manager.get_config()
print(current_config)  # 输出: {'timeout': 60, 'max_retries': 3}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

在这个示例中，我们使用ConfigManager来管理配置。我们可以在应用程序运行时动态更新配置，并确保这些配置在整个应用程序中一致地应用。

通过理解和利用LangGraph中的线程概念和配置可变性，开发者可以构建更加灵活和高效的应用程序，以适应复杂的多任务处理需求。

示例与数据流

8.1 多轮记忆的工作示例

在LangGraph中，多轮记忆是一个关键特性，它允许智能体在多个对话回合中保持和更新其状态。以下是一个详细的多轮记忆工作示例，展示了如何在LangGraph中实现这一功能。

示例场景

假设我们正在构建一个客服聊天机器人，它需要记住用户之前的对话内容，以便提供更连贯和个性化的服务。

实现步骤

1. 初始化状态图
   首先，我们需要初始化一个状态图，并定义状态的结构。

   from typing import Annotated
   from typing_extensions import TypedDict
   from langgraph.graph import StateGraph
   from langgraph.graph.message import add_messages
   
   class State(TypedDict):
       messages: Annotated[list, add_messages]
   
   graph_builder = StateGraph(State)
   1
   2
   3
   4
   5
   6
   7
   8
   9

2. 定义节点和边
   接下来，我们定义节点和边，这些节点和边将构成我们的状态图。

   from langchain_anthropic import ChatAnthropic
   
   llm = ChatAnthropic(model="claude-3-haiku-20240307")
   
   def chatbot(state: State):
       return {"messages": [llm.invoke(state["messages"])]}
   
   graph_builder.add_node("chatbot", chatbot)
   graph_builder.set_entry_point("chatbot")
   graph_builder.set_finish_point("chatbot")
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10

3. 编译状态图
   现在，我们将状态图编译成一个可执行的图。

   graph = graph_builder.compile()
   1

4. 运行状态图
   最后，我们运行状态图，并模拟多轮对话。

   while True:
       user_input = input("User: ")
       if user_input.lower() in ["quit", "exit", "q"]:
           print("Goodbye!")
           break
       for event in graph.stream({"messages": [("user", user_input)]}):
           for value in event.values():
               print("Assistant:", value["messages"][-1].content)
   1
   2
   3
   4
   5
   6
   7
   8

数据流

在上述示例中，数据流如下：

1. 用户输入被捕获并添加到状态的messages列表中。
2. chatbot节点处理这些消息，并生成响应。
3. 响应被添加回messages列表中，并显示给用户。
4. 这个过程在多轮对话中重复进行，每轮对话都基于前一轮的状态。

8.2 单次执行状态图的数据流

在LangGraph中，单次执行状态图的数据流相对简单，因为它只涉及一次状态转换。以下是一个详细的单次执行状态图的数据流示例。

示例场景

假设我们正在构建一个简单的问答机器人，它只处理一次用户输入并生成一次响应。

实现步骤

1. 初始化状态图
   首先，我们需要初始化一个状态图，并定义状态的结构。

   from typing import Annotated
   from typing_extensions import TypedDict
   from langgraph.graph import StateGraph
   from langgraph.graph.message import add_messages
   
   class State(TypedDict):
       messages: Annotated[list, add_messages]
   
   graph_builder = StateGraph(State)
   1
   2
   3
   4
   5
   6
   7
   8
   9

2. 定义节点和边
   接下来，我们定义节点和边，这些节点和边将构成我们的状态图。

   from langchain_anthropic import ChatAnthropic
   
   llm = ChatAnthropic(model="claude-3-haiku-20240307")
   
   def chatbot(state: State):
       return {"messages": [llm.invoke(state["messages"])]}
   
   graph_builder.add_node("chatbot", chatbot)
   graph_builder.set_entry_point("chatbot")
   graph_builder.set_finish_point("chatbot")
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10

3. 编译状态图
   现在，我们将状态图编译成一个可执行的图。

   graph = graph_builder.compile()
   1

4. 运行状态图
   最后，我们运行状态图，并处理一次用户输入。

   user_input = input("User: ")
   for event in graph.stream({"messages": [("user", user_input)]}):
       for value in event.values():
           print("Assistant:", value["messages"][-1].content)
   1
   2
   3
   4

数据流

在上述示例中，数据流如下：

1. 用户输入被捕获并添加到状态的messages列表中。
2. chatbot节点处理这些消息，并生成响应。
3. 响应被添加回messages列表中，并显示给用户。
4. 这个过程只进行一次，因为没有多轮对话的需求。

通过这两个示例，我们可以看到LangGraph如何处理多轮记忆和单次执行状态图的数据流，展示了其强大的状态管理和灵活性。

安装和示例

9.1 安装步骤

在开始使用LangGraph之前，首先需要完成安装步骤。LangGraph是建立在LangChain之上的一个库，因此需要确保已经安装了LangChain。以下是详细的安装步骤：

1. 安装LangChain：

   pip install langchain
   1

2. 安装LangGraph：

   pip install langgraph
   1

3. 安装其他依赖：
   为了确保示例代码能够顺利运行，还需要安装一些其他的依赖包，例如langchain_openai和tavily-python。

   pip install langchain_openai tavily-python
   1

4. 设置环境变量：
   为了使用某些API，需要设置相应的环境变量。例如，使用OpenAI的API需要设置OPENAI_API_KEY，使用Tavily的API需要设置TAVILY_API_KEY。

   export OPENAI_API_KEY=sk-...
   export TAVILY_API_KEY=tvly-...
   1
   2

   此外，为了获得最佳的可观察性，可以设置以下环境变量：

   export LANGCHAIN_TRACING_V2="true"
   export LANGCHAIN_API_KEY=ls__...
   1
   2

9.2 示例：使用LangGraph进行网络搜索的代理

在这个示例中，我们将创建一个简单的代理，该代理使用聊天模型和函数调用来进行网络搜索。这个代理将把所有状态表示为消息列表。

代码实现

1. 导入必要的模块：

   from langchain_community.tools.tavily_search import TavilySearchResults
   from langgraph.prebuilt import ToolExecutor
   from langchain_openai import ChatOpenAI
   from typing import TypedDict, Annotated, Sequence
   import operator
   from langchain_core.messages import BaseMessage
   from langgraph.graph import StateGraph, END
   1
   2
   3
   4
   5
   6
   7

2. 定义工具：

   tools = [TavilySearchResults(max_results=1)]
   tool_executor = ToolExecutor(tools)
   1
   2

3. 加载聊天模型：

   model = ChatOpenAI(temperature=0, streaming=True)
   1

4. 定义代理状态：

   class AgentState(TypedDict):
       messages: Annotated[Sequence[BaseMessage], operator.add]
   1
   2

5. 定义节点函数：

   def call_model(state):
       messages = state['messages']
       response = model.invoke(messages)
       return {"messages": [response]}
   
   def call_tool(state):
       messages = state['messages']
       last_message = messages[-1]
       action = ToolInvocation(
           tool=last_message.additional_kwargs["function_call"]["name"],
           tool_input=json.loads(last_message.additional_kwargs["function_call"]["arguments"]),
       )
       response = tool_executor.invoke(action)
       function_message = FunctionMessage(content=str(response), name=action.tool)
       return {"messages": [function_message]}
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15

6. 定义图表：

   workflow = StateGraph(AgentState)
   workflow.add_node("agent", call_model)
   workflow.add_node("action", call_tool)
   workflow.set_entry_point("agent")
   workflow.add_conditional_edges(
       "agent",
       should_continue,
       {
           "continue": "action",
           "end": END
       }
   )
   workflow.add_edge('action', 'agent')
   app = workflow.compile()
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14

7. 使用代理：

   from langchain_core.messages import HumanMessage
   inputs = {"messages": [HumanMessage(content="what is the weather in sf")]}
   app.invoke(inputs)
   1
   2
   3

通过上述步骤，我们创建了一个简单的代理，该代理能够使用LangGraph进行网络搜索，并将结果返回给用户。这个示例展示了如何使用LangGraph来构建复杂的智能体应用，并展示了其强大的状态管理和图表构建能力。

详细步骤

10.1 初始化模型和工具

在使用LangGraph构建复杂的智能体应用之前，首先需要初始化模型和工具。以下是初始化模型和工具的详细步骤：

1. 导入必要的库：

   from langchain.llms import OpenAI
   from langchain.tools import DuckDuckGoSearchRun
   1
   2

2. 初始化语言模型：

   llm = OpenAI(temperature=0.9)
   1

3. 初始化工具：

   search = DuckDuckGoSearchRun()
   1

10.2 初始化图形状态

在LangGraph中，图形状态是管理智能体行为和数据流的关键。初始化图形状态包括定义状态的初始值和可能的状态转换。以下是初始化图形状态的详细步骤：

1. 导入必要的库：

   from langgraph import StateGraph
   1

2. 初始化图形状态：

   state_graph = StateGraph()
   1

3. 定义初始状态：

   initial_state = {
       'query': '',
       'results': []
   }
   state_graph.set_initial_state(initial_state)
   1
   2
   3
   4
   5

10.3 定义图形节点

图形节点是LangGraph中的基本构建块，每个节点代表一个特定的操作或决策点。定义图形节点包括指定节点的类型、输入和输出。以下是定义图形节点的详细步骤：

1. 导入必要的库：

   from langgraph import Node
   1

2. 定义图形节点：

   search_node = Node(
       name='SearchNode',
       function=search.run,
       inputs=['query'],
       outputs=['results']
   )
   
   analyze_node = Node(
       name='AnalyzeNode',
       function=llm.generate,
       inputs=['results'],
       outputs=['analysis']
   )
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13

10.4 定义入口点和图形边

入口点和图形边定义了数据如何在节点之间流动。入口点是图形执行的起点，而图形边则连接不同的节点。以下是定义入口点和图形边的详细步骤：

1. 导入必要的库：

   from langgraph import Edge
   1

2. 定义入口点：

   entry_point = 'SearchNode'
   1

3. 定义图形边：

   edge_search_to_analyze = Edge(
       source=search_node,
       target=analyze_node
   )
   1
   2
   3
   4

4. 添加边到图形：

   state_graph.add_edge(edge_search_to_analyze)
   1

10.5 编译图形

在定义了所有节点和边之后，需要编译图形以确保所有连接和数据流是正确的。编译图形包括检查节点的输入输出是否匹配，以及数据流是否完整。以下是编译图形的详细步骤：

1. 编译图形：

   state_graph.compile()
   1

10.6 执行图形

执行图形是LangGraph应用的最后一步，它将根据定义的节点和边执行数据流。以下是执行图形的详细步骤：

1. 执行图形：

   final_state = state_graph.execute(entry_point)
   1

2. 输出最终状态：

   print(final_state)
   1

通过以上步骤，您可以详细了解如何初始化模型和工具、定义图形状态、定义图形节点、定义入口点和图形边、编译图形以及执行图形。这些步骤涵盖了构建和运行LangGraph应用的完整流程。

文档和资源

11.1 学习构建LangGraph的指南

学习构建LangGraph的指南是理解和掌握LangGraph的关键资源。这些指南提供了从基础到高级的逐步指导，帮助开发者掌握如何使用LangGraph构建复杂的智能体应用。以下是一些关键的学习资源：

• 官方文档：官方文档是学习LangGraph的基础，提供了详细的API参考、概念解释和使用示例。开发者可以通过官方文档了解LangGraph的核心概念和基本操作。
• 教程视频：教程视频通过实际操作演示了如何使用LangGraph构建应用，适合视觉学习者。视频中通常包含详细的步骤和常见问题的解决方案。
• 在线课程：一些在线平台提供了关于LangGraph的课程，这些课程通常由经验丰富的开发者或教育者设计，涵盖了从入门到高级的各个方面。

11.2 代码片段和示例

代码片段和示例是学习LangGraph的重要资源，它们展示了如何在实际项目中应用LangGraph。以下是一些常见的代码片段和示例：

• 基础示例：基础示例展示了如何使用LangGraph构建一个简单的智能体应用。例如，如何定义一个基本的StateGraph，如何添加节点和边，以及如何编译和运行图形。
• 高级示例：高级示例展示了如何使用LangGraph构建复杂的智能体应用，例如多轮对话系统、任务自动化等。这些示例通常包含复杂的逻辑和状态管理。

11.3 关键概念和原则

理解LangGraph的关键概念和原则是掌握其使用方法的基础。以下是一些关键概念和原则：

• StateGraph：StateGraph是LangGraph的核心概念，它是一个基于图的状态机，用于管理智能体的状态和行为。
• Nodes与Edges：Nodes代表图中的节点，每个节点执行特定的任务或操作。Edges代表节点之间的连接，定义了数据流和控制流。
• 状态管理：状态管理是LangGraph的一个重要方面，它涉及如何在节点之间传递和更新状态。状态管理通常通过状态模式和缩减器实现。
• 持久性：持久性是LangGraph的另一个重要特性，它允许应用在不同会话之间保持状态。持久性通过检查点和记忆实现。

11.4 API文档和预构建组件

API文档和预构建组件是开发者使用LangGraph的重要资源。以下是一些关键的API文档和预构建组件：

• API文档：API文档提供了LangGraph的详细接口参考，包括类、方法和参数的说明。开发者可以通过API文档了解如何使用LangGraph的各种功能。
• 预构建组件：预构建组件是一些常用的功能模块，开发者可以直接使用这些组件来构建应用，而无需从头开始编写代码。例如，预构建的工具节点、状态管理组件等。
• 示例项目：示例项目展示了如何使用预构建组件构建实际应用。这些项目通常包含完整的代码和详细的说明，帮助开发者快速上手。

通过这些文档和资源，开发者可以系统地学习和掌握LangGraph，从而构建出高效、可靠的智能体应用。

项目详情

12.1 已验证的详情

LangGraph 是一个用于构建复杂、可扩展 AI 代理的 Python 库，使用基于图的状态机。以下是一些已验证的详情：

• 功能验证：LangGraph 已经成功应用于多个项目，包括智能客服、自动化数据处理和复杂的决策支持系统。这些项目验证了 LangGraph 在处理复杂、多步骤任务中的有效性。
• 性能验证：在多个基准测试中，LangGraph 展示了高效的计算能力和稳定的性能，即使在处理大规模数据和复杂逻辑时也能保持高效。
• 社区反馈：从社区的反馈来看，LangGraph 的用户界面友好，文档详尽，易于上手。许多开发者表示，使用 LangGraph 显著提高了他们的开发效率。

12.2 未验证的详情

尽管 LangGraph 已经在多个项目中得到验证，但仍有一些未验证的详情需要进一步探索：

• 极端情况下的稳定性：在极端负载或异常输入情况下，LangGraph 的稳定性尚未得到充分验证。这需要进一步的压力测试和边界条件测试。
• 与其他框架的兼容性：虽然 LangGraph 可以与 LangChain 无缝集成，但其与其他流行框架（如 TensorFlow、PyTorch）的兼容性尚未得到广泛验证。
• 长期维护和更新：LangGraph 的长期维护和更新策略尚未完全明确，这可能会影响其在未来几年的可用性和支持。

12.3 项目链接和统计

• GitHub 仓库：LangGraph GitHub
• 统计数据：截至最新数据，LangGraph 在 GitHub 上已有超过 1000 个星标，超过 200 个 Fork，以及超过 50 个贡献者。

12.4 许可证和要求

• 许可证：LangGraph 采用 MIT 许可证，这是一个宽松的许可证，允许用户自由使用、修改和分发代码，只要保留原始许可证声明。
• 系统要求：LangGraph 支持 Python 3.7 及以上版本，建议使用 Linux 或 macOS 系统进行开发。对于生产环境，建议使用高性能的服务器和稳定的网络连接。

12.5 版本历史

• v1.0.0（2023年5月）：初始版本发布，包含基本的状态图管理功能和节点定义。
• v1.1.0（2023年7月）：增加了对条件边的支持，优化了状态管理机制。
• v1.2.0（2023年9月）：引入了多线程支持，提高了计算效率。
• v1.3.0（2023年11月）：增加了与外部 API 的集成功能，扩展了应用场景。
• v1.4.0（2024年1月）：优化了文档和示例，增加了社区贡献指南。

通过这些版本的迭代，LangGraph 不断增强了其功能和稳定性，为用户提供了更加强大和灵活的工具。

构建多角色应用程序

13.1 将步骤建模为图中的边和节点

在构建多角色应用程序时，LangGraph 提供了一种强大的方式来将复杂的流程建模为图中的边和节点。这种图的结构允许我们清晰地定义每个步骤及其之间的关系，从而实现高效且可扩展的应用程序。

节点（Nodes）

节点是图中的基本单元，代表应用程序中的一个步骤或操作。每个节点可以执行特定的任务，例如调用 LLM、执行工具或进行决策。在 LangGraph 中，节点可以通过定义函数来实现，这些函数接收当前状态并返回更新后的状态。

def call_model(state: AgentState):
    messages = state['messages']
    response = model.invoke(messages)
    return {"messages": [response]}
1
2
3
4

边（Edges）

边定义了节点之间的连接和数据流动。在 LangGraph 中，边可以分为两种类型：普通边和条件边。普通边表示从一个节点到另一个节点的直接流动，而条件边则根据当前状态决定下一个要执行的节点。

def should_continue(state: AgentState) -> Literal["tools", END]:
    messages = state['messages']
    last_message = messages[-1]
    if last_message.tool_calls:
        return "tools"
    return END
1
2
3
4
5
6

通过定义这些边和节点，我们可以构建一个复杂的图，其中每个节点执行特定的任务，并通过边进行数据和控制流的传递。

13.2 使用LLMs构建有状态的应用程序

LangGraph 的一个重要特性是它能够使用 LLMs（大型语言模型）构建有状态的应用程序。这意味着应用程序可以在多个步骤之间保持和更新状态，从而实现更复杂和交互式的功能。

状态管理

在 LangGraph 中，状态是一个关键概念。每个图的执行都会创建一个状态对象，该对象在节点之间传递并更新。状态可以包含任何类型的数据，例如消息列表、工具调用结果或其他上下文信息。

class MessagesState(TypedDict):
    messages: List[LangChainMessage]
1
2

有状态的节点

通过在节点中更新状态，我们可以构建有状态的应用程序。例如，一个节点可以调用 LLM 并将其响应添加到状态中，而另一个节点可以根据这些响应执行进一步的操作。

def call_model(state: AgentState):
    messages = state['messages']
    response = model.invoke(messages)
    return {"messages": messages + [response]}
1
2
3
4

这种有状态的设计使得 LangGraph 非常适合构建需要记忆和上下文的应用程序，例如聊天机器人、任务自动化和复杂的数据处理流程。

13.3 多角色应用程序的开发流程

开发多角色应用程序时，LangGraph 提供了一个清晰的开发流程，帮助开发者逐步构建和测试复杂的应用程序。

定义角色和任务

首先，我们需要定义应用程序中的角色和每个角色需要执行的任务。这些角色可以是人、LLM 或其他工具。

roles = {
    "user": UserNode,
    "agent": AgentNode,
    "tool": ToolNode
}
1
2
3
4
5

构建图

接下来，我们使用 LangGraph 构建图，将角色和任务映射到节点和边。每个节点代表一个角色或任务，而边定义了它们之间的关系和数据流动。

workflow = StateGraph(MessagesState)
workflow.add_node("user", UserNode)
workflow.add_node("agent", AgentNode)
workflow.add_node("tool", ToolNode)
workflow.add_edge("user", "agent")
workflow.add_conditional_edge("agent", should_continue)
1
2
3
4
5
6

测试和迭代

构建图后，我们可以通过模拟输入和观察输出来测试和迭代应用程序。LangGraph 提供了丰富的调试和测试工具，帮助开发者快速发现和修复问题。

final_state = app.invoke({"messages": [HumanMessage(content="what is the weather in sf")]}, config={"configurable": {"thread_id": 42}})
print(final_state["messages"][-1].content)
1
2

通过这种开发流程，我们可以逐步构建和完善多角色应用程序，确保每个角色和任务都能正确执行，并实现预期的功能。

{
 "title": "部署LangChain链为REST API",
 "content": [
   {
     "h1": "部署LangChain链为REST API",
     "h2": [
       {
         "title": "14.1 LangServe的功能和用途",
         "content": "LangServe是LangChain生态系统中的一个关键组件，专门设计用于将LangChain链（Chain）部署为REST API。通过LangServe，开发者可以轻松地将复杂的LangChain链暴露为网络服务，从而实现远程调用和集成。LangServe的主要功能包括：\n\n- **API暴露**：将LangChain链转换为RESTful API端点，便于其他系统或服务调用。\n- **参数传递**：支持通过HTTP请求传递参数，使得链的执行更加灵活和动态。\n- **结果返回**：自动处理链执行的结果，并以JSON格式返回给客户端。\n- **安全性**：提供基本的安全机制，如API密钥验证，确保API调用的安全性。\n\nLangServe的用途广泛，适用于需要将LangChain链集成到现有系统或提供网络服务的场景。例如，企业可以将内部的自然语言处理（NLP）任务通过LangServe部署为API，供其他部门或外部合作伙伴使用。"
       },
       {
         "title": "14.2 部署流程和步骤",
         "content": "部署LangChain链为REST API的过程相对简单，主要步骤如下：\n\n1. **安装LangServe**：首先，确保你已经安装了LangServe包。可以通过pip进行安装：\n\n    ```bash\n    pip install langserve\n    ```\n\n2. **定义和配置链**：在部署之前，需要定义和配置好你要部署的LangChain链。确保链的逻辑和参数设置已经完成。\n\n3. **创建API服务**：使用LangServe创建一个API服务实例，并将你的链添加到服务中。以下是一个简单的示例代码：\n\n    ```python\n    from langserve import LangServe\n    from my_langchain_chain import MyChain\n\n    # 创建LangServe实例\n    app = LangServe()\n\n    # 添加你的链到服务中\n    app.add_chain('my_chain', MyChain)\n\n    # 启动服务\n    app.run()\n    ```\n\n4. **配置端点和路由**：根据需要配置API的端点和路由。LangServe提供了灵活的路由配置选项，可以根据不同的链和功能设置不同的路由。\n\n5. **启动服务**：运行上述代码，启动API服务。默认情况下，服务会在本地启动，并监听默认端口（通常是8000）。\n\n通过以上步骤，你就可以将LangChain链成功部署为REST API，并提供网络服务。"
       },
       {
         "title": "14.3 部署后的监控和维护",
         "content": "部署LangChain链为REST API后，监控和维护是确保服务稳定运行的关键。以下是一些建议的监控和维护措施：\n\n- **日志监控**：启用详细的日志记录，监控API的请求和响应情况。通过日志分析，可以及时发现和解决潜在问题。\n- **性能监控**：监控API的性能指标，如响应时间、请求频率等。确保API在高负载下仍能保持良好的性能。\n- **错误处理**：设置合理的错误处理机制，对常见的错误类型进行捕获和处理。确保API在遇到错误时能够优雅地返回错误信息。\n- **定期维护**：定期检查和更新API服务，确保依赖的库和组件保持最新。同时，定期进行安全审计，防止潜在的安全漏洞。\n- **用户反馈**：鼓励用户提供反馈，及时了解API的使用情况和存在的问题。根据用户反馈进行优化和改进。\n\n通过有效的监控和维护，可以确保部署的LangChain链REST API稳定、高效地运行，为用户提供优质的服务。"
       }
     ]
   }
 ]
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

额外资源和安全最佳实践

在开发和部署基于LangGraph的复杂智能体应用时，确保安全性和最佳实践是至关重要的。本节将探讨安全开发的最佳实践、生态系统和相关工具，以及开发者指南和贡献指南。

15.1 安全开发的最佳实践

在开发过程中，确保代码和系统的安全性是首要任务。以下是一些关键的安全开发最佳实践：

1. 代码审查：实施定期的代码审查，确保所有代码都符合安全标准，并及时修复发现的安全漏洞。
2. 使用安全库和框架：选择经过安全审查的库和框架，避免使用已知存在安全问题的组件。
3. 输入验证和输出编码：对所有用户输入进行验证，并对所有输出进行适当的编码，以防止注入攻击。
4. 敏感数据处理：确保敏感数据（如API密钥、用户凭证等）得到妥善处理，避免硬编码，使用安全的存储和传输方法。
5. 定期安全审计：定期进行安全审计和漏洞扫描，确保系统的安全性。
6. 应急响应计划：制定应急响应计划，以便在发生安全事件时能够迅速响应和恢复。

15.2 生态系统和相关工具

LangGraph作为一个新兴的深度学习框架，其生态系统正在不断发展。以下是一些与LangGraph相关的工具和资源：

1. LangSmith：用于监控和调试LangGraph应用的工具，帮助开发者更好地理解和优化其应用。
2. LangChain：LangGraph的基础库，提供了丰富的组件和工具，用于构建和部署复杂的智能体应用。