神奇cpp

这个屌丝很懒，什么也没留下！

热门标签

一個令人著迷的話題！LangGraph 有何用途？_langgraph 应用场景

作者：神奇cpp | 2024-08-11 19:40:48

踩

langgraph 应用场景

LangGraph 是一个用于使用 LLM 建立有状态、多参与者应用程式的函式库，它建构在 LangChain 之上（并旨在与 LangChain 一起使用）。它扩展了 LangChain 表达式语言，能够以循环方式跨多个计算步骤协调多个链（或参与者）。它的灵感来自 Pregel 和 Apache Beam。目前公开的介面是受 NetworkX 启发的。

主要用途是为您的法学硕士申请添加周期。至关重要的是，LangGraph 并未仅针对 DAG 工作流程进行最佳化。如果你想建立一个DAG，你应该使用LangChain表达式语言。

循环对于类似代理的行为非常重要，您可以在循环中呼叫 LLM，询问它下一步要采取什么操作。

以下是 LangGraph 的一些主要功能：

可读性：LangGraph 被设计为具有高度可读性，使用自然语言语法和结构。
效率：透过利用程式设计和自然语言的优势，LangGraph 旨在减少编码时间并提高生产力。
灵活性：LangGraph 允许开发人员编写比传统程式语言更灵活、适应性更强的程式码。

LangGraph 语法的一些范例：

x = 5（赋值）
if x > 10: print("x 大于 10") （条件语句）
for i in range(3): print(i) （循环）

LangGraph 的目标是创建一种易于学习、使用和维护的语言，使其成为想要快速且有效率地编写程式码的开发人员的有吸引力的选择。然而，与任何新的程式语言一样，它还处于早期阶段，社区、采用率和生态系统是决定其成功的关键因素。

主要特征：

自然语言语法：LangGraph使用类似自然语言的语法，使非技术人员更容易理解和使用。
可读性：该语言被设计为具有高度可读性，并减少开发人员的认知负担。
效率：LangGraph 旨在透过提供捷径和简化常见程式设计任务来减少编码时间。
灵活性：它允许开发人员编写适应不断变化的需求并可以处理复杂任务的程式码。

为什么要学 LangGraph？

简化您的开发工作流程：LangGraph 的语法和结构旨在易于理解，非常适合初学者或希望简化编码过程的人。
改善协作：借助 LangGraph 受自然语言启发的语法，开发人员可以轻松地与非技术团队成员解释和共享程式码。
保持领先地位：作为一种相对较新的语言，LangGraph 仍在不断发展，提供了为其发展做出贡献并塑造其未来的机会。

LangGraph 有哪些潜在应用？

Web 开发：LangGraph 的易用性和灵活性使其成为 Web 开发任务的有前途的候选者。
资料分析：其受自然语言启发的语法可以简化资料分析任务，从而更轻松地处理大型资料集。
人工智慧 (AI)：随着人工智慧的不断发展，LangGraph 可以用作与人工智慧系统互动或创建更类似于人类的人工智慧模型的一种方式。

LangGraph 有哪些限制？

成熟度：LangGraph仍然是一种新兴语言，其生态系统尚未完全建立。
相容性：由于它是一种相对较新的语言，因此与现有软体和框架的兼容性可能会受到限制。
效能：与任何新语言一样，效能最佳化和记忆体管理是需要解决的关键方面。

安装

pip install langgraph

快速开始

LangGraph 的核心概念之一是状态。每个图执行都会创建一个状态，该状态在执行时在图中的节点之间传递，并且每个节点在执行后用其返回值更新此内部状态。图形更新其内部状态的方式由所选图形的类型或自定义函数定义。

LangGraph 中的状态可能非常通用，但为了让事情更容易开始，我们将展示一个示例，其中使用内置 MessageGraph 类将图的状态限制为聊天消息列表。当将 LangGraph 与 LangChain 聊天模型一起使用时，这很方便，因为我们可以直接返回聊天模型输出。

首先，安装LangChain OpenAI集成包：

pip install langchain_openai

我们还需要导出一些环境变量：

export OPENAI_API_KEY=sk-...

现在我们准备好了！下图包含一个名为“oracle”的节点，它执行聊天模型，然后返回结果：


from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langgraph.graph import END, MessageGraph
 
model = ChatOpenAI(temperature=0)
 
graph = MessageGraph()
 
graph.add_node("oracle", model)
graph.add_edge("oracle", END)
 
graph.set_entry_point("oracle")
 
runnable = graph.compile()

Let's run it!

runnable.invoke(HumanMessage("What is 1 + 1?"))

[HumanMessage(content='What is 1 + 1?'), AIMessage(content='1 + 1 equals 2.')]

那么我们在这里做了什么？让我们一步步分解：

首先，我们初始化模型和 MessageGraph。
接下来，我们向图中添加一个名为“oracle”的节点，它只是使用给定的输入调用模型。
我们从这个“oracle”节点添加一条边到特殊字符串 END。这意味着执行将在当前节点之后结束。
我们将“oracle”设置为该图的入口点。
我们编译该图，确保不能对其进行更多修改。

然后，当我们执行该图时：

LangGraph 将输入消息添加到内部状态，然后将状态传递到入口点节点“oracle”。
“oracle”节点执行，调用聊天模型。
聊天模型返回一条 AIMessage。 LangGraph 将其添加到状态中。
执行进行到特殊的 END 值并输出最终状态。

结果，我们得到了两个聊天消息的列表作为输出。

与 LCEL 互动

对于那些已经熟悉 LangChain 的人来说，add_node 实际上接受任何函数或可运行程序作为输入。在上面的示例中，模型“按原样”使用，但我们也可以传入一个函数：


def call_oracle(messages: list):
    return model.invoke(messages)
 
graph.add_node("oracle", call_oracle)

只需确保您注意到可运行的输入是整个当前状态这一事实即可。所以这会失败：


# 这不适用于 MessageGraph！
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant named {name} who always speaks in pirate dialect"),
    MessagesPlaceholder(variable_name="messages"),
])
 
chain = prompt | model
 
# 状态是消息列表，但我们的链需要一个字典输入：
#
# { "name": some_string, "messages": [] }
#
# 因此，图在这里执行时会抛出异常。
graph.add_node("oracle", chain)

条件边

现在，让我们转向一些不那么琐碎的事情。因为数学对于法学硕士来说可能很困难，所以让我们允许法学硕士使用工具调用有条件地调用“乘法”节点。

我们将使用附加的“乘法”重新创建图表，如果它是工具调用，则该“乘法”将获取最新消息的结果，并计算结果。我们还将计算器作为工具绑定到 OpenAI 模型，以允许模型选择性地使用响应当前状态所需的工具：


import json
from langchain_core.messages import ToolMessage
from langchain_core.tools import tool
from langchain_core.utils.function_calling import convert_to_openai_tool
 
@tool
def multiply(first_number: int, second_number: int):
    """Multiplies two numbers together."""
    return first_number * second_number
 
model = ChatOpenAI(temperature=0)
model_with_tools = model.bind(tools=[convert_to_openai_tool(multiply)])
 
graph = MessageGraph()
 
def invoke_model(state: List[BaseMessage]):
    return model_with_tools.invoke(state)
 
graph.add_node("oracle", invoke_model)
 
def invoke_tool(state: List[BaseMessage]):
    tool_calls = state[-1].additional_kwargs.get("tool_calls", [])
    multiply_call = None
 
    for tool_call in tool_calls:
        if tool_call.get("function").get("name") == "multiply":
            multiply_call = tool_call
 
    if multiply_call is None:
        raise Exception("No adder input found.")
 
    res = multiply.invoke(
        json.loads(multiply_call.get("function").get("arguments"))
    )
 
    return ToolMessage(
        tool_call_id=multiply_call.get("id"),
        content=res
    )
 
graph.add_node("multiply", invoke_tool)
 
graph.add_edge("multiply", END)
 
graph.set_entry_point("oracle")

现在让我们想一想——我们希望发生什么？

如果“oracle”节点返回一条需要工具调用的消息，我们要执行“multiply”节点
如果没有，我们可以结束执行

我们可以使用条件边来实现这一点，条件边使用函数根据当前状态将执行路由到节点。

看起来是这样的：


def router(state: List[BaseMessage]):
    tool_calls = state[-1].additional_kwargs.get("tool_calls", [])
    if len(tool_calls):
        return "multiply"
    else:
        return "end"
 
graph.add_conditional_edges("oracle", router, {
    "multiply": "multiply",
    "end": END,
})

如果模型输出包含工具调用，我们将移至“乘法”节点。否则，我们就结束了。

伟大的！现在剩下的就是编译图表并尝试一下。与数学相关的问题将转至计算器工具：


runnable = graph.compile()
 
runnable.invoke(HumanMessage("What is 123 * 456?"))


 
[HumanMessage(content='What is 123 * 456?'),
 AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_OPbdlm8Ih1mNOObGf3tMcNgb', 'function': {'arguments': '{"first_number":123,"second_number":456}', 'name': 'multiply'}, 'type': 'function'}]}),
 ToolMessage(content='56088', tool_call_id='call_OPbdlm8Ih1mNOObGf3tMcNgb')]

而对话响应是直接输出的：

runnable.invoke(HumanMessage("What is your name?"))


[HumanMessage(content='What is your name?'),
 AIMessage(content='My name is Assistant. How can I assist you today?')]

循环次数

现在，让我们看一个更一般的循环示例。我们将从 LangChain 重新创建 AgentExecutor 类。代理本身将使用聊天模型和函数调用。该代理将其所有状态表示为消息列表。

我们需要安装一些 LangChain 软件包以及 Tavily 来用作示例工具。

pip install -U langchain langchain_openai tavily-python

我们还需要导出一些额外的环境变量以供 OpenAI 和 Tavily API 访问。


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

或者，我们可以设置 LangSmith 以获得一流的可观察性。


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

设置工具

如上所述，我们首先定义要使用的工具。对于这个简单的示例，我们将通过 Tavily 使用内置搜索工具。然而，创建自己的工具确实很容易 - 请参阅此处的文档了解如何做到这一点。


from langchain_community.tools.tavily_search import TavilySearchResults
 
tools = [TavilySearchResults(max_results=1)]

我们现在可以将这些工具包装在一个简单的 LangGraph ToolExecutor 中。此类接收 ToolIncation 对象、调用该工具并返回输出。 ToolIncation 是具有 tool 和 tool_input 属性的任何类。


from langgraph.prebuilt import ToolExecutor
 
tool_executor = ToolExecutor(tools)

设置模型

现在我们需要加载我们想要使用的聊天模型。这次，我们将使用旧的函数调用接口。本演练将使用 OpenAI，但我们可以选择任何支持 OpenAI 函数调用的模型。


from langchain_openai import ChatOpenAI
 
# We will set streaming=True so that we can stream tokens
# See the streaming section for more information on this.
model = ChatOpenAI(temperature=0, streaming=True)

完成此操作后，我们应该确保模型知道它可以调用这些工具。我们可以通过将LangChain工具转换为OpenAI函数调用的格式，然后将它们绑定到模型类来实现。


from langchain.tools.render import format_tool_to_openai_function
 
functions = [format_tool_to_openai_function(t) for t in tools]
model = model.bind_functions(functions)

定义代理状态

这次，我们将使用更通用的 StateGraph。该图由传递到每个节点的状态对象参数化。请记住，每个节点都会返回更新该状态的操作。这些操作可以设置状态的特定属性（例如覆盖现有值）或添加到现有属性。是设置还是添加是通过注释用于构造图形的状态对象来指示的。

对于这个例子，我们将跟踪的状态只是一个消息列表。我们希望每个节点只将消息添加到该列表中。因此，我们将使用带有一个键（消息）的 TypedDict 并对其进行注释，以便消息属性始终添加到第二个参数（operator.add）。


from typing import TypedDict, Annotated, Sequence
import operator
from langchain_core.messages import BaseMessage
 
 
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]

您可以将初始示例中使用的 MessageGraph 视为该图的预配置版本，其中状态直接是消息数组，并且更新步骤始终是将节点的返回值附加到内部状态。

定义节点

我们现在需要在图中定义一些不同的节点。在 langgraph 中，节点可以是函数，也可以是可运行的。为此，我们需要两个主要节点：

代理：负责决定采取什么（如果有）行动。
调用工具的函数：如果代理决定采取操作，则该节点将执行该操作。

我们还需要定义一些边。其中一些边缘可能是有条件的。它们是有条件的原因是，根据节点的输出，可以采用多个路径之一。在该节点运行之前，所采用的路径是未知的（LLM 决定）。

条件边缘：调用代理后，我们应该：

a. 如果代理说要采取行动，那么应该调用调用工具的函数
b. 如果代理说已经完成，那么就应该完成
Normal Edge：调用工具后，它应该始终返回给代理来决定下一步做什么

让我们定义节点以及决定如何采用条件边的函数。


from langgraph.prebuilt import ToolInvocation
import json
from langchain_core.messages import FunctionMessage
 
# 定义判断是否继续的函数
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"
 
# 定义调用模型的函数
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]
    # 我们从 function_call 构造一个 ToolInspiration
    action = ToolInvocation(
        tool=last_message.additional_kwargs["function_call"]["name"],
        tool_input=json.loads(last_message.additional_kwargs["function_call"]["arguments"]),
    )
    # 我们调用 tool_executor 并得到响应
    response = tool_executor.invoke(action)
    #我们使用响应来创建 FunctionMessage
    function_message = FunctionMessage(content=str(response), name=action.tool)
    # 我们返回一个列表，因为这将被添加到现有列表中
    return {"messages": [function_message]}

定义图表

我们现在可以将它们放在一起并定义图表！


from langgraph.graph import StateGraph, END
# 定义一个新图
workflow = StateGraph(AgentState)
 
# 定义我们将在其之间循环的两个节点
workflow.add_node("agent", call_model)
workflow.add_node("action", call_tool)
 
# 将入口点设置为`agent`
# 这意味着该节点是第一个被调用的节点
workflow.set_entry_point("agent")
 
# 我们现在添加一个条件边
workflow.add_conditional_edges(
# 首先，我们定义起始节点。 我们使用“代理”。
     # 这意味着这些是调用 `agent` 节点后获取的边。
     "agent",
     # 接下来，我们传入函数来确定接下来调用哪个节点。
     should_continue,
     # 最后我们传入一个映射。
     # 键是字符串，值是其他节点。
     # END 是一个特殊的节点，标记图应该结束。
     # 将会发生的事情是我们将调用 `should_continue`，然后输出
     # 将与此映射中的键进行匹配。
     # 根据匹配的节点，将调用该节点。
     {
         # 如果是`tools`，那么我们称之为工具节点。
          "continue": "action",
         # 否则我们就结束了。
         "end": END
    }
)
 
# 我们现在添加从 `tools` 到 `agent` 的正常边缘。
# 这意味着在调用`tools`之后，接下来调用`agent`节点。
workflow.add_edge('action', 'agent')
 
# 最后，我们编译它！
# 这会将其编译成 LangChain Runnable，
# 意味着你可以像使用任何其他可运行程序一样使用它
app = workflow.compile()

用它！

我们现在可以使用它了！现在，它公开了与所有其他 LangChain 可运行程序相同的接口。该可运行程序接受消息列表。


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

这可能需要一点时间——它在幕后进行了一些调用。为了开始看到一些中间结果，我们可以使用流式传输 - 有关更多信息，请参阅下文。

流媒体

LangGraph 支持多种不同类型的流式传输。

流节点输出

使用 LangGraph 的好处之一是可以轻松地传输由每个节点生成的输出。


inputs = {"messages": [HumanMessage(content="what is the weather in sf")]}
for output in app.stream(inputs):
    # Stream() 产生字典，其输出以节点名称为键
    for key, value in output.items():
        print(f"Output from node '{key}':")
        print("---")
        print(value)
    print("\n---\n")


Output from node 'agent':
---
{'messages': [AIMessage(content='', additional_kwargs={'function_call': {'arguments': '{\n  "query": "weather in San Francisco"\n}', 'name': 'tavily_search_results_json'}})]}
 
---
 
Output from node 'action':
---
{'messages': [FunctionMessage(content="[{'url': 'https://weatherspark.com/h/m/557/2024/1/Historical-Weather-in-January-2024-in-San-Francisco-California-United-States', 'content': 'January 2024 Weather History in San Francisco California, United States  Daily Precipitation in January 2024 in San Francisco Observed Weather in January 2024 in San Francisco  San Francisco Temperature History January 2024 Hourly Temperature in January 2024 in San Francisco  Hours of Daylight and Twilight in January 2024 in San FranciscoThis report shows the past weather for San Francisco, providing a weather history for January 2024. It features all historical weather data series we have available, including the San Francisco temperature history for January 2024. You can drill down from year to month and even day level reports by clicking on the graphs.'}]", name='tavily_search_results_json')]}
 
---
 
Output from node 'agent':
---
{'messages': [AIMessage(content="I couldn't find the current weather in San Francisco. However, you can visit [WeatherSpark](https://weatherspark.com/h/m/557/2024/1/Historical-Weather-in-January-2024-in-San-Francisco-California-United-States) to check the historical weather data for January 2024 in San Francisco.")]}
 
---
 
Output from node '__end__':
---
{'messages': [HumanMessage(content='what is the weather in sf'), AIMessage(content='', additional_kwargs={'function_call': {'arguments': '{\n  "query": "weather in San Francisco"\n}', 'name': 'tavily_search_results_json'}}), FunctionMessage(content="[{'url': 'https://weatherspark.com/h/m/557/2024/1/Historical-Weather-in-January-2024-in-San-Francisco-California-United-States', 'content': 'January 2024 Weather History in San Francisco California, United States  Daily Precipitation in January 2024 in San Francisco Observed Weather in January 2024 in San Francisco  San Francisco Temperature History January 2024 Hourly Temperature in January 2024 in San Francisco  Hours of Daylight and Twilight in January 2024 in San FranciscoThis report shows the past weather for San Francisco, providing a weather history for January 2024. It features all historical weather data series we have available, including the San Francisco temperature history for January 2024. You can drill down from year to month and even day level reports by clicking on the graphs.'}]", name='tavily_search_results_json'), AIMessage(content="I couldn't find the current weather in San Francisco. However, you can visit [WeatherSpark](https://weatherspark.com/h/m/557/2024/1/Historical-Weather-in-January-2024-in-San-Francisco-California-United-States) to check the historical weather data for January 2024 in San Francisco.")]}
 
---

流媒体 LLM 代币

您还可以访问每个节点生成的 LLM 令牌。在这种情况下，只有“代理”节点生成 LLM 令牌。为了使其正常工作，您必须使用支持流式传输的LLM，并在构建LLM时对其进行设置（例如ChatOpenAI（model =“gpt-3.5-turbo-1106”，streaming = True））


inputs = {"messages": [HumanMessage(content="what is the weather in sf")]}
async for output in app.astream_log(inputs, include_types=["llm"]):
    # astream_log() yields the requested logs (here LLMs) in JSONPatch format
    for op in output.ops:
        if op["path"] == "/streamed_output/-":
            # this is the output from .stream()
            ...
        elif op["path"].startswith("/logs/") and op["path"].endswith(
            "/streamed_output/-"
        ):
            # because we chose to only include LLMs, these are LLM tokens
            print(op["value"])


content='' additional_kwargs={'function_call': {'arguments': '', 'name': 'tavily_search_results_json'}}
content='' additional_kwargs={'function_call': {'arguments': '{\n', 'name': ''}}
content='' additional_kwargs={'function_call': {'arguments': ' ', 'name': ''}}
content='' additional_kwargs={'function_call': {'arguments': ' "', 'name': ''}}
content='' additional_kwargs={'function_call': {'arguments': 'query', 'name': ''}}
...

何时使用

什么时候应该使用它而不是 LangChain 表达式语言？

如果你需要循环。

Langchain 表达式语言允许您轻松定义链 (DAG)，但没有良好的添加循环的机制。 langgraph 添加了该语法。

操作指南

这些指南展示了如何以特定方式使用 LangGraph。

异步

如果您在异步工作流程中运行 LangGraph，您可能希望默认创建异步节点。有关如何执行此操作的演练，请参阅此文档

流媒体代币

有时，语言模型需要一段时间才能响应，您可能希望将令牌流式传输给最终用户。有关如何执行此操作的指南，请参阅此文档

坚持

LangGraph 具有内置的持久性，允许您保存图形在该点的状态并从那里恢复。有关如何执行此操作的演练，请参阅此文档

人在回路

LangGraph 内置了对人机交互工作流程的支持。当您希望在进入特定节点之前让人工检查当前状态时，这非常有用。有关如何执行此操作的演练，请参阅此文档

可视化图表

使用 LangGraph 创建的代理可能很复杂。为了更容易理解幕后发生的事情，我们添加了打印和可视化图表的方法。这可以创建 ascii 艺术和 png。有关如何执行此操作的演练，请参阅此文档

“时间旅行”

通过“时间旅行”功能，您可以跳转到图形执行中的任何点，修改状态，然后从那里重新运行。这对于调试工作流程以及面向最终用户的工作流程非常有用，以允许他们更正状态。有关如何执行此操作的演练，请参阅此文档

例子

ChatAgentExecutor：带函数调用

该代理执行器将消息列表作为输入并输出消息列表。所有代理状态都表示为消息列表。这里具体使用了OpenAI函数调用。对于支持函数调用的基于聊天的较新模型，推荐使用代理执行器。

入门笔记本：逐步从头开始创建此类执行器
高级入口点：演练如何使用聊天代理执行器的高级入口点。

修改

我们还有很多示例强调如何稍微修改基本聊天代理执行器。这些都是基于入门笔记本构建的，因此建议您首先从入门笔记本开始。

人机交互：如何添加人机交互组件
强制先调用工具：如何始终先调用特定工具
以特定格式响应：如何强制代理以特定格式响应
动态直接返回工具输出：如何动态地让代理选择是否将工具的结果直接返回给用户
管理代理步骤：如何更明确地管理代理采取的中间步骤

代理执行者

该代理执行器使用现有的LangChain代理。

入门笔记本：逐步从头开始创建此类执行器
高级入口点：演练如何使用聊天代理执行器的高级入口点。

修改

我们还有很多示例强调如何稍微修改基本聊天代理执行器。这些都是基于入门笔记本构建的，因此建议您首先从入门笔记本开始。

人机交互：如何添加人机交互组件
强制先调用工具：如何始终先调用特定工具
管理代理步骤：如何更明确地管理代理采取的中间步骤

规划代理示例

以下笔记本实现了“计划和执行”风格的代理架构原型，其中 LLM 计划器将用户请求分解为程序，执行器执行程序，LLM 综合基于响应（和/或动态重新计划）关于程序输出。

计划和执行：一个简单的代理，其中包含生成多步骤任务列表的计划程序、调用计划中的工具的执行程序以及响应或生成更新计划的重新计划程序。基于 Wang 等人的 Plan-and-solve 论文。等人。
无观察推理：规划器生成一个任务列表，其观察结果保存为变量。变量可以在后续任务中使用，以减少进一步重新规划的需要。基于 Xu 等人的 ReWOO 论文。等人。
LLMCompiler：规划器生成具有可变响应的任务 DAG。任务被急切地流式传输和执行，以最大限度地减少工具执行时间。基于 Kim 等人的论文。等人。

反思/自我批评

当输出质量成为主要问题时，通常会结合自我批评或反思和外部验证来完善系统的输出。以下示例展示了实现此类设计的研究。

基本反射：在图表中添加一个简单的“反射”步骤，以提示您的系统修改其输出。
反思：批评代理响应中缺失和多余的方面，以指导后续步骤。基于 Shinn 等人的《Reflexion》。等人。
语言代理树搜索：并行执行多个代理，使用反射和环境奖励来驱动蒙特卡罗树搜索。基于 Zhou 等人的 LATS。等人。

多代理示例

多代理协作：如何创建两个共同完成任务的代理
带主管的多代理：如何通过使用法学硕士作为“主管”来协调各个代理来分配工作
分层代理团队：如何将代理“团队”编排为可以协作解决问题的嵌套图

网络研究

STORM：写作系统，可生成关于任何主题的维基百科风格的文章，应用大纲生成（规划）+多视角问答以增加广度和可靠性。基于 Shao 等人的 STORM。等人。

通过模拟评估聊天机器人

在多轮情况下评估聊天机器人通常很困难。做到这一点的一种方法是模拟。

聊天机器人评估作为多代理模拟：如何模拟“虚拟用户”和聊天机器人之间的对话
对数据集进行评估：在 LangSmith 数据集上对您的助手进行基准测试，该数据集要求模拟客户对您的聊天机器人进行红组。

多模式示例

WebVoyager：支持视觉的 Web 浏览代理，使用标记集提示来导航 Web 浏览器并执行任务

表链

Chain of Table 是一个框架，在回答有关表格数据的问题时可实现 SOTA 性能。 Github 用户 CYQIQ 的这个实现使用 LangGraph 来控制流程。

文档

只有几个新的 API 可供使用。

状态图

主要入口点是 StateGraph。

from langgraph.graph import StateGraph

该类负责构建图。它公开了一个受 NetworkX 启发的界面。该图由传递到每个节点的状态对象参数化。

__init__

    def __init__(self, schema: Type[Any]) -> None:

构建图时，您需要传入状态的模式。然后每个节点返回操作来更新该状态。这些操作可以设置状态的特定属性（例如覆盖现有值）或添加到现有属性。是设置还是添加是通过注释用于构造图形的状态对象来指示的。

指定模式的推荐方法是使用类型化字典：fromtyping import TypedDict

然后，您可以使用 from 键入 import Annotated 来注释不同的属性。目前，唯一支持的注解是 import 操作符；运算符.add. 此注释将使任何返回此属性的节点将新结果添加到现有值中。

让我们看一个例子：


from typing import TypedDict, Annotated, Union
from langchain_core.agents import AgentAction, AgentFinish
import operator
 
 
class AgentState(TypedDict):
   # 输入字符串
   input: str
   # 对代理的给定调用的结果
   # 需要 `None` 作为有效类型，因为这就是它的开头
   agent_outcome: Union[AgentAction, AgentFinish, None]
   # 行动清单和相应的观察结果
   # 这里我们用 `operator.add` 来注释它来指示操作
   # 该状态应该添加到现有值中（而不是覆盖它）
   intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]

然后我们可以这样使用它：


# 使用该状态初始化 StateGraph
graph = StateGraph(AgentState)
# 创建节点和边
...
# 编译图表
app = graph.compile()
 
# 输入应该是字典，因为状态是 TypedDict
inputs = {
   # 我们假设这是输入
   "input": "hi"
   # 假设 agent_outcome 由图表设置为某个点
   # 不需要提供，默认为None
   # 假设“intermediate_steps”是由图表随着时间的推移而构建的
   # 不需要提供，默认为空列表
   # `intermediate_steps` 是一个空列表而不是 `None` 的原因是
   # 它用 `operator.add` 注释
}

`.add_node`

    def add_node(self, key: str, action: RunnableLike) -> None:

此方法向图中添加一个节点。它需要两个参数：

key：代表节点名称的字符串。这一定是独一无二的。
action：调用该节点时要执行的操作。这应该是一个函数或一个可运行的。

`.add_edge`

    def add_edge(self, start_key: str, end_key: str) -> None:

创建从一个节点到下一个节点的边。这意味着第一个节点的输出将传递到下一个节点。它需要两个参数。

start_key：表示起始节点名称的字符串。该密钥必须已在图中注册。
end_key：表示结束节点名称的字符串。该密钥必须已在图中注册。

`.add_conditional_edges`


    def add_conditional_edges(
        self,
        start_key: str,
        condition: Callable[..., str],
        conditional_edge_mapping: Dict[str, str],
    ) -> None:

此方法添加条件边。这意味着只会采用下游边之一，具体选择哪一条取决于起始节点的结果。这需要三个参数：

start_key：表示起始节点名称的字符串。该密钥必须已在图中注册。
条件：调用以确定下一步做什么的函数。输入将是起始节点的输出。它应该返回一个存在于conditional_edge_mapping中并表示要采用的边缘的字符串。
conditional_edge_mapping：字符串到字符串的映射。键应该是可以按条件返回的字符串。如果返回该条件，这些值应该是要调用的下游节点。

`.set_entry_point`

    def set_entry_point(self, key: str) -> None:

图表的入口点。这是首先被调用的节点。它只需要一个参数：

key：应该首先调用的节点的名称。

`.set_conditional_entry_point`


    def set_conditional_entry_point(
        self,
        condition: Callable[..., str],
        conditional_edge_mapping: Optional[Dict[str, str]] = None,
    ) -> None:

此方法添加一个条件入口点。这意味着当调用图时，它会调用条件Callable来决定首先进入哪个节点。

条件：调用以确定下一步做什么的函数。输入将是图形的输入。它应该返回一个存在于conditional_edge_mapping中并表示要采用的边缘的字符串。
conditional_edge_mapping：字符串到字符串的映射。键应该是可以按条件返回的字符串。如果返回该条件，这些值应该是要调用的下游节点。

`.set_finish_point`

    def set_finish_point(self, key: str) -> None:

这是图表的出口点。当调用该节点时，结果将是图表的最终结果。它只有一个参数：

key：节点名称，调用时会返回调用结果作为最终输出

注意：如果您之前在任何时候创建了一条到 END 的边（条件或正常），则不需要调用此函数

图形


from langgraph.graph import Graph
 
graph = Graph()

它与 StateGraph 具有相同的接口，但它不会随着时间的推移更新状态对象，而是依赖于传递每个步骤的完整状态。这意味着从一个节点返回的任何内容都是下一个节点的输入。

结尾

from langgraph.graph import END

这是一个特殊的节点，代表图的末端。这意味着传递到该节点的任何内容都将是该图的最终输出。它可以用在两个地方：

作为add_edge中的end_key
作为 conditional_edge_mapping 中的值传递给 add_conditional_edges

预建示例

我们还添加了一些方法，以便轻松使用常见的预构建图表和组件。

工具执行器

from langgraph.prebuilt import ToolExecutor

这是一个简单的帮助器类，用于帮助调用工具。它由工具列表参数化：


tools = [...]
tool_executor = ToolExecutor(tools)

然后它公开一个可运行的接口。它可用于调用工具：您可以传入 AgentAction，它将查找相关工具并使用适当的输入调用它。

chat_agent_executor.create_function_calling_executor

from langgraph.prebuilt import chat_agent_executor

这是一个辅助函数，用于创建与利用函数调用的聊天模型配合使用的图表。可以通过传入模型和工具列表来创建。该模型必须是支持OpenAI函数调用的模型。


from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.prebuilt import chat_agent_executor
from langchain_core.messages import HumanMessage
 
tools = [TavilySearchResults(max_results=1)]
model = ChatOpenAI()
 
app = chat_agent_executor.create_function_calling_executor(model, tools)
 
inputs = {"messages": [HumanMessage(content="what is the weather in sf")]}
for s in app.stream(inputs):
    print(list(s.values())[0])
    print("----")

chat_agent_executor.create_tool_calling_executor

from langgraph.prebuilt import chat_agent_executor

这是一个辅助函数，用于创建与利用工具调用的聊天模型一起使用的图表。可以通过传入模型和工具列表来创建。该模型必须是支持OpenAI工具调用的模型。


from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.prebuilt import chat_agent_executor
from langchain_core.messages import HumanMessage
 
tools = [TavilySearchResults(max_results=1)]
model = ChatOpenAI()
 
app = chat_agent_executor.create_tool_calling_executor(model, tools)
 
inputs = {"messages": [HumanMessage(content="what is the weather in sf")]}
for s in app.stream(inputs):
    print(list(s.values())[0])
    print("----")

create_agent_executor

from langgraph.prebuilt import create_agent_executor

这是一个辅助函数，用于创建与 LangChain Agent 一起使用的图表。可以通过传入代理和工具列表来创建。


from langgraph.prebuilt import create_agent_executor
from langchain_openai import ChatOpenAI
from langchain import hub
from langchain.agents import create_openai_functions_agent
from langchain_community.tools.tavily_search import TavilySearchResults
 
tools = [TavilySearchResults(max_results=1)]
 
# Get the prompt to use - you can modify this!
prompt = hub.pull("hwchase17/openai-functions-agent")
 
# Choose the LLM that will drive the agent
llm = ChatOpenAI(model="gpt-3.5-turbo-1106")
 
# Construct the OpenAI Functions agent
agent_runnable = create_openai_functions_agent(llm, tools, prompt)
 
app = create_agent_executor(agent_runnable, tools)
 
inputs = {"input": "what is the weather in sf", "chat_history": []}
for s in app.stream(inputs):
    print(list(s.values())[0])
    print("----")

声明：本文内容由网友自发贡献，转载请注明出处：【wpsshop博客】