LangGraph 函数式 API 简介

您是否曾经希望利用 LangGraph 的核心功能，例如人机环路、持久化/内存和流式传输，而无需显式定义图？

我们很高兴地宣布发布 LangGraph 的 函数式 API，该 API 在 Python 和 JavaScript 中可用。

函数式 API 允许您使用更传统的编程范例来利用 LangGraph 功能，从而更轻松地构建包含人机环路交互、短期和长期记忆以及流式传输功能的 AI 工作流程。

函数式 API 由两个装饰器组成——entrypoint 和 task——它们允许您使用标准函数定义工作流程，并使用常规循环和条件语句来控制执行流程。这使得在现有应用程序中采用 LangGraph 的功能变得容易，而无需重构代码。

此 API 是 Graph API (StateGraph) 的补充，可以与它结合使用，因为两个 API 都使用相同的底层运行时。这允许您混合和匹配这两种范例，以创建利用两者优势的复杂工作流程。

在这篇文章中，我们将了解如何使用函数式 API 来利用 LangGraph 的关键功能。

构建模块

函数式 API 使用两个原语来定义工作流程

入口点 (Entrypoint)：工作流程的起点，封装了工作流程逻辑并管理执行流程，包括处理长时间运行的 任务 (tasks) 和中断。
任务 (Task)：一个离散的工作单元，例如 API 调用或数据处理步骤，可以从 入口点 (entrypoint) 内异步执行。调用任务会返回一个类似 Future 的对象，可以等待该对象以获取结果或同步解析。

人机环路

假设您正在构建一个内容生成应用程序，帮助用户创建文章。在最终确定输出之前，您的用户需要审查并批准草稿。

以下是它的工作方式：AI 撰写草稿，然后暂停等待用户反馈。一旦他们批准或拒绝，系统就会从中断的地方继续，无需重新运行整个工作流程或与复杂的状态管理作斗争。

如果没有合适的工具，您将不得不自己构建持久化层和管道逻辑。但是借助 LangGraph 的人机环路功能和函数式 API，这变得非常简单。使用 interrupt 函数，您可以在等待用户输入时无限期地暂停工作流程。当用户输入被收集后，您可以使用 Command 原语恢复，由于任务结果持久化，因此可以跳过先前完成的任务。

from langgraph.checkpoint.memory import MemorySaver
from langgraph.func import entrypoint, task
from langgraph.types import interrupt

@task
def write_essay(topic: str) -> str:
  """Write an essay about the given topic."""
  time.sleep(1) # This is a placeholder for a long-running task.
  return f"An essay about topic: {topic}"

@entrypoint(checkpointer=MemorySaver())
def workflow(topic: str) -> dict:
  """A simple workflow that writes an essay and asks for a review."""
  essay = write_essay("cat").result()
  is_approved = interrupt({
    # Any json-serializable payload provided to interrupt as argument.
    # It will be surfaced on the client side as an Interrupt when streaming data
    # from the workflow.
    "essay": essay, # The essay we want reviewed.
    # We can add any additional information that we need.
    # For example, introduce a key called "action" with some instructions.
    "action": "Please approve/reject the essay",
  })

  return {
    "essay": essay, # The essay that was generated
    "is_approved": is_approved, # Response from HIL
  }

一个撰写文章草稿并暂停以供人工审核的工作流程示例。

为什么这很重要：人机环路（或“环上”）工作流程将人工输入融入自动化流程，允许在最重要的地方进行审查、验证或更正。这种方法在基于 LLM 的应用程序中非常宝贵，因为偶尔会出现不准确的情况。对于低容错用例（如合规性、决策制定或内容创建），人工参与通过在关键阶段启用审查、覆盖或调整来确保可靠性。

有关上述工作流程的更详细示例，请查看函数式 API 文档（Python 和 JavaScript）。
有关人机环路模式的概述，请参阅概念文档（Python 和 JavaScript）。
有关持久性的更多信息，请参阅：Python, JavaScript。

短期记忆

构建聊天机器人或对话代理？您需要短期记忆来跟踪对话历史记录——否则，您的聊天机器人将无法与用户保持连贯的对话。

在函数式 API 中，您可以使用以下方法处理短期记忆

previous 参数：自动为您提供对话线程中上次检查点的状态。
entrypoint.final() 类型：允许您为工作流程返回最终值，并可选择为下一个检查点保存不同的值。

from langgraph.checkpoint.memory import MemorySaver
from langgraph.func import entrypoint

# Set a checkpointer to enable persistence.
# Additional implementations are available.
checkpointer = MemorySaver()

@entrypoint(checkpointer=checkpointer)
def conversational_agent(user_message, *, previous: Any = None):
    # Initialize messages list from previous state
    messages = previous or []

    # Add the new user message to the conversation history
    messages.append(user_message)

    # Get agent's response based on conversation history.
    # Replace with call_llm with actual implementation.
    new_messages = call_llm(messages) 

    # Add agent's messages to conversation history
    messages.extend(new_messages)

    # Return agent's messages as output 
    # while saving full conversation history
    return entrypoint.final(value=new_messages, save=messages)

对话代理实现示例

LangGraph 的内置持久化层允许您实现短期记忆，该记忆可以维护对话历史记录并为多个用户工作，而无需复杂的设置或管理。

有关更多详细信息，请参阅有关在 Python 和 JavaScript 中添加长期记忆（跨线程持久性）的操作指南。
您可以在此处找到有关内存的更多概念信息：Python 和 JavaScript。

长期记忆

某些应用程序，如推荐系统或个人助理，需要记住用户偏好以提供更好的体验。这称为 长期记忆——您的应用程序通过存储和更新有关用户在不同对话中的信息，随着时间的推移学习和适应。

您可以在 LangGraph 中使用函数式 API 中的 store 参数来实现长期记忆。 store 参数提供对持久存储层的访问，该存储层可用于存储和检索与同一用户的不同交互之间的数据。

from langgraph.checkpoint.memory import MemorySaver
from langgraph.func import entrypoint
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore

# Set a checkpointer to enable persistence.
# Additional implementations are available.
checkpointer = MemorySaver()

# Configure a store for long-term memory.
# Additional implementations are available.
store = InMemoryStore()

@entrypoint(checkpointer=checkpointer, store=store)
def workflow(
    some_input: dict,
    *,
    store: BaseStore
):
    # Use the store for long-term memory operations
    stored_data = store.get("user_info")
    # Your workflow will also be able to update the stored data.
    # A common way to do this is by having an LLM invoke tool calls that
    # update the stored data based on the conversation, user input, etc.

工作流程可以使用 LangGraph 的 BaseStore 接口来实现长期记忆。

有关更多详细信息，请参阅有关在 Python 和 JavaScript 中添加长期记忆（跨线程持久性）的操作指南。
您可以在此处找到有关内存的更多概念信息：Python 和 JavaScript。

流式传输

为最终用户构建响应式应用程序？实时更新是保持用户参与应用程序进度的关键。

您需要流式传输三种主要类型的数据

工作流程进度（例如，“任务 1 已完成”）。
LLM 令牌（在生成时）。
自定义更新（例如，“已获取 10/100 条记录”）。

LangGraph 通过内置的流式传输支持使这变得容易。当您定义入口点时，您将获得一个 stream 方法来发送实时数据。它返回一个生成器，因此您可以按发生顺序产生更新。您可以使用 stream_mode 参数订阅不同的流，订阅工作流程进度 (updates)、LLM 令牌 (messages) 或自定义数据 (custom)。

from langgraph.func import entrypoint
from langgraph.types import StreamWriter

@entrypoint(checkpointer=checkpointer)
def workflow(inputs, writer: StreamWriter):
  writer("Processing started")  # Write to custom stream
  # Do stuff (e.g., call tasks, call llms)
  writer("Processing completed")
  return result

# Consume the stream
for chunk in main.stream(input_data, stream_mode=["custom", "updates", "messages"], config=config):
  print(chunk)

您可以在此处找到有关流式传输的更多概念信息：Python 和 JavaScript。

可观测性

函数式 API 提供内置的可观测性功能，以监控和调试工作流程。 入口点 (entrypoints) 和任务的输入和输出可以记录到 LangSmith，LangChain 的可观测性平台。这使您可以跟踪工作流程的进度，识别瓶颈，并排除问题并改进您的工作流程。

部署

如果您使用 入口点 (entrypoint) 创建了工作流程，则可以使用 LangGraph 平台将其部署到生产环境。

Graph API 与函数式 API

函数式 API 和 Graph API (StateGraph) 提供了两种不同的范例，用于在 LangGraph 中创建。以下是一些主要区别

控制流： 函数式 API 不需要考虑图结构。您可以使用标准 Python 结构来定义工作流程。这通常会减少您需要为控制流编写的代码量。
状态管理： Graph API 需要声明状态，并且可能需要定义 reducer 来管理对图状态的更新。 入口点 (entrypoints) 和 任务 (tasks) 不需要显式的状态管理，因为它们的状态范围限定为函数，并且不在函数之间共享。
时间旅行： 在 Graph API 中，检查点更加精细，在每次节点执行后（或者如果某些节点并行执行，则在节点组之后）生成。在 函数式 API 中，检查点在每次 入口点 (entrypoint) 执行后生成。当任务执行时，它们会更新与 入口点 (entrypoint) 关联的现有检查点，但它不会生成新的检查点。因此，时间旅行在 Graph API 中得到更好的支持。
可视化： Graph API 可以轻松地将工作流程可视化为图形，这对于调试、理解工作流程以及与他人共享非常有用。 函数式 API 不支持可视化，因为执行流程是在运行时动态生成的。

由于 函数式 API 和 Graph API 都使用相同的底层运行时，因此您可以在同一项目中混合和匹配它们。例如，您可以从入口点调用图形，或者可以在图形中使用任务等。

结论

LangGraph 中的 函数式 API 提供了一种构建 AI 工作流程的灵活方法，具有强大的功能，例如人机环路交互、状态管理、持久性和流式传输。这些功能使开发人员能够创建复杂的应用程序，有效地将自动化与人工输入相结合。