AI 智能体开发框架 LangGraph

简介

LangGraph 是一个使用 LLM 构建有状态、多参与者应用程序的库，用于创建代理和多代理工作流。LangGraph 允许定义涉及循环的流程。它提供了对应用程序的流程和状态的细粒度控制。

langchain langgraph

LangGraph 架构设计

Google Pregel: 大规模图处理系统，支持并行
Apache Beam：进行批量和流式数据处理的最简单方法。一次编写，随处运行，适用于关键任务生产工作负载的数据处理。
NetworkX 是一个 Python 包，用于创建、操作和研究复杂网络的结构、动态和功能

agent multi

LangGraph 特点

循环和分支：在您的应用中实现循环和条件。
持久性：自动保存状态。随时暂停和恢复执行，支持错误恢复、人机循环工作流、时间旅行等特性。
流式传输支持：流式传输每个节点生成的输出（包括令牌流式传输）。
与 LangChain 集成：LangGraph 与 LangChain 无缝集成

安装

pip install -U langgraph

官方示例

from typing import Literal

import nest_asyncio
from IPython.display import Image, display
from langchain_core.messages import HumanMessage
from langchain_core.runnables.graph import MermaidDrawMethod
from langchain_core.tools import tool
from langchain_ollama import ChatOllama
from langgraph.graph import END, START, StateGraph, MessagesState
from langgraph.prebuilt import ToolNode


# 定义工具
@tool
def search(city: str):
    """
    天气查询
    """
    if city.lower() in ['北京', 'beijing']:
        return '北京现在是12度'
    return "现在是多云"


# 创建工具列表
tools = [search]

# 创建工具节点，用于自动执行工具调用
tool_node = ToolNode(tools)

# 初始化模型，本地或者OpenAI
model = ChatOllama(model="qwen2.5", temperature=0).bind_tools(tools)


# model = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(tools)


# 分支节点，代表有分叉的路由
def should_continue(state: MessagesState) -> Literal["tools", END]:
    messages = state['messages']
    last_message = messages[-1]
    # If the LLM makes a tool call, then we route to the "tools" node
    if last_message.tool_calls:
        return "tools"
    # Otherwise, we stop (reply to the user)
    return END


# 调用大模型
def call_model(state: MessagesState):
    messages = state['messages']
    response = model.invoke(messages)
    # We return a list, because this will get added to the existing list
    return {"messages": [response]}


# 初始化图对象
workflow = StateGraph(MessagesState)

# 创建节点，每个节点对应一个处理函数
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)

# 开始节点 START，下一步是agent节点
workflow.add_edge(START, "agent")

# 多条边，从agent开始，根据条件路由到不同的下一步节点。
workflow.add_conditional_edges(
    # First, we define the start node. We use `agent`.
    # This means these are the edges taken after the `agent` node is called.
    "agent",
    # Next, we pass in the function that will determine which node is called next.
    should_continue,
)

# We now add a normal edge from `tools` to `agent`.
# This means that after `tools` is called, `agent` node is called next.
workflow.add_edge("tools", 'agent')

# Finally, we compile it!
# This compiles it into a LangChain Runnable,
# meaning you can use it as you would any other runnable.
# Note that we're (optionally) passing the memory when compiling the graph
app = workflow.compile()
print(app)
app.get_graph().print_ascii()
"""
        +-----------+
        | __start__ |
        +-----------+
              *
              *
              *
          +-------+
          | agent |
          +-------+
         *         .
       **           ..
      *               .
+-------+         +---------+
| tools |         | __end__ |
+-------+         +---------+
目前北京的温度是12度。请注意保暖哦！
"""


def test_graph_render():
    nest_asyncio.apply()  # Required for Jupyter Notebook to run async functions

    app.get_graph().print_ascii()
    # display(Image(app.get_graph().draw_png()))
    display(Image(app.get_graph().draw_mermaid_png()))
    # display(Image(app.get_graph().draw_png()))

    display(Image(app.get_graph().draw_mermaid_png(draw_method=MermaidDrawMethod.PYPPETEER)))


def test_graph():
    # Use the Runnable
    final_state = app.invoke(
        {"messages": [HumanMessage(content="北京天气如何")]},
        config={"configurable": {"thread_id": 42}}
    )
    print(final_state["messages"][-1].content)

LangGraph 的核心步骤

图的定义 workflow = StateGraph(MessagesState)
图的节点定义 add_node ToolNode ChatOpenAI
图的边定义 add_edge add_conditional_edges
图的编译 workflow.compile
图的调用 app.invoke

```python{data-line-numbers="2-15"} """ +-----------+ | start | +-----------+ * * +-------+ | agent | +-------+ * . ** .. * . +-------+ +---------+ | tools | | end | +-------+ +---------+ """

<!-- ![LangGraph graph](assets/2024-11-20-03-20-11.png)
{style="flex: 0.5"} -->

---

## LangGraph 的核心是将工作流程建模为图

- **状态 State**：表示应用程序当前快照的共享数据结构。它可以是任何 Python 类型，但通常是 TypedDict 或 Pydantic BaseModel。
- **节点 Nodes**：对代理逻辑进行编码的 Python 函数。它们接收当前状态作为输入，执行一些计算或副作用，并返回更新的状态。
- **边 Edges**：根据当前状态确定下一步要执行哪个节点的 Python 函数。它们可以是条件分支或固定转换。

![LangGraph graph](assets/2024-11-20-03-42-00.png)
{style="flex: 0.5"}

---

## 图的类型

- **StateGraph**：通用 Agent 流程编排，最常用的类型
- MessageGraph：简单的继承自 StateGraph, 聊天工作流
- prebuild react agent: 工具调用 Agent

---

## 内置 ReAct 智能体工作流 Agent create_react_agent

```python
@tool
def weather(city: str) -> str:
    """
    查询城市的天气
    :param city: 城市名字 beijing shenzhen shanghai
    :return:
    """
    if city == 'beijing':
        return 'beijing is 12 degrees'
    elif city == 'shenzhen':
        return 'shenzhen is 30 degrees'


def test_agent():
    llm = ChatOllama(model='qwen2.5', base_url='http://127.0.0.1:11435')
    tools = [weather]
    agent = create_react_agent(model=llm, tools=tools)
    r = agent.invoke({
        'messages': [
            ('user', '北京天气如何')
        ]
    })
    print(r)

状态消息

LangGraph 的底层图算法使用消息传递来定义通用程序。当一个节点完成其操作时，它会沿着一条或多条边向其他节点发送消息。然后，这些接收节点执行其功能，将生成的消息传递给下一组节点，然后继续该过程。

class MessagesState(TypedDict)
class OverallState(TypedDict)
class RequestAssistance(BaseModel)

Reducers 映射归约器

Reducers 是理解节点更新如何应用于状态的关键。状态中的每个键都有自己独立的 Reducer 函数。如果没有明确指定 Reducer 函数，则假定对该键的所有更新都应覆盖它

LangGraph graph

class MessagesState(TypedDict):
    # 定义了归约器 就执行add_message(old, new). 否则就是覆盖。
    messages: Annotated[list[AnyMessage], add_messages]

# 调用大模型
def call_model(state: MessagesState):
    messages = state['messages']
    response = model.invoke(messages)
    # We return a list, because this will get added to the existing list
    return {"messages": [response]}

workflow.add_node("agent", call_model)

图节点 Nodes

节点通常是 python 函数，其中第一个位置参数是状态，并且（可选）第二个位置参数是“config”，包含可选的可配置参数（例如 thread_id）。
START END 是特殊节点，代表开始和停止。

from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph

builder = StateGraph(dict)

def my_node(state: dict, config: RunnableConfig):
    print("In node: ", config["configurable"]["user_id"])
    return {"results": f"Hello, {state['input']}!"}

# The second argument is optional
def my_other_node(state: dict):
    return state

builder.add_node("my_node", my_node)
builder.add_node("other_node", my_other_node)

图的关系边 Edges

串行：边定义了逻辑的路由方式以及图决定停止的方式。这是代理如何工作以及不同节点如何相互通信的重要组成部分。
并行：一个节点可以有多个传出边。如果一个节点有多个传出边，则所有这些目标节点将作为下一个超级步骤的一部分并行执行。
常规边：直接从一个节点转到下一个节点。
条件边：调用函数来确定下一个要转到哪个节点。
入口点：当用户输入到达时首先调用哪个节点。
条件入口点：调用函数来确定当用户输入到达时首先调用哪个节点。

图的关系边实现

graph.add_edge("node_a", "node_b")

graph.add_conditional_edges("node_a", routing_function)

graph.add_edge(START, "node_a")

graph.add_conditional_edges(START, routing_function)

可视化

# 简单预览
app.get_graph().print_ascii()

# 不推荐 依赖远程服务
display(Image(app.get_graph().draw_mermaid_png()))

# 不推荐 环境配置复杂
display(Image(app.get_graph().draw_png()))

# 不推荐 依赖远程js
display(Image(app.get_graph().draw_mermaid_png(draw_method=MermaidDrawMethod.PYPPETEER)))

利用反向代理进行分析

# 反向代理 ollama
# mitmdump -p 11435 -m reverse:http://127.0.0.1:11434 --flow-detail 4

# 反向代理 openai
# mitmdump -p 8007 -m reverse:https://api.openai.com --flow-detail 4

llm = ChatOllama(model="qwen2.5", base_url='http://127.0.0.1:11435')

LangSmith 介绍

LangSmith 是一个用于构建生产级 LLM 应用程序的平台。它允许您密切监控和评估您的应用程序，以便您可以快速而自信地交付

LangSmith

LangSmith 配置

export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_ENDPOINT="https://api.smith.langchain.com"
export LANGCHAIN_API_KEY="xxxxxxxxx"
export LANGCHAIN_PROJECT="hogwarts-ai-testing"

LangSmith 流程分析

langsmith

Web 自动化 Agent

def test_selenium_agent():
    tools = [open, click, send_keys]
    llm = ChatOpenAI(model="gpt-4o-mini")
    agent = create_react_agent(llm, tools)
    agent.invoke(input={
        'messages': [
            ('system', '你是一名自动化测试专家，一次只返回一个动作'),
            ('user', '打开百度，搜索"ollama"，点击搜索结果中的第一条链接。'),
        ]
    })

多 Agent 示例

总结

LangGraph 架构设计与特点
LangGraph 安装
LangGraph 的核心步骤