> ## Documentation Index
> Fetch the complete documentation index at: https://langchain-docs-zh.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# 使用函数式 API

> 功能 API 允许您通过最小化对现有代码的修改，将 LangGraph 的关键特性（持久化、记忆、人机交互 和 流式输出）添加到您的应用程序中。

[**功能 API**](https://langchain-docs-zh.mintlify.site/langchain/oss/langgraph/functional-api.md) 允许您通过最小化对现有代码的修改，将 LangGraph 的关键特性（[持久化](https://langchain-docs-zh.mintlify.site/langchain/oss/langgraph/persistence.md)、[记忆](https://langchain-docs-zh.mintlify.site/langchain/oss/langgraph/add-memory.md)、[人机交互](https://langchain-docs-zh.mintlify.site/langchain/oss/langgraph/interrupts.md) 和 [流式输出](https://langchain-docs-zh.mintlify.site/langchain/oss/langgraph/streaming.md)）添加到您的应用程序中。

> [!TIP]
>
> 有关功能 API 的概念信息，请参阅 [功能 API](https://langchain-docs-zh.mintlify.site/langchain/oss/langgraph/functional-api.md)。

## 创建简单工作流

在定义一个 `entrypoint` 时，输入仅限于函数的第一个参数。若要传递多个输入，可以使用字典。

:::python

```python theme={null}
@entrypoint(checkpointer=checkpointer)
def my_workflow(inputs: dict) -> int:
    value = inputs["value"]
    another_value = inputs["another_value"]
    ...

my_workflow.invoke({"value": 1, "another_value": 2})
```

:::

:::js

```typescript theme={null}
const checkpointer = new MemorySaver();

const myWorkflow = entrypoint(
  { checkpointer, name: "myWorkflow" },
  async (inputs: { value: number; anotherValue: number }) => {
    const value = inputs.value;
    const anotherValue = inputs.anotherValue;
    // ...
  }
);

await myWorkflow.invoke({ value: 1, anotherValue: 2 });
```

:::

> [accordion] Extended example: simple workflow
>
> :::python
>
> ```python theme={null}
> from langchain_core.utils.uuid import uuid7
> from langgraph.func import entrypoint, task
> from langgraph.checkpoint.memory import InMemorySaver
>
> # Task that checks if a number is even
> @task
> def is_even(number: int) -> bool:
>     return number % 2 == 0
>
> # Task that formats a message
> @task
> def format_message(is_even: bool) -> str:
>     return "The number is even." if is_even else "The number is odd."
>
> # Create a checkpointer for persistence
> checkpointer = InMemorySaver()
>
> @entrypoint(checkpointer=checkpointer)
> def workflow(inputs: dict) -> str:
>     """Simple workflow to classify a number."""
>     even = is_even(inputs["number"]).result()
>     return format_message(even).result()
>
> # Run the workflow with a unique thread ID
> config = {"configurable": {"thread_id": str(uuid7())}}
> result = workflow.invoke({"number": 7}, config=config)
> print(result)
> ```
>
> :::
>
> :::js
>
> ```typescript theme={null}
> import { v7 as uuid7 } from "uuid";
> import { entrypoint, task, MemorySaver } from "@langchain/langgraph";
>
> // Task that checks if a number is even
> const isEven = task("isEven", async (number: number) => {
>   return number % 2 === 0;
> });
>
> // Task that formats a message
> const formatMessage = task("formatMessage", async (isEven: boolean) => {
>   return isEven ? "The number is even." : "The number is odd.";
> });
>
> // Create a checkpointer for persistence
> const checkpointer = new MemorySaver();
>
> const workflow = entrypoint(
>   { checkpointer, name: "workflow" },
>   async (inputs: { number: number }) => {
>     // Simple workflow to classify a number
>     const even = await isEven(inputs.number);
>     return await formatMessage(even);
>   }
> );
>
> // Run the workflow with a unique thread ID
> const config = { configurable: { thread_id: uuid7() } };
> const result = await workflow.invoke({ number: 7 }, config);
> console.log(result);
> ```
>
> :::

> [accordion] Extended example: Compose an essay with an LLM
>
> 此示例展示了如何使用 `@task` 和 `@entrypoint` 装饰器的语法。由于提供了检查点器，工作流的结果将被持久化到检查点器中。
>
> :::python
>
> ```python theme={null}
> from langchain_core.utils.uuid import uuid7
> from langchain.chat_models import init_chat_model
> from langgraph.func import entrypoint, task
> from langgraph.checkpoint.memory import InMemorySaver
>
> model = init_chat_model('gpt-3.5-turbo')
>
> # Task: generate essay using an LLM
> @task
> def compose_essay(topic: str) -> str:
>     """Generate an essay about the given topic."""
>     return model.invoke([
>         {"role": "system", "content": "You are a helpful assistant that writes essays."},
>         {"role": "user", "content": f"Write an essay about {topic}."}
>     ]).content
>
> # Create a checkpointer for persistence
> checkpointer = InMemorySaver()
>
> @entrypoint(checkpointer=checkpointer)
> def workflow(topic: str) -> str:
>     """Simple workflow that generates an essay with an LLM."""
>     return compose_essay(topic).result()
>
> # Execute the workflow
> config = {"configurable": {"thread_id": str(uuid7())}}
> result = workflow.invoke("the history of flight", config=config)
> print(result)
> ```
>
> :::
>
> :::js
>
> ```typescript theme={null}
> import { v7 as uuid7 } from "uuid";
> import { ChatOpenAI } from "@langchain/openai";
> import { entrypoint, task, MemorySaver } from "@langchain/langgraph";
>
> const model = new ChatOpenAI({ model: "gpt-3.5-turbo" });
>
> // Task: generate essay using an LLM
> const composeEssay = task("composeEssay", async (topic: string) => {
>   // Generate an essay about the given topic
>   const response = await model.invoke([
>     { role: "system", content: "You are a helpful assistant that writes essays." },
>     { role: "user", content: `Write an essay about ${topic}.` }
>   ]);
>   return response.content as string;
> });
>
> // Create a checkpointer for persistence
> const checkpointer = new MemorySaver();
>
> const workflow = entrypoint(
>   { checkpointer, name: "workflow" },
>   async (topic: string) => {
>     // Simple workflow that generates an essay with an LLM
>     return await composeEssay(topic);
>   }
> );
>
> // Execute the workflow
> const config = { configurable: { thread_id: uuid7() } };
> const result = await workflow.invoke("the history of flight", config);
> console.log(result);
> ```
>
> :::

## 并行执行

通过并发调用任务并等待结果，可以并行执行任务。这对于提高 IO 密集型任务的性能（例如调用 LLM API）很有用。

:::python

```python theme={null}
@task
def add_one(number: int) -> int:
    return number + 1

@entrypoint(checkpointer=checkpointer)
def graph(numbers: list[int]) -> list[str]:
    futures = [add_one(i) for i in numbers]
    return [f.result() for f in futures]
```

:::

:::js

```typescript theme={null}
const addOne = task("addOne", async (number: number) => {
  return number + 1;
});

const graph = entrypoint(
  { checkpointer, name: "graph" },
  async (numbers: number[]) => {
    return await Promise.all(numbers.map(addOne));
  }
);
```

:::

> [accordion] Extended example: parallel LLM calls
>
> 这个示例展示了如何使用 `@task` 并行运行多个 LLM 调用。每个调用生成关于不同主题的一个段落，并将结果连接成一个文本输出。
>
> :::python
>
> ```python theme={null}
> from langchain_core.utils.uuid import uuid7
> from langchain.chat_models import init_chat_model
> from langgraph.func import entrypoint, task
> from langgraph.checkpoint.memory import InMemorySaver
>
> # Initialize the LLM model
> model = init_chat_model("gpt-3.5-turbo")
>
> # Task that generates a paragraph about a given topic
> @task
> def generate_paragraph(topic: str) -> str:
>     response = model.invoke([
>         {"role": "system", "content": "You are a helpful assistant that writes educational paragraphs."},
>         {"role": "user", "content": f"Write a paragraph about {topic}."}
>     ])
>     return response.content
>
> # Create a checkpointer for persistence
> checkpointer = InMemorySaver()
>
> @entrypoint(checkpointer=checkpointer)
> def workflow(topics: list[str]) -> str:
>     """Generates multiple paragraphs in parallel and combines them."""
>     futures = [generate_paragraph(topic) for topic in topics]
>     paragraphs = [f.result() for f in futures]
>     return "\n\n".join(paragraphs)
>
> # Run the workflow
> config = {"configurable": {"thread_id": str(uuid7())}}
> result = workflow.invoke(["quantum computing", "climate change", "history of aviation"], config=config)
> print(result)
> ```
>
> :::
>
> :::js
>
> ```typescript theme={null}
> import { v7 as uuid7 } from "uuid";
> import { ChatOpenAI } from "@langchain/openai";
> import { entrypoint, task, MemorySaver } from "@langchain/langgraph";
>
> // Initialize the LLM model
> const model = new ChatOpenAI({ model: "gpt-3.5-turbo" });
>
> // Task that generates a paragraph about a given topic
> const generateParagraph = task("generateParagraph", async (topic: string) => {
>   const response = await model.invoke([
>     { role: "system", content: "You are a helpful assistant that writes educational paragraphs." },
>     { role: "user", content: `Write a paragraph about ${topic}.` }
>   ]);
>   return response.content as string;
> });
>
> // Create a checkpointer for persistence
> const checkpointer = new MemorySaver();
>
> const workflow = entrypoint(
>   { checkpointer, name: "workflow" },
>   async (topics: string[]) => {
>     // Generates multiple paragraphs in parallel and combines them
>     const paragraphs = await Promise.all(topics.map(generateParagraph));
>     return paragraphs.join("\n\n");
>   }
> );
>
> // Run the workflow
> const config = { configurable: { thread_id: uuid7() } };
> const result = await workflow.invoke(["quantum computing", "climate change", "history of aviation"], config);
> console.log(result);
> ```
>
> :::
>
> 这个示例使用 LangGraph 的并发模型来提高执行时间，特别是当任务涉及 IO（如 LLM 完成时）。

## 调用图

**函数式 API** 和 [**图 API**](https://langchain-docs-zh.mintlify.site/langchain/oss/langgraph/graph-api.md) 可以在同一个应用程序中一起使用，因为它们共享相同的底层运行时。

:::python

```python theme={null}
from langgraph.func import entrypoint
from langgraph.graph import StateGraph

builder = StateGraph()
...
some_graph = builder.compile()

@entrypoint()
def some_workflow(some_input: dict) -> int:
    # Call a graph defined using the graph API
    result_1 = some_graph.invoke(...)
    # Call another graph defined using the graph API
    result_2 = another_graph.invoke(...)
    return {
        "result_1": result_1,
        "result_2": result_2
    }
```

:::

:::js

```typescript theme={null}
import { entrypoint } from "@langchain/langgraph";
import { StateGraph } from "@langchain/langgraph";

const builder = new StateGraph(/* ... */);
// ...
const someGraph = builder.compile();

const someWorkflow = entrypoint(
  { name: "someWorkflow" },
  async (someInput: Record<string, any>) => {
    // Call a graph defined using the graph API
    const result1 = await someGraph.invoke(/* ... */);
    // Call another graph defined using the graph API
    const result2 = await anotherGraph.invoke(/* ... */);
    return {
      result1,
      result2,
    };
  }
);
```

:::

> [accordion] Extended example: calling a simple graph from the functional API
>
> :::python
>
> ```python theme={null}
> from langchain_core.utils.uuid import uuid7
> from typing import TypedDict
> from langgraph.func import entrypoint
> from langgraph.checkpoint.memory import InMemorySaver
> from langgraph.graph import StateGraph
>
> # Define the shared state type
> class State(TypedDict):
>     foo: int
>
> # Define a simple transformation node
> def double(state: State) -> State:
>     return {"foo": state["foo"] * 2}
>
> # Build the graph using the Graph API
> builder = StateGraph(State)
> builder.add_node("double", double)
> builder.set_entry_point("double")
> graph = builder.compile()
>
> # Define the functional API workflow
> checkpointer = InMemorySaver()
>
> @entrypoint(checkpointer=checkpointer)
> def workflow(x: int) -> dict:
>     result = graph.invoke({"foo": x})
>     return {"bar": result["foo"]}
>
> # Execute the workflow
> config = {"configurable": {"thread_id": str(uuid7())}}
> print(workflow.invoke(5, config=config))  # Output: {'bar': 10}
> ```
>
> :::
>
> :::js
>
> ```typescript theme={null}
> import { v7 as uuid7 } from "uuid";
> import { entrypoint, MemorySaver, StateGraph, StateSchema } from "@langchain/langgraph";
> import * as z from "zod";
>
> // Define the shared state type
> const State = new StateSchema({
>   foo: z.number(),
> });
>
> // Build the graph using the Graph API
> const builder = new StateGraph(State)
>   .addNode("double", (state) => {
>     return { foo: state.foo * 2 };
>   })
>   .addEdge("__start__", "double");
> const graph = builder.compile();
>
> // Define the functional API workflow
> const checkpointer = new MemorySaver();
>
> const workflow = entrypoint(
>   { checkpointer, name: "workflow" },
>   async (x: number) => {
>     const result = await graph.invoke({ foo: x });
>     return { bar: result.foo };
>   }
> );
>
> // Execute the workflow
> const config = { configurable: { thread_id: uuid7() } };
> console.log(await workflow.invoke(5, config)); // Output: { bar: 10 }
> ```
>
> :::

## 调用其他 entrypoints

你可以在一个 **entrypoint** 或 **task** 内部调用其他 **entrypoints**。

:::python

```python theme={null}
@entrypoint() # Will automatically use the checkpointer from the parent entrypoint
def some_other_workflow(inputs: dict) -> int:
    return inputs["value"]

@entrypoint(checkpointer=checkpointer)
def my_workflow(inputs: dict) -> int:
    value = some_other_workflow.invoke({"value": 1})
    return value
```

:::

:::js

```typescript theme={null}
// Will automatically use the checkpointer from the parent entrypoint
const someOtherWorkflow = entrypoint(
  { name: "someOtherWorkflow" },
  async (inputs: { value: number }) => {
    return inputs.value;
  }
);

const myWorkflow = entrypoint(
  { checkpointer, name: "myWorkflow" },
  async (inputs: { value: number }) => {
    const value = await someOtherWorkflow.invoke({ value: 1 });
    return value;
  }
);
```

:::

> [accordion] Extended example: calling another entrypoint
>
> :::python
>
> ```python theme={null}
> from langchain_core.utils.uuid import uuid7
> from langgraph.func import entrypoint
> from langgraph.checkpoint.memory import InMemorySaver
>
> # Initialize a checkpointer
> checkpointer = InMemorySaver()
>
> # A reusable sub-workflow that multiplies a number
> @entrypoint()
> def multiply(inputs: dict) -> int:
>     return inputs["a"] * inputs["b"]
>
> # Main workflow that invokes the sub-workflow
> @entrypoint(checkpointer=checkpointer)
> def main(inputs: dict) -> dict:
>     result = multiply.invoke({"a": inputs["x"], "b": inputs["y"]})
>     return {"product": result}
>
> # Execute the main workflow
> config = {"configurable": {"thread_id": str(uuid7())}}
> print(main.invoke({"x": 6, "y": 7}, config=config))  # Output: {'product': 42}
> ```
>
> :::
>
> :::js
>
> ```typescript theme={null}
> import { v7 as uuid7 } from "uuid";
> import { entrypoint, MemorySaver } from "@langchain/langgraph";
>
> // Initialize a checkpointer
> const checkpointer = new MemorySaver();
>
> // A reusable sub-workflow that multiplies a number
> const multiply = entrypoint(
>   { name: "multiply" },
>   async (inputs: { a: number; b: number }) => {
>     return inputs.a * inputs.b;
>   }
> );
>
> // Main workflow that invokes the sub-workflow
> const main = entrypoint(
>   { checkpointer, name: "main" },
>   async (inputs: { x: number; y: number }) => {
>     const result = await multiply.invoke({ a: inputs.x, b: inputs.y });
>     return { product: result };
>   }
> );
>
> // Execute the main workflow
> const config = { configurable: { thread_id: uuid7() } };
> console.log(await main.invoke({ x: 6, y: 7 }, config)); // Output: { product: 42 }
> ```
>
> :::

## 流式输出

**功能 API** 使用与 **图 API** 相同的流式机制。请阅读 [**流式输出指南**](https://langchain-docs-zh.mintlify.site/langchain/oss/langgraph/streaming.md) 部分以获取更多详细信息。

使用流式 API 从工作流运行中流式传输值块的示例。

:::python

```python theme={null}
config = {"configurable": {"thread_id": str(uuid7())}}

stream = main.stream_events({"x": 5}, config=config, version="v3")
for mode, chunk in stream.interleave("values"):
    print(f"{mode}: {chunk}")
# values: 10
```

> [card] [View example trace](https://smith.langchain.com/public/b2de9987-99f2-40ad-bfb4-6bbf9f01ffb1/r)
>
> 为此示例打开一个公开的 LangSmith 运行。

1. 从 `langgraph.config` 导入 @\[`get_stream_writer`]。
2. 在入口点内获取流式写入器实例。
3. 在计算开始前发出自定义数据。
4. 在计算结果后发出另一个自定义消息。
5. 使用 `stream_events()` 处理流式输出。
6. 从 `interleave("values")` 迭代 `(mode, chunk)` 对。

> [!WARNING]
>
> **Python \< 3.11 的异步**
> 如果使用 Python \< 3.11 并编写异步代码，使用 @\[`get_stream_writer` 将不起作用。请直接使用 `StreamWriter` 类。有关详细信息，请参阅 [Python \< 3.11 的异步](https://langchain-docs-zh.mintlify.site/langchain/oss/langgraph/streaming.md#async)。
>
> ```python theme={null}
> from langgraph.types import StreamWriter
>
> @entrypoint(checkpointer=checkpointer)
> async def main(inputs: dict, writer: StreamWriter) -> int:  # [!code highlight]
> ...
> ```

:::

:::js

```ts theme={null}
const config = {
  configurable: { thread_id: "functional-api-stream-custom-data" },
};

const stream = await main.streamEvents({ x: 5 }, { ...config, version: "v3" });
for await (const chunk of stream.values) {
  console.log(chunk);
}
// 10
```

> [card] [View example trace](https://smith.langchain.com/public/d666205b-1ade-4880-878c-d4410e1dc4d5/r)
>
> 为此示例打开一个公开的 LangSmith 运行。

1. 在计算开始前发出自定义数据。
2. 在计算结果后发出另一个自定义消息。
3. 使用 `streamEvents()` 处理流式输出。
4. 迭代 `stream.values` 以接收值快照。
   :::

## 重试策略

:::python

```python theme={null}
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import entrypoint, task
from langgraph.types import RetryPolicy

# This variable is just used for demonstration purposes to simulate a network failure.
# It's not something you will have in your actual code.
attempts = 0

# Let's configure the RetryPolicy to retry on ValueError.
# The default RetryPolicy is optimized for retrying specific network errors.
retry_policy = RetryPolicy(retry_on=ValueError)

@task(retry_policy=retry_policy)
def get_info():
    global attempts
    attempts += 1

    if attempts < 2:
        raise ValueError('Failure')
    return "OK"

checkpointer = InMemorySaver()

@entrypoint(checkpointer=checkpointer)
def main(inputs, writer):
    return get_info().result()

config = {
    "configurable": {
        "thread_id": "1"
    }
}

main.invoke({'any_input': 'foobar'}, config=config)
```

```pycon theme={null}
'OK'
```

:::

:::js

```typescript theme={null}
import {
  MemorySaver,
  entrypoint,
  task,
  RetryPolicy,
} from "@langchain/langgraph";

// This variable is just used for demonstration purposes to simulate a network failure.
// It's not something you will have in your actual code.
let attempts = 0;

// Let's configure the RetryPolicy to retry on ValueError.
// The default RetryPolicy is optimized for retrying specific network errors.
const retryPolicy: RetryPolicy = { retryOn: (error) => error instanceof Error };

const getInfo = task(
  {
    name: "getInfo",
    retry: retryPolicy,
  },
  () => {
    attempts += 1;

    if (attempts < 2) {
      throw new Error("Failure");
    }
    return "OK";
  }
);

const checkpointer = new MemorySaver();

const main = entrypoint(
  { checkpointer, name: "main" },
  async (inputs: Record<string, any>) => {
    return await getInfo();
  }
);

const config = {
  configurable: {
    thread_id: "1",
  },
};

await main.invoke({ any_input: "foobar" }, config);
```

```
'OK'
```

:::

:::python

## 设置任务和入口点超时

使用 `timeout` 参数与 `@task` 或 `@entrypoint` 限制单个异步尝试可以运行多长时间。以秒为单位提供超时时间或作为 `datetime.timedelta`。

```python theme={null}
import asyncio

from langgraph.errors import NodeTimeoutError
from langgraph.func import entrypoint, task
from langgraph.types import RetryPolicy


@task(
    timeout=1.0,
    retry_policy=RetryPolicy(retry_on=NodeTimeoutError),
)
async def call_api(url: str) -> str:
    await asyncio.sleep(2)
    return f"result from {url}"


@entrypoint(timeout=5.0)
async def workflow(inputs: dict) -> str:
    return await call_api(inputs["url"])


try:
    await workflow.ainvoke({"url": "https://example.com"})
except NodeTimeoutError:
    print("Task timed out")
```

超时仅支持异步任务和入口点。如果你在同步函数上设置 `timeout`，当任务或入口点被声明时，LangGraph 会引发错误。

当任务或入口点超出其超时时，LangGraph 会引发 `NodeTimeoutError`，它继承自 Python 的内置 `TimeoutError`。如果重试策略重试 `TimeoutError` 或 `NodeTimeoutError`，超时的尝试会被重试。超时适用于每次尝试独立计算，因此计时器会在每次重试时重置。

:::

## 缓存任务

:::python

```python theme={null}
import time
from langgraph.cache.memory import InMemoryCache
from langgraph.func import entrypoint, task
from langgraph.types import CachePolicy


@task(cache_policy=CachePolicy(ttl=120))    # [!code highlight]
def slow_add(x: int) -> int:
    time.sleep(1)
    return x * 2


@entrypoint(cache=InMemoryCache())
def main(inputs: dict) -> dict[str, int]:
    result1 = slow_add(inputs["x"]).result()
    result2 = slow_add(inputs["x"]).result()
    return {"result1": result1, "result2": result2}


stream = main.stream_events({"x": 5}, version="v3")
for snapshot in stream.values:
    print(snapshot)
```

1. `ttl` 以秒为单位指定。缓存将在该时间后失效。
   :::

:::js

```typescript theme={null}
import {
  InMemoryCache,
  entrypoint,
  task,
  CachePolicy,
} from "@langchain/langgraph";

const slowAdd = task(
  {
    name: "slowAdd",
    cache: { ttl: 120 },   // [!code highlight]
  },
  async (x: number) => {
    await new Promise((resolve) => setTimeout(resolve, 1000));
    return x * 2;
  }
);

const main = entrypoint(
  { cache: new InMemoryCache(), name: "main" },
  async (inputs: { x: number }) => {
    const result1 = await slowAdd(inputs.x);
    const result2 = await slowAdd(inputs.x);
    return { result1, result2 };
  }
);

const stream = await main.streamEvents({ x: 5 }, { version: "v3" });
for await (const snapshot of stream.values) {
  console.log(snapshot);
}
```

1. `ttl` 以秒为单位指定。缓存将在该时间后失效。
   :::

## 出错后恢复

:::python

```python theme={null}
import time
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import entrypoint, task
from langgraph.types import StreamWriter

# This variable is just used for demonstration purposes to simulate a network failure.
# It's not something you will have in your actual code.
attempts = 0

@task()
def get_info():
    """
    Simulates a task that fails once before succeeding.
    Raises an exception on the first attempt, then returns "OK" on subsequent tries.
    """
    global attempts
    attempts += 1

    if attempts < 2:
        raise ValueError("Failure")  # Simulate a failure on the first attempt
    return "OK"

# Initialize an in-memory checkpointer for persistence
checkpointer = InMemorySaver()

@task
def slow_task():
    """
    Simulates a slow-running task by introducing a 1-second delay.
    """
    time.sleep(1)
    return "Ran slow task."

@entrypoint(checkpointer=checkpointer)
def main(inputs, writer: StreamWriter):
    """
    Main workflow function that runs the slow_task and get_info tasks sequentially.

    Parameters:
    - inputs: Dictionary containing workflow input values.
    - writer: StreamWriter for streaming custom data.

    The workflow first executes `slow_task` and then attempts to execute `get_info`,
    which will fail on the first invocation.
    """
    slow_task_result = slow_task().result()  # Blocking call to slow_task
    get_info().result()  # Exception will be raised here on the first attempt
    return slow_task_result

# Workflow execution configuration with a unique thread identifier
config = {
    "configurable": {
        "thread_id": "1"  # Unique identifier to track workflow execution
    }
}

# This invocation will take ~1 second due to the slow_task execution
try:
    # First invocation will raise an exception due to the `get_info` task failing
    main.invoke({'any_input': 'foobar'}, config=config)
except ValueError:
    pass  # Handle the failure gracefully
```

在恢复执行时，我们不需要重新运行 `slow_task`，因为其结果已经保存在检查点中。

```python theme={null}
main.invoke(None, config=config)
```

```pycon theme={null}
'Ran slow task.'
```

:::

:::js

```typescript theme={null}
import { entrypoint, task, MemorySaver } from "@langchain/langgraph";

// This variable is just used for demonstration purposes to simulate a network failure.
// It's not something you will have in your actual code.
let attempts = 0;

const getInfo = task("getInfo", async () => {
  /**
   * Simulates a task that fails once before succeeding.
   * Throws an exception on the first attempt, then returns "OK" on subsequent tries.
   */
  attempts += 1;

  if (attempts < 2) {
    throw new Error("Failure"); // Simulate a failure on the first attempt
  }
  return "OK";
});

// Initialize an in-memory checkpointer for persistence
const checkpointer = new MemorySaver();

const slowTask = task("slowTask", async () => {
  /**
   * Simulates a slow-running task by introducing a 1-second delay.
   */
  await new Promise((resolve) => setTimeout(resolve, 1000));
  return "Ran slow task.";
});

const main = entrypoint(
  { checkpointer, name: "main" },
  async (inputs: Record<string, any>) => {
    /**
     * Main workflow function that runs the slowTask and getInfo tasks sequentially.
     *
     * Parameters:
     * - inputs: Record<string, any> containing workflow input values.
     *
     * The workflow first executes `slowTask` and then attempts to execute `getInfo`,
     * which will fail on the first invocation.
     */
    const slowTaskResult = await slowTask(); // Blocking call to slowTask
    await getInfo(); // Exception will be raised here on the first attempt
    return slowTaskResult;
  }
);

// Workflow execution configuration with a unique thread identifier
const config = {
  configurable: {
    thread_id: "1", // Unique identifier to track workflow execution
  },
};

// This invocation will take ~1 second due to the slowTask execution
try {
  // First invocation will raise an exception due to the `getInfo` task failing
  await main.invoke({ any_input: "foobar" }, config);
} catch (err) {
  // Handle the failure gracefully
}
```

在恢复执行时，我们不需要重新运行 `slowTask`，因为其结果已经保存在检查点中。

```typescript theme={null}
await main.invoke(null, config);
```

```
'Ran slow task.'
```

:::

## 人机协作

功能 API 支持 [人机协作](https://langchain-docs-zh.mintlify.site/langchain/oss/langgraph/interrupts.md) 工作流，使用 @\[`interrupt`] 函数和 `Command` 原语。

### 基本人机协作工作流

我们将创建三个 [任务](https://langchain-docs-zh.mintlify.site/langchain/oss/langgraph/functional-api.md#任务)：

1. 添加 `"bar"`。
2. 暂停等待人工输入。恢复时，添加人工输入。
3. 添加 `"qux"`。

:::python

```python theme={null}
from langgraph.func import entrypoint, task
from langgraph.types import Command, interrupt


@task
def step_1(input_query):
    """Append bar."""
    return f"{input_query} bar"


@task
def human_feedback(input_query):
    """Append user input."""
    feedback = interrupt(f"Please provide feedback: {input_query}")
    return f"{input_query} {feedback}"


@task
def step_3(input_query):
    """Append qux."""
    return f"{input_query} qux"
```

:::

:::js

```typescript theme={null}
import { entrypoint, task, interrupt, Command } from "@langchain/langgraph";

const step1 = task("step1", async (inputQuery: string) => {
  // Append bar
  return `${inputQuery} bar`;
});

const humanFeedback = task("humanFeedback", async (inputQuery: string) => {
  // Append user input
  const feedback = interrupt(`Please provide feedback: ${inputQuery}`);
  return `${inputQuery} ${feedback}`;
});

const step3 = task("step3", async (inputQuery: string) => {
  // Append qux
  return `${inputQuery} qux`;
});
```

:::

现在我们可以将这些任务组合成一个 [入口点](https://langchain-docs-zh.mintlify.site/langchain/oss/langgraph/functional-api.md#入口点)：

:::python

```python theme={null}
from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()


@entrypoint(checkpointer=checkpointer)
def graph(input_query):
    result_1 = step_1(input_query).result()
    result_2 = human_feedback(result_1).result()
    result_3 = step_3(result_2).result()

    return result_3
```

:::

:::js

```typescript theme={null}
import { MemorySaver } from "@langchain/langgraph";

const checkpointer = new MemorySaver();

const graph = entrypoint(
  { checkpointer, name: "graph" },
  async (inputQuery: string) => {
    const result1 = await step1(inputQuery);
    const result2 = await humanFeedback(result1);
    const result3 = await step3(result2);

    return result3;
  }
);
```

:::

[interrupt()](https://langchain-docs-zh.mintlify.site/langchain/oss/langgraph/interrupts.md#使用-interrupt-暂停) 在任务内部被调用，允许人工审查和编辑前一个任务的输出。先前任务的结果——在这种情况下是 `step_1`——被持久化，以便在 @\[`interrupt` 后不再重新运行。

让我们发送一个查询字符串：

:::python

```python theme={null}
config = {"configurable": {"thread_id": "1"}}

stream = graph.stream_events("foo", config, version="v3")
for message in stream.messages:
    for token in message.text:
        print(token, end="", flush=True)
```

:::

:::js

```typescript theme={null}
const config = { configurable: { thread_id: "1" } };

const stream = await graph.streamEvents("foo", config, { version: "v3" });
for await (const message of stream.messages) {
  for await (const token of message.text) {
    process.stdout.write(token);
  }
}
```

:::

请注意，我们在 `step_1` 后使用 @\[`interrupt` 暂停。中断提供了恢复运行的指令。要恢复，我们发送一个 [`Command`](https://langchain-docs-zh.mintlify.site/langchain/oss/langgraph/interrupts#恢复中断)，其中包含 `human_feedback` 任务期望的数据。

:::python

```python theme={null}
# Continue execution
stream = graph.stream_events(Command(resume="baz"), config, version="v3")
for message in stream.messages:
    for token in message.text:
        print(token, end="", flush=True)
```

:::

:::js

```typescript theme={null}
// Continue execution
const stream = await graph.streamEvents(
  new Command({ resume: "baz" }),
  config,
  { version: "v3" }
);
for await (const message of stream.messages) {
  for await (const token of message.text) {
    process.stdout.write(token);
  }
}
```

:::

恢复后，运行通过剩余步骤并按预期终止。

### 审查工具调用

要在执行前审查工具调用，我们添加一个 `review_tool_call` 函数，该函数调用 [`interrupt`](https://langchain-docs-zh.mintlify.site/langchain/oss/langgraph/interrupts#使用-interrupt-暂停)。当此函数被调用时，执行将被暂停，直到我们发出命令恢复它。

给定一个工具调用，我们的函数将 @\[`interrupt` 以供人工审查。在这种情况下，我们可以：

* 接受工具调用
* 修订工具调用并继续
* 生成自定义工具消息（例如，指示模型重新格式化其工具调用）

:::python

```python theme={null}
from typing import Union

def review_tool_call(tool_call: ToolCall) -> Union[ToolCall, ToolMessage]:
    """Review a tool call, returning a validated version."""
    human_review = interrupt(
        {
            "question": "Is this correct?",
            "tool_call": tool_call,
        }
    )
    review_action = human_review["action"]
    review_data = human_review.get("data")
    if review_action == "continue":
        return tool_call
    elif review_action == "update":
        updated_tool_call = {**tool_call, **{"args": review_data}}
        return updated_tool_call
    elif review_action == "feedback":
        return ToolMessage(
            content=review_data, name=tool_call["name"], tool_call_id=tool_call["id"]
        )
```

:::

:::js

```typescript theme={null}
import { ToolCall } from "@langchain/core/messages/tool";
import { ToolMessage } from "@langchain/core/messages";

function reviewToolCall(toolCall: ToolCall): ToolCall | ToolMessage {
  // Review a tool call, returning a validated version
  const humanReview = interrupt({
    question: "Is this correct?",
    tool_call: toolCall,
  });

  const reviewAction = humanReview.action;
  const reviewData = humanReview.data;

  if (reviewAction === "continue") {
    return toolCall;
  } else if (reviewAction === "update") {
    const updatedToolCall = { ...toolCall, args: reviewData };
    return updatedToolCall;
  } else if (reviewAction === "feedback") {
    return new ToolMessage({
      content: reviewData,
      name: toolCall.name,
      tool_call_id: toolCall.id,
    });
  }

  throw new Error(`Unknown review action: ${reviewAction}`);
}
```

:::

现在我们可以更新我们的 [入口点](https://langchain-docs-zh.mintlify.site/langchain/oss/langgraph/functional-api.md#入口点) 来审查生成的工具调用。如果工具调用被接受或修订，我们按之前的方式执行。否则，我们只需添加人工提供的 @\[`ToolMessage`。先前任务的结果——在这种情况下是初始模型调用——被持久化，以便在 @\[`interrupt` 后不再重新运行。

:::python

```python theme={null}
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph.message import add_messages
from langgraph.types import Command, interrupt


checkpointer = InMemorySaver()


@entrypoint(checkpointer=checkpointer)
def agent(messages, previous):
    if previous is not None:
        messages = add_messages(previous, messages)

    model_response = call_model(messages).result()
    while True:
        if not model_response.tool_calls:
            break

        # Review tool calls
        tool_results = []
        tool_calls = []
        for i, tool_call in enumerate(model_response.tool_calls):
            review = review_tool_call(tool_call)
            if isinstance(review, ToolMessage):
                tool_results.append(review)
            else:  # is a validated tool call
                tool_calls.append(review)
                if review != tool_call:
                    model_response.tool_calls[i] = review  # update message

        # Execute remaining tool calls
        tool_result_futures = [call_tool(tool_call) for tool_call in tool_calls]
        remaining_tool_results = [fut.result() for fut in tool_result_futures]

        # Append to message list
        messages = add_messages(
            messages,
            [model_response, *tool_results, *remaining_tool_results],
        )

        # Call model again
        model_response = call_model(messages).result()

    # Generate final response
    messages = add_messages(messages, model_response)
    return entrypoint.final(value=model_response, save=messages)
```

:::

:::js

```typescript theme={null}
import {
  MemorySaver,
  entrypoint,
  interrupt,
  Command,
  addMessages,
} from "@langchain/langgraph";
import { ToolMessage, AIMessage, BaseMessage } from "@langchain/core/messages";

const checkpointer = new MemorySaver();

const agent = entrypoint(
  { checkpointer, name: "agent" },
  async (
    messages: BaseMessage[],
    previous?: BaseMessage[]
  ): Promise<BaseMessage> => {
    if (previous !== undefined) {
      messages = addMessages(previous, messages);
    }

    let modelResponse = await callModel(messages);
    while (true) {
      if (!modelResponse.tool_calls?.length) {
        break;
      }

      // Review tool calls
      const toolResults: ToolMessage[] = [];
      const toolCalls: ToolCall[] = [];

      for (let i = 0; i < modelResponse.tool_calls.length; i++) {
        const review = reviewToolCall(modelResponse.tool_calls[i]);
        if (review instanceof ToolMessage) {
          toolResults.push(review);
        } else {
          // is a validated tool call
          toolCalls.push(review);
          if (review !== modelResponse.tool_calls[i]) {
            modelResponse.tool_calls[i] = review; // update message
          }
        }
      }

      // Execute remaining tool calls
      const remainingToolResults = await Promise.all(
        toolCalls.map((toolCall) => callTool(toolCall))
      );

      // Append to message list
      messages = addMessages(messages, [
        modelResponse,
        ...toolResults,
        ...remainingToolResults,
      ]);

      // Call model again
      modelResponse = await callModel(messages);
    }

    // Generate final response
    messages = addMessages(messages, modelResponse);
    return entrypoint.final({ value: modelResponse, save: messages });
  }
);
```

:::

## 短期记忆

短期记忆允许在不同的 **线程 ID** 的相同 **调用** 之间存储信息。参见 [短期记忆](https://langchain-docs-zh.mintlify.site/langchain/oss/langgraph/functional-api.md#短期记忆) 获取更多详细信息。

### 管理检查点

您可以查看和删除检查点存储的信息。

#### 查看线程状态

:::python

```python theme={null}
config = {
    "configurable": {
        "thread_id": "1",  # [!code highlight]
        # optionally provide an ID for a specific checkpoint,
        # otherwise the latest checkpoint is shown
        # "checkpoint_id": "1f029ca3-1f5b-6704-8004-820c16b69a5a"  # [!code highlight]

    }
}
graph.get_state(config)  # [!code highlight]
```

```
StateSnapshot(
    values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today?), HumanMessage(content="what's my name?"), AIMessage(content='Your name is Bob.')]}, next=(),
    config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1f5b-6704-8004-820c16b69a5a'}},
    metadata={
        'source': 'loop',
        'writes': {'call_model': {'messages': AIMessage(content='Your name is Bob.')}},
        'step': 4,
        'parents': {},
        'thread_id': '1'
    },
    created_at='2025-05-05T16:01:24.680462+00:00',
    parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}},
    tasks=(),
    interrupts=()
)
```

:::

:::js

```typescript theme={null}
const config = {
  configurable: {
    thread_id: "1",  // [!code highlight]
    // optionally provide an ID for a specific checkpoint,
    // otherwise the latest checkpoint is shown
    // checkpoint_id: "1f029ca3-1f5b-6704-8004-820c16b69a5a" [!code highlight]
  },
};
await graph.getState(config);  // [!code highlight]
```

```
StateSnapshot {
  values: {
    messages: [
      HumanMessage { content: "hi! I'm bob" },
      AIMessage { content: "Hi Bob! How are you doing today?" },
      HumanMessage { content: "what's my name?" },
      AIMessage { content: "Your name is Bob." }
    ]
  },
  next: [],
  config: { configurable: { thread_id: '1', checkpoint_ns: '', checkpoint_id: '1f029ca3-1f5b-6704-8004-820c16b69a5a' } },
  metadata: {
    source: 'loop',
    writes: { call_model: { messages: AIMessage { content: "Your name is Bob." } } },
    step: 4,
    parents: {},
    thread_id: '1'
  },
  createdAt: '2025-05-05T16:01:24.680462+00:00',
  parentConfig: { configurable: { thread_id: '1', checkpoint_ns: '', checkpoint_id: '1f029ca3-1790-6b0a-8003-baf965b6a38f' } },
  tasks: [],
  interrupts: []
}
```

:::

#### 查看线程历史

:::python

```python theme={null}
config = {
    "configurable": {
        "thread_id": "1"  # [!code highlight]
    }
}
list(graph.get_state_history(config))  # [!code highlight]
```

```
[
    StateSnapshot(
        values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content="what's my name?"), AIMessage(content='Your name is Bob.')]},
        next=(),
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1f5b-6704-8004-820c16b69a5a'}},
        metadata={'source': 'loop', 'writes': {'call_model': {'messages': AIMessage(content='Your name is Bob.')}}, 'step': 4, 'parents': {}, 'thread_id': '1'},
        created_at='2025-05-05T16:01:24.680462+00:00',
        parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}},
        tasks=(),
        interrupts=()
    ),
    StateSnapshot(
        values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content="what's my name?")]},
        next=('call_model',),
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}},
        metadata={'source': 'loop', 'writes': None, 'step': 3, 'parents': {}, 'thread_id': '1'},
        created_at='2025-05-05T16:01:23.863421+00:00',
        parent_config={...}
        tasks=(PregelTask(id='8ab4155e-6b15-b885-9ce5-bed69a2c305c', name='call_model', path=('__pregel_pull', 'call_model'), error=None, interrupts=(), state=None, result={'messages': AIMessage(content='Your name is Bob.')}),),
        interrupts=()
    ),
    StateSnapshot(
        values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]},
        next=('__start__',),
        config={...},
        metadata={'source': 'input', 'writes': {'__start__': {'messages': [{'role': 'user', 'content': "what's my name?"}]}}, 'step': 2, 'parents': {}, 'thread_id': '1'},
        created_at='2025-05-05T16:01:23.863173+00:00',
        parent_config={...}
        tasks=(PregelTask(id='24ba39d6-6db1-4c9b-f4c5-682aeaf38dcd', name='__start__', path=('__pregel_pull', '__start__'), error=None, interrupts=(), state=None, result={'messages': [{'role': 'user', 'content': "what's my name?"}]}),),
        interrupts=()
    ),
    StateSnapshot(
        values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]},
        next=(),
        config={...},
        metadata={'source': 'loop', 'writes': {'call_model': {'messages': AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')}}, 'step': 1, 'parents': {}, 'thread_id': '1'},
        created_at='2025-05-05T16:01:23.862295+00:00',
        parent_config={...}
        tasks=(),
        interrupts=()
    ),
    StateSnapshot(
        values={'messages': [HumanMessage(content="hi! I'm bob")]},
        next=('call_model',),
        config={...},
        metadata={'source': 'loop', 'writes': None, 'step': 0, 'parents': {}, 'thread_id': '1'},
        created_at='2025-05-05T16:01:22.278960+00:00',
        parent_config={...}
        tasks=(PregelTask(id='8cbd75e0-3720-b056-04f7-71ac805140a0', name='call_model', path=('__pregel_pull', 'call_model'), error=None, interrupts=(), state=None, result={'messages': AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')}),),
        interrupts=()
    ),
    StateSnapshot(
        values={'messages': []},
        next=('__start__',),
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-0870-6ce2-bfff-1f3f14c3e565'}},
        metadata={'source': 'input', 'writes': {'__start__': {'messages': [{'role': 'user', 'content': "hi! I'm bob"}]}}, 'step': -1, 'parents': {}, 'thread_id': '1'},
        created_at='2025-05-05T16:01:22.277497+00:00',
        parent_config=None,
        tasks=(PregelTask(id='d458367b-8265-812c-18e2-33001d199ce6', name='__start__', path=('__pregel_pull', '__start__'), error=None, interrupts=(), state=None, result={'messages': [{'role': 'user', 'content': "hi! I'm bob"}]}),),
        interrupts=()
    )
]
```

:::

:::js

```typescript theme={null}
const config = {
  configurable: {
    thread_id: "1",  // [!code highlight]
  },
};
const history = [];  // [!code highlight]
for await (const state of graph.getStateHistory(config)) {
  history.push(state);
}
```

```
[
  StateSnapshot {
    values: {
      messages: [
        HumanMessage { content: "hi! I'm bob" },
        AIMessage { content: "Hi Bob! How are you doing today? Is there anything I can help you with?" },
        HumanMessage { content: "what's my name?" },
        AIMessage { content: "Your name is Bob." }
      ]
    },
    next: [],
    config: { configurable: { thread_id: '1', checkpoint_ns: '', checkpoint_id: '1f029ca3-1f5b-6704-8004-820c16b69a5a' } },
    metadata: { source: 'loop', writes: { call_model: { messages: AIMessage { content: "Your name is Bob." } } }, step: 4, parents: {}, thread_id: '1' },
    createdAt: '2025-05-05T16:01:24.680462+00:00',
    parentConfig: { configurable: { thread_id: '1', checkpoint_ns: '', checkpoint_id: '1f029ca3-1790-6b0a-8003-baf965b6a38f' } },
    tasks: [],
    interrupts: []
  },
  // ... more state snapshots
]
```

:::

### 将返回值与保存值解耦

使用 `entrypoint.final` 将返回给调用者的值与在检查点中持久化的值解耦。这在以下情况下很有用：

* 您想返回一个计算结果（例如，摘要或状态），但保存一个不同的内部值供下次调用使用。
* 您需要控制传递给下次运行的前一个参数的内容。

:::python

```python theme={null}
from langgraph.func import entrypoint
from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()

@entrypoint(checkpointer=checkpointer)
def accumulate(n: int, *, previous: int | None) -> entrypoint.final[int, int]:
    previous = previous or 0
    total = previous + n
    # Return the *previous* value to the caller but save the *new* total to the checkpoint.
    return entrypoint.final(value=previous, save=total)

config = {"configurable": {"thread_id": "my-thread"}}

print(accumulate.invoke(1, config=config))  # 0
print(accumulate.invoke(2, config=config))  # 1
print(accumulate.invoke(3, config=config))  # 3
```

:::

:::js

```typescript theme={null}
import { entrypoint, MemorySaver } from "@langchain/langgraph";

const checkpointer = new MemorySaver();

const accumulate = entrypoint(
  { checkpointer, name: "accumulate" },
  async (n: number, previous?: number) => {
    const prev = previous || 0;
    const total = prev + n;
    // Return the *previous* value to the caller but save the *new* total to the checkpoint.
    return entrypoint.final({ value: prev, save: total });
  }
);

const config = { configurable: { thread_id: "my-thread" } };

console.log(await accumulate.invoke(1, config)); // 0
console.log(await accumulate.invoke(2, config)); // 1
console.log(await accumulate.invoke(3, config)); // 3
```

:::

### 对话模型示例

使用功能 API 和 @\[`InMemorySaver`] 检查点的简单对话模型示例。

该模型能够记住之前的对话并从中断处继续。

:::python

```python theme={null}
from langchain_core.messages import BaseMessage
from langgraph.graph import add_messages
from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import InMemorySaver
from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(model="claude-sonnet-4-6")

@task
def call_model(messages: list[BaseMessage]):
    response = model.invoke(messages)
    return response

checkpointer = InMemorySaver()

@entrypoint(checkpointer=checkpointer)
def workflow(inputs: list[BaseMessage], *, previous: list[BaseMessage]):
    if previous:
        inputs = add_messages(previous, inputs)

    response = call_model(inputs).result()
    return entrypoint.final(value=response, save=add_messages(inputs, response))

config = {"configurable": {"thread_id": "1"}}
input_message = {"role": "user", "content": "hi! I'm bob"}
stream = workflow.stream_events([input_message], config, version="v3")
for snapshot in stream.values:
    print(snapshot)

input_message = {"role": "user", "content": "what's my name?"}
stream = workflow.stream_events([input_message], config, version="v3")
for snapshot in stream.values:
    print(snapshot)
```

:::

:::js

```typescript theme={null}
import { BaseMessage } from "@langchain/core/messages";
import {
  addMessages,
  entrypoint,
  task,
  MemorySaver,
} from "@langchain/langgraph";
import { ChatAnthropic } from "@langchain/anthropic";

const model = new ChatAnthropic({ model: "claude-sonnet-4-6" });

const callModel = task(
  "callModel",
  async (messages: BaseMessage[]): Promise<BaseMessage> => {
    const response = await model.invoke(messages);
    return response;
  }
);

const checkpointer = new MemorySaver();

const workflow = entrypoint(
  { checkpointer, name: "workflow" },
  async (
    inputs: BaseMessage[],
    previous?: BaseMessage[]
  ): Promise<BaseMessage> => {
    let messages = inputs;
    if (previous) {
      messages = addMessages(previous, inputs);
    }

    const response = await callModel(messages);
    return entrypoint.final({
      value: response,
      save: addMessages(messages, response),
    });
  }
);

const config = { configurable: { thread_id: "1" } };
const inputMessage = { role: "user", content: "hi! I'm bob" };

const stream1 = await workflow.streamEvents([inputMessage], { ...config, version: "v3" });
for await (const snapshot of stream1.values) {
  console.log(snapshot);
}

const inputMessage2 = { role: "user", content: "what's my name?" };
const stream2 = await workflow.streamEvents([inputMessage2], { ...config, version: "v3" });
for await (const snapshot of stream2.values) {
  console.log(snapshot);
}
```

:::

## 长期记忆

[长期记忆](https://langchain-docs-zh.mintlify.site/langchain/oss/concepts/memory.md#长期记忆)允许跨不同的 **thread ids** 存储信息。这可以用于在一个对话中学习有关特定用户的信息，并在另一个对话中使用它。

## 工作流

* [查看关于工作流和智能体](https://langchain-docs-zh.mintlify.site/langchain/oss/langgraph/workflows-agents.md)的指南，了解更多使用 Functional API 构建工作流的示例。

## 与其他库集成

* [使用函数式 API 为其他框架添加 LangGraph 的功能](https://docs.langchain.com/langsmith/deploy-other-frameworks)：将 LangGraph 的功能（如持久化、记忆和流式输出）添加到那些不提供这些功能的其他智能体框架中。]

***

> 官方原文：[src/oss/langgraph/use-functional-api.mdx](https://github.com/langchain-ai/docs/blob/main/src/oss/langgraph/use-functional-api.mdx) · 非官方翻译，译文由 AI 生成，以英文原文为准。


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.