图片来自 DALL-E 3

LangChain是构建由 Lardge 语言模型提供支持的应用程序的领先框架之一。借助LangChain 表达语言(LCEL)，定义和执行分步操作序列（也称为链）变得更加简单。用更专业的术语来说，LangChain 允许我们创建 DAG（有向无环图）。

随着 LLM 应用程序（尤其是 LLM 代理）的发展，我们开始将 LLM 不仅用于执行，还用作推理引擎。这种转变引入了经常涉及重复（循环）和复杂条件的交互。在这种情况下，LCEL 不够用，因此 LangChain 实现了一个新模块 — LangGraph。

LangGraph（正如你从名字中猜到的那样）将所有交互建模为循环图。这些图表支持开发高级工作流程以及与多个循环和 if 语句的交互，使其成为创建代理和多代理工作流程的便捷工具。

在本文中，我将探讨 LangGraph 的主要特性和功能，包括多代理应用程序。我们将构建一个可以回答不同类型问题的系统，并深入研究如何实现人机交互设置。

LangGraph 基础知识

LangGraph 是 LangChain 生态系统的一部分，因此我们将继续使用众所周知的概念，例如提示模板、工具等。但是，LangGraph 带来了许多额外的概念。让我们来讨论一下。

LangGraph 是为了定义循环图而创建的。图由以下元素组成：

• 节点代表实际操作，可以是 LLM、代理或函数。此外，特殊的 END 节点标记执行结束。
• 边连接节点并确定图的执行流程。基本边只是将一个节点链接到另一个节点，条件边则包含 if 语句和其他逻辑。

另一个重要概念是图的状态。状态是图组件之间协作的基础元素。它代表图的快照，任何部分（无论是节点还是边）都可以在执行期间访问和修改该快照以检索或更新信息。

此外，状态在持久性中起着至关重要的作用。它会在每个步骤后自动保存，让您可以随时暂停和恢复执行。此功能支持开发更复杂的应用程序，例如需要纠错或包含人机交互的应用程序。

单代理工作流程

从头开始构建代理

让我们从简单的开始，尝试使用 LangGraph 作为基本用例——带有工具的代理。

与往常一样，我们将首先定义代理的工具。由于我将在此示例中使用 ClickHouse 数据库，因此我定义了一个函数来执行任何查询。如果您愿意，可以使用其他数据库，因为我们不会依赖任何特定于数据库的功能。

CH_HOST = 'http://localhost:8123' # default address 
import requests

def get_clickhouse_data(query, host = CH_HOST, connection_timeout = 1500):
  r = requests.post(host, params = {'query': query}, 
    timeout = connection_timeout)
  if r.status_code == 200:
      return r.text
  else: 
      return 'Database returned the following error:\n' + r.text

让 LLM 工具可靠且不易出错至关重要。如果数据库返回错误，我会将此反馈提供给 LLM，而不是抛出异常并停止执行。然后，LLM 代理将有机会修复错误并再次调用该函数。

让我们定义一个名为 的工具execute_sql，它可以执行任何 SQL 查询。我们使用pydantic来指定工具的结构，确保 LLM 代理拥有有效使用该工具所需的所有信息。

from langchain_core.tools import tool
from pydantic.v1 import BaseModel, Field
from typing import Optional

class SQLQuery(BaseModel):
  query: str = Field(description="SQL query to execute")

@tool(args_schema = SQLQuery)
def execute_sql(query: str) -> str:
  """Returns the result of SQL query execution"""
  return get_clickhouse_data(query)  

我们可以打印创建的工具的参数来查看传递给LLM的信息。

print(f'''
name: {execute_sql.name}
description: {execute_sql.description}
arguments: {execute_sql.args}
''')

# name: execute_sql
# description: Returns the result of SQL query execution
# arguments: {'query': {'title': 'Query', 'description': 
#   'SQL query to execute', 'type': 'string'}}

一切看起来都很好。我们已经设置了必要的工具，现在可以继续定义 LLM 代理了。正如我们上面所讨论的，LangGraph 中代理的基石是其状态，这使得我们图的不同部分之间能够共享信息。

我们当前的示例相对简单。因此，我们只需要存储消息的历史记录。让我们定义代理状态。

# useful imports
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage

# defining agent state
class AgentState(TypedDict):
   messages: Annotated[list[AnyMessage], operator.add]

AgentState我们在— —中定义了一个参数，messages它是 类的对象列表AnyMessage。此外，我们用operator.add(reducer) 对其进行了注释。此注释可确保每次节点返回消息时，都会将其附加到状态中的现有列表中。如果没有此运算符，则每条新消息都会替换先前的值，而不是添加到列表中。

下一步是定义代理本身。让我们从__init__功能开始。我们将为代理指定三个参数：模型、工具列表和系统提示。

class SQLAgent:
  # initialising the object
  def __init__(self, model, tools, system_prompt = ""):
    self.system_prompt = system_prompt
    
    # initialising graph with a state 
    graph = StateGraph(AgentState)

    # adding nodes 
    graph.add_node("llm", self.call_llm)
    graph.add_node("function", self.execute_function)
    graph.add_conditional_edges(
        "llm",
        self.exists_function_calling,
        {True: "function", False: END}
    )
    graph.add_edge("function", "llm")

    # setting starting point
    graph.set_entry_point("llm")

    self.graph = graph.compile()
    self.tools = {t.name: t for t in tools}
    self.model = model.bind_tools(tools)

在初始化函数中，我们概述了图的结构，其中包括两个节点：llm和action。节点是实际的动作，因此我们有与之关联的函数。稍后我们将定义函数。

此外，我们有一个条件边，它决定我们是否需要执行函数或生成最终答案。对于此边，我们需要指定前一个节点（在我们的例子中是llm），决定下一步的函数，以及基于函数输出（格式化为字典）的后续步骤的映射。如果exists_function_calling返回 True，我们将跟随函数节点。否则，执行将在特殊END节点处结束，这标志着流程的结束。

function我们在和之间添加了一条边llm。它只是连接这两个步骤，并且会在没有任何条件的情况下执行。

定义好主要结构后，就该创建上面列出的所有函数了。第一个是call_llm。此函数将执行 LLM 并返回结果。

代理状态将自动传递给函数，以便我们可以从中使用保存的系统提示和模型。

class SQLAgent:
  <...>

  def call_llm(self, state: AgentState):
    messages = state['messages']
    # adding system prompt if it's defined
    if self.system_prompt:
        messages = [SystemMessage(content=self.system_prompt)] + messages
    
    # calling LLM
    message = self.model.invoke(messages)

    return {'messages': [message]}

因此，我们的函数返回一个字典，用于更新代理状态。由于我们将operator.add其用作状态的 Reducer，因此返回的消息将附加到存储在状态中的消息列表中。

我们需要的下一个函数是execute_function运行工具的函数。如果 LLM 代理决定调用某个工具，我们将在参数中看到它message.tool_calls。

class SQLAgent:
  <...>  

  def execute_function(self, state: AgentState):
    tool_calls = state['messages'][-1].tool_calls

    results = []
    for tool in tool_calls:
      # checking whether tool name is correct
      if not t['name'] in self.tools:
      # returning error to the agent 
      result = "Error: There's no such tool, please, try again" 
      else:
      # getting result from the tool
      result = self.tools[t['name']].invoke(t['args'])
      
      results.append(
        ToolMessage(
          tool_call_id=t['id'], 
          name=t['name'], 
          content=str(result)
        )
    )
    return {'messages': results}

在此函数中，我们迭代 LLM 返回的工具调用，并调用这些工具或返回错误消息。最后，我们的函数返回一个包含单个键的字典，messages该字典将用于更新图形状态。

只剩下一个函数——条件边的函数，它定义我们是否需要执行工具或提供最终结果。这很简单。我们只需要检查最后一条消息是否包含任何工具调用。

class SQLAgent:
  <...>  

  def exists_function_calling(self, state: AgentState):
    result = state['messages'][-1]
    return len(result.tool_calls) > 0

现在是时候为其创建一个代理和 LLM 模型了。我将使用新的 OpenAI GPT 4o mini 模型（doc），因为它比 GPT 3.5 更便宜且性能更好。

import os

# setting up credentioals
os.environ["OPENAI_MODEL_NAME"]='gpt-4o-mini'  
os.environ["OPENAI_API_KEY"] = ''

# system prompt
prompt = '''You are a senior expert in SQL and data analysis. 
So, you can help the team to gather needed data to power their decisions. 
You are very accurate and take into account all the nuances in data.
Your goal is to provide the detailed documentation for the table in database 
that will help users.'''

model = ChatOpenAI(model="gpt-4o-mini")
doc_agent = SQLAgent(model, [execute_sql], system=prompt)

LangGraph 为我们提供了非常方便的可视化图表功能。要使用它，您需要安装pygraphviz。

对于配备 M1/M2 芯片的 Mac 来说，这有点棘手，因此这里为您提供了一些小窍门：

! brew install graphviz
! python3 -m pip install -U --no-cache-dir  \
    --config-settings="--global-option=build_ext" \
    --config-settings="--global-option=-I$(brew --prefix graphviz)/include/" \
    --config-settings="--global-option=-L$(brew --prefix graphviz)/lib/" \
    pygraphviz

弄清楚安装后，这是我们的图表。

from IPython.display import Image
Image(doc_agent.graph.get_graph().draw_png())

如您所见，我们的图有循环。使用 LCEL 实现这样的功能将非常具有挑战性。

最后，是时候执行我们的代理了。我们需要将包含问题的初始消息集作为 传递HumanMessage。

messages = [HumanMessage(content="What info do we have in ecommerce_db.users table?")]
result = doc_agent.graph.invoke({"messages": messages})

在result变量中，我们可以观察到执行期间生成的所有消息。该过程按预期进行：

• 代理决定使用查询来调用该函数describe ecommerce.db_users。
• 然后，LLM 处理来自该工具的信息并提供了一个用户友好的答案。

result['messages']

# [
#   HumanMessage(content='What info do we have in ecommerce_db.users table?'), 
#   AIMessage(content='', tool_calls=[{'name': 'execute_sql', 'args': {'query': 'DESCRIBE ecommerce_db.users;'}, 'id': 'call_qZbDU9Coa2tMjUARcX36h0ax', 'type': 'tool_call'}]), 
#   ToolMessage(content='user_id\tUInt64\t\t\t\t\t\ncountry\tString\t\t\t\t\t\nis_active\tUInt8\t\t\t\t\t\nage\tUInt64\t\t\t\t\t\n', name='execute_sql', tool_call_id='call_qZbDU9Coa2tMjUARcX36h0ax'), 
#   AIMessage(content='The `ecommerce_db.users` table contains the following columns: <...>')
# ]

这是最终结果。看起来相当不错。

print(result['messages'][-1].content)

# The `ecommerce_db.users` table contains the following columns:
# 1. **user_id**: `UInt64` - A unique identifier for each user.
# 2. **country**: `String` - The country where the user is located.
# 3. **is_active**: `UInt8` - Indicates whether the user is active (1) or inactive (0).
# 4. **age**: `UInt64` - The age of the user.

使用预建代理

我们已经学习了如何从头开始构建代理。但是，我们可以利用 LangGraph 的内置功能来完成类似这样的简单任务。

我们可以使用预先构建的 ReAct 代理来获得类似的结果：可以与工具协同工作的代理。

from langgraph.prebuilt import create_react_agent
prebuilt_doc_agent = create_react_agent(model, [execute_sql],
  state_modifier = system_prompt)

它和我们之前构建的代理相同。我们稍后会尝试一下，但首先，我们需要了解另外两个重要概念：持久性和流式传输。

持久性和流式传输

持久性是指在不同交互中保持上下文的能力。当应用程序可以从用户那里获取额外输入时，持久性对于代理用例至关重要。

LangGraph 在每个步骤后自动保存状态，允许您暂停或恢复执行。此功能支持实现高级业务逻辑，例如错误恢复或人机交互。

添加持久性的最简单方法是使用内存 SQLite 数据库。

from langgraph.checkpoint.sqlite import SqliteSaver
memory = SqliteSaver.from_conn_string(":memory:")

对于现成的代理，我们可以在创建代理时将内存作为参数传递。

prebuilt_doc_agent = create_react_agent(model, [execute_sql], 
  checkpointer=memory)

如果您使用自定义代理，则需要在编译图形时将内存作为检查指针传递。

class SQLAgent:
  def __init__(self, model, tools, system_prompt = ""):
    <...>
    self.graph = graph.compile(checkpointer=memory)
    <...>

让我们执行代理并探索 LangGraph 的另一个功能：流式传输。通过流式传输，我们可以将执行的每个步骤的结果作为流中的单独事件接收。当需要同时处理多个对话（或线程）时，此功能对于生产应用程序至关重要。

LangGraph 不仅支持事件流，还支持 token 级流。我想到 token 流的唯一用例是逐字实时显示答案（类似于 ChatGPT 实现）。

让我们尝试使用新预建的代理进行流式传输。我还将使用pretty_print消息函数使结果更具可读性。

# defining thread
thread = {"configurable": {"thread_id": "1"}}
messages = [HumanMessage(content="What info do we have in ecommerce_db.users table?")]

for event in prebuilt_doc_agent.stream({"messages": messages}, thread):
    for v in event.values():
        v['messages'][-1].pretty_print()

# ================================== Ai Message ==================================
# Tool Calls:
#  execute_sql (call_YieWiChbFuOlxBg8G1jDJitR)
#  Call ID: call_YieWiChbFuOlxBg8G1jDJitR
#   Args:
#     query: SELECT * FROM ecommerce_db.users LIMIT 1;
# ================================= Tool Message =================================
# Name: execute_sql
# 1000001 United Kingdom 0 70
# 
# ================================== Ai Message ==================================
# 
# The `ecommerce_db.users` table contains at least the following information for users:
# 
# - **User ID** (e.g., `1000001`)
# - **Country** (e.g., `United Kingdom`)
# - **Some numerical value** (e.g., `0`)
# - **Another numerical value** (e.g., `70`)
# 
# The specific meaning of the numerical values and additional columns 
# is not clear from the single row retrieved. Would you like more details 
# or a broader query?

有趣的是，代理无法提供足够好的结果。由于代理没有查找表架构，因此很难猜测所有列的含义。我们可以通过使用同一线程中的后续问题来改善结果。

followup_messages = [HumanMessage(content="I would like to know the column names and types. Maybe you could look it up in database using describe.")]

for event in prebuilt_doc_agent.stream({"messages": followup_messages}, thread):
    for v in event.values():
        v['messages'][-1].pretty_print()

# ================================== Ai Message ==================================
# Tool Calls:
#   execute_sql (call_sQKRWtG6aEB38rtOpZszxTVs)
#  Call ID: call_sQKRWtG6aEB38rtOpZszxTVs
#   Args:
#     query: DESCRIBE ecommerce_db.users;
# ================================= Tool Message =================================
# Name: execute_sql
# 
# user_id UInt64     
# country String     
# is_active UInt8     
# age UInt64     
# 
# ================================== Ai Message ==================================
# 
# The `ecommerce_db.users` table has the following columns along with their data types:
# 
# | Column Name | Data Type |
# |-------------|-----------|
# | user_id     | UInt64    |
# | country     | String    |
# | is_active   | UInt8     |
# | age         | UInt64    |
# 
# If you need further information or assistance, feel free to ask!

这次，我们从代理那里得到了完整的答案。由于我们提供的是相同的线程，代理能够从之前的讨论中获取上下文。这就是持久性的工作原理。

让我们尝试改变话题并提出相同的后续问题。

new_thread = {"configurable": {"thread_id": "42"}}
followup_messages = [HumanMessage(content="I would like to know the column names and types. Maybe you could look it up in database using describe.")]

for event in prebuilt_doc_agent.stream({"messages": followup_messages}, new_thread):
    for v in event.values():
        v['messages'][-1].pretty_print()

# ================================== Ai Message ==================================
# Tool Calls:
#   execute_sql (call_LrmsOGzzusaLEZLP9hGTBGgo)
#  Call ID: call_LrmsOGzzusaLEZLP9hGTBGgo
#   Args:
#     query: DESCRIBE your_table_name;
# ================================= Tool Message =================================
# Name: execute_sql
# 
# Database returned the following error:
# Code: 60. DB::Exception: Table default.your_table_name does not exist. (UNKNOWN_TABLE) (version 23.12.1.414 (official build))
# 
# ================================== Ai Message ==================================
# 
# It seems that the table `your_table_name` does not exist in the database. 
# Could you please provide the actual name of the table you want to describe?

代理缺乏回答我们问题所需的上下文，这并不奇怪。线程旨在隔离不同的对话，确保每个线程都维护自己的上下文。

在实际应用中，管理内存至关重要。对话可能会变得非常冗长，在某些时候，每次都将整个历史记录传递给 LLM 是不切实际的。因此，值得修剪或过滤消息。我们不会在这里深入讨论细节，但您可以在LangGraph 文档中找到有关它的指导。压缩对话历史记录的另一种方法是使用摘要（示例）。

我们已经学习了如何使用 LangGraph 构建具有单个代理的系统。下一步是将多个代理组合到一个应用程序中。

多智能体系统

作为多代理工作流的一个示例，我想构建一个可以处理来自各个领域的问题的应用程序。我们将有一组专家代理，每个代理专门处理不同类型的问题，以及一个路由代理，它将找到最适合解决每个查询的专家。这样的应用程序有许多潜在的用例：从自动化客户支持到回答内部聊天中同事的问题。

首先，我们需要创建代理状态——帮助代理共同解决问题的信息。我将使用以下字段：

• question—— 初始顾客要求；
• question_type— 定义哪个代理将处理该请求的类别；
• answer— 对问题的建议答案；
• feedback— 用于将来收集一些反馈的字段。

class MultiAgentState(TypedDict):
    question: str
    question_type: str
    answer: str
    feedback: str

我没有使用任何减速器，所以我们的状态将只存储每个字段的最新版本。

然后，让我们创建一个路由器节点。它将是一个简单的 LLM 模型，定义问题的类别（数据库、LangChain 或一般问题）。

question_category_prompt = '''You are a senior specialist of analytical support. Your task is to classify the incoming questions. 
Depending on your answer, question will be routed to the right team, so your task is crucial for our team. 
There are 3 possible question types: 
- DATABASE - questions related to our database (tables or fields)
- LANGCHAIN- questions related to LangGraph or LangChain libraries
- GENERAL - general questions
Return in the output only one word (DATABASE, LANGCHAIN or  GENERAL).
'''

def router_node(state: MultiAgentState):
  messages = [
    SystemMessage(content=question_category_prompt), 
    HumanMessage(content=state['question'])
  ]
  model = ChatOpenAI(model="gpt-4o-mini")
  response = model.invoke(messages)
  return {"question_type": response.content}

现在我们有了第一个节点——路由器——让我们构建一个简单的图表来测试工作流程。

memory = SqliteSaver.from_conn_string(":memory:")

builder = StateGraph(MultiAgentState)
builder.add_node("router", router_node)

builder.set_entry_point("router")
builder.add_edge('router', END)

graph = builder.compile(checkpointer=memory)

让我们用不同类型的问题来测试我们的工作流程，看看它的实际表现如何。这将帮助我们评估路由代理是否正确地将问题分配给适当的专家代理。

thread = {"configurable": {"thread_id": "1"}}
for s in graph.stream({
    'question': "Does LangChain support Ollama?",
}, thread):
    print(s)

# {'router': {'question_type': 'LANGCHAIN'}}

thread = {"configurable": {"thread_id": "2"}}
for s in graph.stream({
    'question': "What info do we have in ecommerce_db.users table?",
}, thread):
    print(s)
# {'router': {'question_type': 'DATABASE'}}

thread = {"configurable": {"thread_id": "3"}}
for s in graph.stream({
    'question': "How are you?",
}, thread):
    print(s)

# {'router': {'question_type': 'GENERAL'}}

效果很好。我建议您逐步构建复杂图表并独立测试每个步骤。通过这种方法，您可以确保每次迭代都按预期工作，并且可以节省大量调试时间。

接下来，让我们为专家代理创建节点。我们将使用 ReAct 代理和我们之前构建的 SQL 工具作为数据库代理。

# database expert
sql_expert_system_prompt = '''
You are an expert in SQL, so you can help the team 
to gather needed data to power their decisions. 
You are very accurate and take into account all the nuances in data. 
You use SQL to get the data before answering the question.
'''

def sql_expert_node(state: MultiAgentState):
    model = ChatOpenAI(model="gpt-4o-mini")
    sql_agent = create_react_agent(model, [execute_sql],
        state_modifier = sql_expert_system_prompt)
    messages = [HumanMessage(content=state['question'])]
    result = sql_agent.invoke({"messages": messages})
    return {'answer': result['messages'][-1].content}

对于与 LangChain 相关的问题，我们将使用 ReAct 代理。为了让代理能够回答有关图书馆的问题，我们将为其配备一个搜索引擎工具。我为此选择了Tavily，因为它提供了针对 LLM 应用程序优化的搜索结果。

如果您没有帐户，您可以注册免费使用 Tavily（每月最多 1K 个请求）。要开始使用，您需要在环境变量中指定 Tavily API 密钥。

# search expert 
from langchain_community.tools.tavily_search import TavilySearchResults
os.environ["TAVILY_API_KEY"] = 'tvly-...'
tavily_tool = TavilySearchResults(max_results=5)

search_expert_system_prompt = '''
You are an expert in LangChain and other technologies. 
Your goal is to answer questions based on results provided by search.
You don't add anything yourself and provide only information baked by other sources. 
'''

def search_expert_node(state: MultiAgentState):
    model = ChatOpenAI(model="gpt-4o-mini")
    sql_agent = create_react_agent(model, [tavily_tool],
        state_modifier = search_expert_system_prompt)
    messages = [HumanMessage(content=state['question'])]
    result = sql_agent.invoke({"messages": messages})
    return {'answer': result['messages'][-1].content}

对于一般问题，我们将利用简单的 LLM 模型，而无需特定工具。

# general model
general_prompt = '''You're a friendly assistant and your goal is to answer general questions.
Please, don't provide any unchecked information and just tell that you don't know if you don't have enough info.
'''

def general_assistant_node(state: MultiAgentState):
    messages = [
        SystemMessage(content=general_prompt), 
        HumanMessage(content=state['question'])
    ]
    model = ChatOpenAI(model="gpt-4o-mini")
    response = model.invoke(messages)
    return {"answer": response.content}

最后缺少的是路由的条件函数。这很简单——我们只需要从路由器节点定义的状态传播问题类型。

def route_question(state: MultiAgentState):
    return state['question_type']

现在是时候创建我们的图表了。

builder = StateGraph(MultiAgentState)
builder.add_node("router", router_node)
builder.add_node('database_expert', sql_expert_node)
builder.add_node('langchain_expert', search_expert_node)
builder.add_node('general_assistant', general_assistant_node)
builder.add_conditional_edges(
    "router", 
    route_question,
    {'DATABASE': 'database_expert', 
     'LANGCHAIN': 'langchain_expert', 
     'GENERAL': 'general_assistant'}
)


builder.set_entry_point("router")
builder.add_edge('database_expert', END)
builder.add_edge('langchain_expert', END)
builder.add_edge('general_assistant', END)
graph = builder.compile(checkpointer=memory)

现在，我们可以通过几个问题测试该设置，看看它的表现如何。

thread = {"configurable": {"thread_id": "2"}}
results = []
for s in graph.stream({
  'question': "What info do we have in ecommerce_db.users table?",
}, thread):
  print(s)
  results.append(s)
print(results[-1]['database_expert']['answer'])

# The `ecommerce_db.users` table contains the following columns:
# 1. **User ID**: A unique identifier for each user.
# 2. **Country**: The country where the user is located.
# 3. **Is Active**: A flag indicating whether the user is active (1 for active, 0 for inactive).
# 4. **Age**: The age of the user.
# Here are some sample entries from the table:
# 
# | User ID | Country        | Is Active | Age |
# |---------|----------------|-----------|-----|
# | 1000001 | United Kingdom  | 0         | 70  |
# | 1000002 | France         | 1         | 87  |
# | 1000003 | France         | 1         | 88  |
# | 1000004 | Germany        | 1         | 25  |
# | 1000005 | Germany        | 1         | 48  |
# 
# This gives an overview of the user data available in the table.

做得好！它给出了与数据库相关的问题的相关结果。让我们尝试询问 LangChain。

thread = {"configurable": {"thread_id": "42"}}
results = []
for s in graph.stream({
    'question': "Does LangChain support Ollama?",
}, thread):
    print(s)
    results.append(s)

print(results[-1]['langchain_expert']['answer'])

# Yes, LangChain supports Ollama. Ollama allows you to run open-source 
# large language models, such as Llama 2, locally, and LangChain provides 
# a flexible framework for integrating these models into applications. 
# You can interact with models run by Ollama using LangChain, and there are 
# specific wrappers and tools available for this integration.
# 
# For more detailed information, you can visit the following resources:
# - [LangChain and Ollama Integration](https://js.langchain.com/v0.1/docs/integrations/llms/ollama/)
# - [ChatOllama Documentation](https://js.langchain.com/v0.2/docs/integrations/chat/ollama/)
# - [Medium Article on Ollama and LangChain](https://medium.com/@abonia/ollama-and-langchain-run-llms-locally-900931914a46)

太棒了！一切都进展顺利，很明显 Tavily 的搜索对于 LLM 申请非常有效。

添加人机交互

我们在创建回答问题的工具方面做得非常出色。然而，在很多情况下，让人类参与进来以批准建议的行动或提供额外的反馈是有益的。让我们添加一个步骤，在将最终结果返回给用户之前，我们可以从人类那里收集反馈。

最简单的方法是添加两个附加节点：

• human收集反馈的节点，
• 重新审视答案的节点editor，同时考虑反馈。

让我们创建这些节点：

• 人类节点：这将是一个虚拟节点，它不会执行任何操作。
• 编辑器节点：这将是一个 LLM 模型，它接收所有相关信息（客户问题、草稿答案和提供的反馈）并修改最终答案。

def human_feedback_node(state: MultiAgentState):
    pass

editor_prompt = '''You're an editor and your goal is to provide the final answer to the customer, taking into account the feedback. 
You don't add any information on your own. You use friendly and professional tone.
In the output please provide the final answer to the customer without additional comments.
Here's all the information you need.

Question from customer: 
----
{question}
----
Draft answer:
----
{answer}
----
Feedback: 
----
{feedback}
----
'''

def editor_node(state: MultiAgentState):
  messages = [
    SystemMessage(content=editor_prompt.format(question = state['question'], answer = state['answer'], feedback = state['feedback']))
  ]
  model = ChatOpenAI(model="gpt-4o-mini")
  response = model.invoke(messages)
  return {"answer": response.content}

让我们将这些节点添加到我们的图中。此外，我们需要在人工节点之前引入中断，以确保流程暂停以获得人工反馈。

builder = StateGraph(MultiAgentState)
builder.add_node("router", router_node)
builder.add_node('database_expert', sql_expert_node)
builder.add_node('langchain_expert', search_expert_node)
builder.add_node('general_assistant', general_assistant_node)
builder.add_node('human', human_feedback_node)
builder.add_node('editor', editor_node)

builder.add_conditional_edges(
  "router", 
  route_question,
  {'DATABASE': 'database_expert', 
  'LANGCHAIN': 'langchain_expert', 
  'GENERAL': 'general_assistant'}
)


builder.set_entry_point("router")

builder.add_edge('database_expert', 'human')
builder.add_edge('langchain_expert', 'human')
builder.add_edge('general_assistant', 'human')
builder.add_edge('human', 'editor')
builder.add_edge('editor', END)
graph = builder.compile(checkpointer=memory, interrupt_before = ['human'])

现在，当我们运行图表时，执行将在人类节点之前停止。

thread = {"configurable": {"thread_id": "2"}}

for event in graph.stream({
    'question': "What are the types of fields in ecommerce_db.users table?",
}, thread):
    print(event)


# {'question_type': 'DATABASE', 'question': 'What are the types of fields in ecommerce_db.users table?'}
# {'router': {'question_type': 'DATABASE'}}
# {'database_expert': {'answer': 'The `ecommerce_db.users` table has the following fields:\n\n1. **user_id**: UInt64\n2. **country**: String\n3. **is_active**: UInt8\n4. **age**: UInt64'}}

让我们获取客户输入并根据反馈更新状态。

user_input = input("Do I need to change anything in the answer?")
# Do I need to change anything in the answer? 
# It looks wonderful. Could you only make it a bit friendlier please?

graph.update_state(thread, {"feedback": user_input}, as_node="human")

我们可以检查状态以确认反馈已填充并且序列中的下一个节点是editor。

print(graph.get_state(thread).values['feedback'])
# It looks wonderful. Could you only make it a bit friendlier please?

print(graph.get_state(thread).next)
# ('editor',)

我们可以继续执行。None作为输入传递将从暂停点恢复该过程。

for event in graph.stream(None, thread, stream_mode="values"):
  print(event)

print(event['answer'])

# Hello! The `ecommerce_db.users` table has the following fields:
# 1. **user_id**: UInt64
# 2. **country**: String
# 3. **is_active**: UInt8
# 4. **age**: UInt64
# Have a nice day!

编辑考虑了我们的反馈，并在我们的最终消息中添加了一些礼貌用语。这是一个非常棒的结果！

我们可以为编辑器配备人性化工具，以更具代理性的方式实现人机交互。

让我们调整一下编辑器。我稍微修改了提示并将工具添加到代理中。

from langchain_community.tools import HumanInputRun
human_tool = HumanInputRun()

editor_agent_prompt = '''You're an editor and your goal is to provide the final answer to the customer, taking into the initial question.
If you need any clarifications or need feedback, please, use human. Always reach out to human to get the feedback before final answer.
You don't add any information on your own. You use friendly and professional tone. 
In the output please provide the final answer to the customer without additional comments.
Here's all the information you need.

Question from customer: 
----
{question}
----
Draft answer:
----
{answer}
----
'''

model = ChatOpenAI(model="gpt-4o-mini")
editor_agent = create_react_agent(model, [human_tool])
messages = [SystemMessage(content=editor_agent_prompt.format(question = state['question'], answer = state['answer']))]
editor_result = editor_agent.invoke({"messages": messages})

# Is the draft answer complete and accurate for the customer's question about the types of fields in the ecommerce_db.users table?
# Yes, but could you please make it friendlier.

print(editor_result['messages'][-1].content)
# The `ecommerce_db.users` table has the following fields:
# 1. **user_id**: UInt64
# 2. **country**: String
# 3. **is_active**: UInt8
# 4. **age**: UInt64
# 
# If you have any more questions, feel free to ask!

因此，编辑向人工提出了以下问题：“对于客户关于 ecommerce_db.users 表中字段类型的问题，草稿答案是否完整准确？”。收到反馈后，编辑完善了答案，使其更加用户友好。

让我们更新主图以纳入新代理，而不是使用两个单独的节点。使用这种方法，我们不再需要中断。

def editor_agent_node(state: MultiAgentState):
  model = ChatOpenAI(model="gpt-4o-mini")
  editor_agent = create_react_agent(model, [human_tool])
  messages = [SystemMessage(content=editor_agent_prompt.format(question = state['question'], answer = state['answer']))]
  result = editor_agent.invoke({"messages": messages})
  return {'answer': result['messages'][-1].content}

builder = StateGraph(MultiAgentState)
builder.add_node("router", router_node)
builder.add_node('database_expert', sql_expert_node)
builder.add_node('langchain_expert', search_expert_node)
builder.add_node('general_assistant', general_assistant_node)
builder.add_node('editor', editor_agent_node)

builder.add_conditional_edges(
  "router", 
  route_question,
  {'DATABASE': 'database_expert', 
   'LANGCHAIN': 'langchain_expert', 
    'GENERAL': 'general_assistant'}
)

builder.set_entry_point("router")

builder.add_edge('database_expert', 'editor')
builder.add_edge('langchain_expert', 'editor')
builder.add_edge('general_assistant', 'editor')
builder.add_edge('editor', END)

graph = builder.compile(checkpointer=memory)

thread = {"configurable": {"thread_id": "42"}}
results = []

for event in graph.stream({
  'question': "What are the types of fields in ecommerce_db.users table?",
}, thread):
  print(event)
  results.append(event)
  结果.append（事件）

此图表的工作方式与上一个图表类似。我个人更喜欢这种方法，因为它利用了工具，使解决方案更加灵活。例如，代理可以多次联系人工并根据需要优化问题。

就是这样。我们构建了一个多代理系统，它可以回答来自不同领域的问题，并考虑人类的反馈。

概括

在本文中，我们探索了 LangGraph 库及其用于构建单代理和多代理工作流的应用程序。我们已经研究了它的一系列功能，现在是时候总结它的优点和缺点了。

总的来说，我发现 LangGraph 是一个用于构建复杂 LLM 应用程序的非常强大的框架：

• LangGraph 是一个低级框架，提供广泛的定制选项，让您可以构建您所需要的内容。
• 由于 LangGraph 建立在 LangChain 之上，因此它可以无缝集成到其生态系统中，从而轻松利用现有的工具和组件。

然而，LangGrpah 仍有一些可以改进的地方：

• LangGraph 的灵活性伴随着更高的入门门槛。虽然您可以在 15-30 分钟内了解 CrewAI 的概念，但需要一些时间才能熟悉并掌握 LangGraph。
• LangGraph 为您提供了更高级别的控制，但它缺少 CrewAI 的一些很酷的预构建功能，例如协作或即用型RAG工具。
• LangGraph 不像 CrewAI 那样强制执行最佳实践（例如角色扮演或护栏）。因此，它可能会导致较差的结果。

我想说 CrewAI 对于新手和常见用例来说是一个更好的框架，因为它可以帮助您快速获得良好的结果并提供指导以防止错误。

如果您想构建高级应用程序并需要更多控制，LangGraph 是您的最佳选择。请记住，您需要花时间学习 LangGraph，并对最终解决方案负全部责任，因为该框架不会提供指导来帮助您避免常见错误。

参考：

https://towardsdatascience.com/from-basics-to-advanced-exploring-langgraph-e8c1cf4db787