项目详情

深入了解 langgraph 的功能与特性

🦜🕸️LangGraph

Version

⚡ 构建语言智能体的图形 ⚡

[!NOTE] 在找 JS 版本？点击这里（JS 文档）。

概述

LangGraph 是一个用于构建有状态的多角色应用程序的库，使用大语言模型（LLMs）来创建智能体和多智能体工作流程。与其他 LLM 框架相比，它提供了以下核心优势：循环、可控性和持久性。LangGraph 允许您定义包含循环的流程，这对于大多数智能体架构至关重要，从而区别于基于有向无环图（DAG）的解决方案。作为一个非常底层的框架，它提供了对应用程序流程和状态的精细控制，这对于创建可靠的智能体至关重要。此外，LangGraph 包含内置的持久性功能，支持高级的人机协作和记忆功能。

LangGraph 的灵感来源于 Pregel 和 Apache Beam。公共接口借鉴了 NetworkX 的设计。LangGraph 由 LangChain Inc 开发，虽然与 LangChain 紧密集成，但可以独立使用。

主要特性

循环和分支：在您的应用程序中实现循环和条件语句。
持久性：在图形的每一步自动保存状态。支持错误恢复、人机协作工作流、时间旅行等功能，可以在任何时候暂停和恢复图形执行。
人机协作：在图形执行过程中中断，以批准或编辑智能体计划的下一个动作。
流支持：在每个节点产生输出时进行流式传输（包括 Token 流式传输）。
与 LangChain 的集成：LangGraph 与和无缝集成（但不强制要求使用它们）。

python

from typing import Annotated, Literal, TypedDict

from langchain_core.messages import HumanMessage
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langgraph.checkpoint import MemorySaver
from langgraph.graph import END, StateGraph, MessagesState
from langgraph.prebuilt import ToolNode


# 定义智能体使用的工具
@tool
def search(query: str):
    """调用上网搜索。"""
    # 这是一个占位符，但不要告诉大语言模型……
    if "sf" in查询.lower()或"san francisco"在查询.lower()中:
        return ["60度，起雾。"]
    return ["90度，晴天。"]

tools = [search]

tool_node = ToolNode(tools)

model = ChatAnthropic(model="claude-3-5-sonnet-20240620", temperature=0).bind_tools(tools)

# 定义是否继续的函数
def should_continue(state: MessagesState) -> Literal["tools", END]:
    messages = state['messages']
    last_message = messages[-1]
    # 如果大语言模型发出工具调用，则路由到 "tools" 节点
    if last_message.tool_calls:
        return "tools"
    # 否则，停止（回复用户）
    return END

# 定义调用模型的函数
def call_model(state: MessagesState):
    messages = state['messages']
    response = model.invoke(messages)
    # 返回一个列表，因为这将被添加到现有列表中
    return {"messages": [response]}

# 定义新图形
workflow = StateGraph(MessagesState)

# 定义我们将循环的两个节点
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)

# 设置入口点为 "agent"
# 这意味着这是第一个被调用的节点
workflow.set_entry_point("agent")

# 我们现在添加一个条件边
workflow.add_conditional_edges(
    # 首先，我们定义起始节点。我们使用 "agent"。
    # 这意味着这些是 "agent" 节点被调用后采取的边。
    "agent",
    # 接下来，我们传递决定下一个被调用节点的函数。
    should_continue,
)

# 我们现在添加从 "tools" 到 "agent" 的普通边。
# 这意味着调用 "tools" 后，下一步调用 "agent" 节点。
workflow.add_edge("tools", 'agent')

# 初始化内存以在图形运行之间持久化状态
checkpointer = MemorySaver()

# 最后，我们编译它！
# 这将其编译成一个 LangChain 可运行对象，
# 这意味着你可以像使用其他可运行对象一样使用它。
# 注意，我们在编译图形时（可选地）传递了内存
app = workflow.compile(checkpointer=checkpointer)

# 使用可运行对象
final_state = app.invoke(
    {"messages": [HumanMessage(content="sf 的天气如何")]},
    config={"configurable": {"thread_id": 42}}
)
final_state["messages"][-1].content

AI 奇想空间

AI 奇想空间

langgraph

项目详情

🦜🕸️LangGraph

概述

主要特性

相关项目

supermemory

安装

示例

分步讲解

文档

langchain

Flowise