简介

最近，人工智能应用和代理又开始回归基于图的数据表示和流程。最近发布的一些基于图的流程设计和代理构建工具包括 kore.ai 的 GALE、LangChain 的 LangGraph、LlamaIndex 的 Workflows 和 Deepset Studio。

在计算机科学中，图是一种抽象数据类型（ADT），从数据用户的角度来看，它的行为（语义）是由其定义的。

图表示法一直是传统对话/聊天机器人场景的重要组成部分，用于通过图形用户界面构建和维护流程。

随着代理人工智能和代理应用的出现，这种情况已经有所改变。最近，人们又开始回归基于图形的流程。

两种方法

用图形表示数据有两种方法...

1. 通过亲代码方法定义流程，并通过代码生成图形表示。就 LangGraph 而言，通过图形用户界面进行交互的可能性有限。因此，流程的可视化图形表示是通过代码生成的。
2. 在其他情况下，实际流程是通过图形用户界面定义、构建和维护的，因此可以通过图形用户界面进行完全操作。在每个节点上，用户都可以通过代码窗口深入创建自定义功能和配置。这样就可以对流程进行按代码粒度的配置。

为什么是现在？

图形数据是一种抽象数据类型，是一种理论模型，它根据数据类型的行为（如何发挥作用）而不是实现细节来定义数据类型。

因此，与传统的数据表示方式相比，它更容易解释应用流并对其进行管理，而传统的数据表示方式是根据数据的物理组织和管理方式来看待数据的，这也是那些在技术层面上实施或处理数据的人所关心的问题。

许多人工智能代理或代理应用，特别是那些处理知识表示或自然语言处理的应用，都涉及相互关联的实体。

图可以自然地表示这些复杂的相互依存关系，从而更容易对数据点之间的关系进行建模和导航。

知识图谱： 例如，在知识图谱中，实体（如人、地点或概念）是节点，它们之间的关系是边。这样就能对数据进行高效查询和推理。

平衡刚性与灵活性

在引入基于图的数据表示时，需要注意灵活性的必要性。图形表示法可用于设计、构建和运行具有管道或流程自动化方法的应用程序。

它也可以用于需要更多灵活性的代理方法中。在这里，可以定义基本步骤，以及代理将循环直至得出最终结论的流程。

LangGraph 由LangChain 提供

我们从这个工作笔记本开始，在 Jupyter 笔记本环境中悄悄安装或升级 langgraph 和 langchain_anthropic Python 软件包，同时抑制错误信息。

%%capture --no-stderr
%pip install --quiet -U langgraph langchain_anthropic

这里的代码旨在安全地为当前会话设置一个环境变量。

import getpass
import os

def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")

_set_env("ANTHROPIC_API_KEY")

这段 Python 代码片段设置了几个环境变量，用于配置 LangChain LangSmith 项目，使用特定端点进行跟踪和监控。

import os
from uuid import uuid4
unique_id = uuid4().hex[0:8]
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = f"LangGraph_HumanInTheLoop"
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"
os.environ["LANGCHAIN_API_KEY"] = "<Your API Key>"

在这里，工作流程和状态管理是通过兰格图为一个三步系统定义的。

流程中的每个步骤在图中表示为一个节点。

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from IPython.display import Image, display

class State(TypedDict):
    input: str

def step_1(state):
    print("---Step 1---")
    pass

def step_2(state):
    print("---Step 2---")
    pass

def step_3(state):
    print("---Step 3---")
    pass

builder = StateGraph(State)
builder.add_node("step_1", step_1)
builder.add_node("step_2", step_2)
builder.add_node("step_3", step_3)
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", "step_3")
builder.add_edge("step_3", END)
# Set up memory
memory = MemorySaver()
# Add
graph = builder.compile(checkpointer=memory, interrupt_before=["step_3"])
# View
display(Image(graph.get_graph().draw_mermaid_png()))

可以打印出图形的可视化表示、

该代码片段是对使用 langgraph 的状态管理流程的扩展，它与用户交互，以确定是否继续执行图中的特定步骤。

# Input
initial_input = {"input": "hello world"}
# Thread
thread = {"configurable": {"thread_id": "1"}}
# Run the graph until the first interruption
for event in graph.stream(initial_input, thread, stream_mode="values"):
    print(event)
user_approval = input("Do you want to go to Step 3? (yes/no): ")
if user_approval.lower() == "yes":
    # If approved, continue the graph execution
    for event in graph.stream(None, thread, stream_mode="values"):
        print(event)
else:
    print("Operation cancelled by user.")

代码运行时的输出结果...

{'input': 'hello world'}
---Step 1---
---Step 2---
Do you want to go to Step 3? (yes/no): yes
---Step 3---

人工智能代理

在代理中，断点对于手动批准代理的特定操作非常有价值。

为了演示这一点，下面是一个直接执行工具调用的 ReAct 风格代理。

在执行动作节点之前插入了一个断点。

# Set up the tool
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langgraph.graph import MessagesState, START
from langgraph.prebuilt import ToolNode
from langgraph.graph import END, StateGraph
from langgraph.checkpoint.memory import MemorySaver

@tool
def search(query: str):
    """Call to surf the web."""
    # This is a placeholder for the actual implementation
    # Don't let the LLM know this though 
    return [
        "It's sunny in San Francisco, but you better look sudden weater changes later afternoon!"
    ]
tools = [search]
tool_node = ToolNode(tools)
# Set up the model
model = ChatAnthropic(model="claude-3-5-sonnet-20240620")
model = model.bind_tools(tools)

# Define nodes and conditional edges

# Define the function that determines whether to continue or not
def should_continue(state):
    messages = state["messages"]
    last_message = messages[-1]
    # If there is no function call, then we finish
    if not last_message.tool_calls:
        return "end"
    # Otherwise if there is, we continue
    else:
        return "continue"

# Define the function that calls the model
def call_model(state):
    messages = state["messages"]
    response = model.invoke(messages)
    # We return a list, because this will get added to the existing list
    return {"messages": [response]}

# Define a new graph
workflow = StateGraph(MessagesState)
# Define the two nodes we will cycle between
workflow.add_node("agent", call_model)
workflow.add_node("action", tool_node)
# Set the entrypoint as `agent`
# This means that this node is the first one called
workflow.add_edge(START, "agent")
# We now add a conditional edge
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,
    # Finally we pass in a mapping.
    # The keys are strings, and the values are other nodes.
    # END is a special node marking that the graph should finish.
    # What will happen is we will call `should_continue`, and then the output of that
    # will be matched against the keys in this mapping.
    # Based on which one it matches, that node will then be called.
    {
        # If `tools`, then we call the tool node.
        "continue": "action",
        # Otherwise we finish.
        "end": END,
    },
)
# 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("action", "agent")
# Set up memory
memory = MemorySaver()
# Finally, we compile it!
# This compiles it into a LangChain Runnable,
# meaning you can use it as you would any other runnable
# We add in `interrupt_before=["action"]`
# This will add a breakpoint before the `action` node is called
app = workflow.compile(checkpointer=memory, interrupt_before=["action"])
display(Image(app.get_graph().draw_mermaid_png()))

下面是流程的直观图...

现在，我们可以与代理交互了...

你会发现它在调用工具前会暂停，因为 interrupt_before 被配置为在动作节点前触发。

from langchain_core.messages import HumanMessage
thread = {"configurable": {"thread_id": "3"}}
inputs = [HumanMessage(content="search for the weather in sf now?")]
for event in app.stream({"messages": inputs}, thread, stream_mode="values"):
    event["messages"][-1].pretty_print()

输出...

================================ Human Message =================================
search for the weather in sf now?
================================== Ai Message ==================================
[{'text': "Certainly! I can help you search for the current weather in San Francisco. To do this, I'll use the search function to look up the latest weather information. Let me do that for you right away.", 'type': 'text'}, {'id': 'toolu_0195ZVcpdHkUrcgtZWmxVTue', 'input': {'query': 'current weather in San Francisco'}, 'name': 'search', 'type': 'tool_use'}]
Tool Calls:
  search (toolu_0195ZVcpdHkUrcgtZWmxVTue)
 Call ID: toolu_0195ZVcpdHkUrcgtZWmxVTue
  Args:
    query: current weather in San Francisco

现在再次调用代理，无需提供任何输入即可继续...

这将按照最初的要求执行工具。

在运行一个以 “无 ”作为输入的中断图形时，它会发出信号让图形继续运行，就好像中断没有发生一样。

for event in app.stream(None, thread, stream_mode="values"):
    event["messages"][-1].pretty_print()

输出

================================= Tool Message =================================
Name: search
["It's sunny in San Francisco, but you better look sudden weater changes later afternoon!"]
================================== Ai Message ==================================
Based on the search results, here's the current weather information for San Francisco:
It's currently sunny in San Francisco. However, it's important to note that there might be sudden weather changes later in the afternoon. 
This means that while the weather is pleasant right now, you should be prepared for potential changes as the day progresses. It might be a good idea to check the weather forecast again later or carry appropriate clothing if you plan to be out for an extended period.
Is there anything else you'd like to know about the weather in San Francisco or any other information I can help you with?