多智能体系统/ LangGraph

是的，你没听错，Smiths来了！距离我上一个帖子已经有一段时间了。如果你是新来的，嗨。我想在这篇文章中谈谈 LangGraph 以及一点关于 LangSmith 的内容。最近，我们开始实现 Agent Supervisor ；这是一种实现多代理系统的_方式_。

但在我们深入技术之前，我们能等一下吗？这是来自《黑客帝国》的 Agent Smith 。记得他们是矩阵创造的一段代码，用来保持系统中的“秩序”，这个系统让人类处于模拟中。（_即使系统崩溃了，向Neo致敬_）。Smiths处于一个等级体系中；有“Smith”命令其他Smith去做事情。（_他后来可以在模拟中复制自己的存在。实际上，这是一种处理有许多请求的系统的非常酷的方式，也许是LangChain的下一个动作；_）

经过我的玩笑时间，LangGraph是一个非常酷的库；在LangChain无法处理的情况下，它非常有用。它提供了解决复杂问题、用例或流程的方案，当需要时，一个代理系统并不够好。正如我之前所说，我将讨论 Multi-Agent Systems ，主要是 Supervisor 的实现，因为有不同的方法可以组合这些代理。由于我们正在尝试实现一个聊天机器人，我们从 Customer Support Bot 教程中获得了很多帮助。它清晰地展示了你可以在聊天中做什么，以及如何控制你的代理去做正确的事情。

LangGraph 介绍

LangGraph 建立在 LangChain 之上，完全兼容 LangChain 生态系统。它基本上是一个用于构建复杂、可扩展 AI 代理的 Python 库，采用基于图的状态机。如果您曾经尝试过 LangChain，您应该会发现当您希望代理在生产环境中运行时，它的不足之处。在生产环境中，通常需要更多的控制。您可能希望始终强制代理首先调用特定工具。您可能希望对工具的调用方式有更多控制。您可能希望根据代理所处的状态拥有不同的提示。

那么，这些“状态机”是什么呢？它们赋予您通过与 LLM 的人机交互循环工作的能力。它跟踪哪个代理在运行，使用了哪个工具，如果您希望，也可以有记忆。我现在不想深入讨论记忆，但它与我们在 LangChain 中看到的有所不同。检查点通过持久化其状态为您的代理提供“记忆”。

StateGraph

StateGraph 是一个表示图的类。通过传递 state 定义来初始化此类。该状态由图中的节点更新，这些节点以键值存储的形式返回操作。

from langgraph.graph import StateGraph
from typing import TypedDict, List, Annotated
import Operator
from langchain_core.messages import BaseMessage

class State(TypedDict):
    input: str
    messages: Annotated[Sequence[BaseMessage], operator.add]

graph = StateGraph(State)

节点

创建 StateGraph 后，您可以使用 graph.add_node(name, value) 语法添加节点。value 参数应该是一个函数或可运行的 LCEL，将被调用（即可执行工具或 LLM）

graph.add_node("model", model)
graph.add_node("tools", tool_executor)

请记住，我们将循环遍历这个图，因此在过程中退出某个地方是很重要的。END 节点用于表示图的结束。

from langgraph.graph import END

graph.add_node("end", END)

边

添加节点后，您可以添加边以创建图。现在有三种类型的边：

1 - 起始边 : 这是连接图的起点和特定节点的边。下面的代码意味着我们的图将从“model”节点开始，正如我们之前命名的那样。

graph.set_entry_point("model")

2 - 普通边 : 这些边确保一个节点总是在另一个节点之后被调用。下面的代码意味着当我们调用“tools”节点时，“model”节点将始终在其之后被调用。

graph.add_edge("tools", "model")

3 - 条件边 : 这些是 LLM 用于确定首先访问哪个节点的边。您不会严格指定去哪里；LLM 通过检查状态和用户输入来决定目的地。

条件边有三个参数。第一个是将决定下一步做什么的节点。第二个是一个函数，用于确定下一个调用哪个节点。第三个是一个映射，其中键应该是函数（2）可能返回的值，值应该是要去的节点的名称。

graph.add_conditional_edge(
    "model",
    should_continue,
    {
        "end": END,
        "continue": "tools"
    }
)

编译

在我们定义图形后，可以将其编译为可运行的程序。这个可运行的程序具有与 LangChain 可运行程序相同的方法（.invoke、.stream、.astream_log 等）

多智能体系统

当单个智能体需要按顺序执行过多工具时，它可能会失败。因此，在多智能体系统中，我们将问题进行划分，并通过不同的智能体逐步解决每个步骤，将任务路由到合适的专家。

控制流程有三种方式，协作、监督和层级团队。您可以根据项目需求查看官方文档，对于本帖，我们将遵循监督模式。

代理监督员

我们将创建一个代理组，每个代理将拥有特定的工具来完成任务。代理监督员将帮助我们分配任务。

在这个例子中，我们将有2个代理和1个监督员。第一个代理将生成随机数，另一个代理将为这些随机数绘制图表。正如我们所期望的，监督员将分配任务，当随机数生成代理完成后，它将把轮子交给另一个代理。

让我们先定义基础内容。

from langchain_openai import ChatOpenAI
from typing import Annotated, List, Tuple, Union
from langchain.tools import BaseTool, StructuredTool, Tool
from langchain_experimental.tools import PythonREPLTool
from langchain_core.tools import tool
import random

llm = ChatOpenAI(model="gpt-3.5-turbo")

python_repl_tool = PythonREPLTool()

@tool("random_number", return_direct=False)
def random_number(input:str) -> str:
    """Returns a random number between 0-100. input the word 'random'"""
    return random.randint(0, 100)

tools = [random_number,python_repl_tool]

继续编写辅助函数

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI

def create_agent(llm: ChatOpenAI, tools: list, system_prompt: str):
    prompt = ChatPromptTemplate.from_messages(
        [
            (
                "system",
                system_prompt,
            ),
            MessagesPlaceholder(variable_name="messages"),
            MessagesPlaceholder(variable_name="agent_scratchpad"),
        ]
    )
    agent = create_openai_tools_agent(llm, tools, prompt)
    executor = AgentExecutor(agent=agent, tools=tools)
    return executor

def agent_node(state, agent, name):
    result = agent.invoke(state)
    return {"messages": [HumanMessage(content=result["output"], name=name)]}

现在，让我们开始创建我们的图，并将这两个代理作为节点添加：

import operator
from typing import Annotated, Any, Dict, List, Optional, Sequence, TypedDict
import functools
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langgraph.graph import StateGraph, END

random_agent = create_agent(llm, [random_number], "You get random numbers")
random_node = functools.partial(agent_node, agent=random_agent, name="Random_Number_Generator")

code_agent = create_agent(llm, [python_repl_tool], "You generate charts using matplotlib.")
code_node = functools.partial(agent_node, agent=code_agent, name="Coder")

是时候创建我们的监督员了！

from langchain.output_parsers.openai_functions import JsonOutputFunctionsParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

members = ["Random_Number_Generator", "Coder"]
system_prompt = (
    "You are a supervisor tasked with managing a conversation between the"
    " following workers:  {members}. Given the following user request,"
    " respond with the worker to act next. Each worker will perform a"
    " task and respond with their results and status. When finished,"
    " respond with FINISH."
)

options = ["FINISH"] + members

function_def = {
    "name": "route",
    "description": "Select the next role.",
    "parameters": {
        "title": "routeSchema",
        "type": "object",
        "properties": {
            "next": {
                "title": "Next",
                "anyOf": [
                    {"enum": options},
                ],
            }
        },
        "required": ["next"],
    },
}
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system_prompt),
        MessagesPlaceholder(variable_name="messages"),
        (
            "system",
            "Given the conversation above, who should act next?"
            " Or should we FINISH? Select one of: {options}",
        ),
    ]
).partial(options=str(options), members=", ".join(members))

supervisor_chain = (
    prompt
    | llm.bind_functions(functions=[function_def], function_call="route")
    | JsonOutputFunctionsParser()
)

让我们创建我们的图！( 请阅读注释 )

首先我们定义状态并添加我们的代理节点，以及监督员节点。

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

    next: str

workflow = StateGraph(AgentState)

workflow.add_node("Random_Number_Generator", random_node)
workflow.add_node("Coder", code_node)
workflow.add_node("Supervisor", supervisor_chain)

workflow.add_edge("Random_Number_Generator", "supervisor")
workflow.add_edge("Coder", "supervisor")

workflow.add_conditional_edges(
                    "supervisor",
                    lambda x: x["next"],
                    {
                       "Random_Number_Generator": "Random_Number_Generator",
                       "Coder": "Coder",
                       "FINISH": END
                    })

workflow.set_entry_point("supervisor")

graph = workflow.compile()

让我们试试，你可以流式传输或直接调用图。

for s in graph.stream(
    {
        "messages": [
            HumanMessage(content="Get 10 random numbers and generate a histogram")
        ]
    }, config={"recursion_limit": 20}
):
    if "__end__" not in s:
        print(s)
        print("----")

输出

从输出中我们可以看到，我们从声明的监督员开始。监督员将我们引导到随机数生成器。在随机数生成器完成任务后，它返回到监督员，因为我们添加了一条边。然后监督员引导到编码器，编码器完成后返回到监督员。当任务完成时，监督员完成处理。

🥳

LangSmith

LangSmith 是一个用于 LLM 应用开发、监控和测试的平台。就我个人而言，我主要使用它进行监控，因此我只会提到这一方面。如果您启用 LangSmith 跟踪，您可以 debug 您的 LLM。

有关文档，请点击这里。

例如，我们上面进行的运行如下所示：

如果您想要更多细节：

当您有许多工具的代理时，这非常有用。在我们的案例中，他们只有一个工具，但当问题变得更加复杂并且您想理解内部发生了什么时，LangSmith 将真正帮助您跟踪步骤。您也可以使用调试控制台来做到这一点，但有了这个工具，何必麻烦呢？

您可以通过添加环境变量来启用跟踪：

os.environ["LANGCHAIN_TRACING_V2"] = "true"

进一步链接：

希望您喜欢！下次见 :)

Barry's Home