mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-09-10 03:37:51 +02:00
feat: add docs translations (#5552)
Co-authored-by: Eugene Yurtsev <eyurtsev@gmail.com> Co-authored-by: Tat Dat Duong <david@duong.cz>
This commit is contained in:
Hunter Lovell
co-authored by
Eugene Yurtsev
Tat Dat Duong
Eugene Yurtsev
Tat Dat Duong
parent
72e418e4d0
commit
d59091672f
@@ -9,12 +9,21 @@ search:
|
||||
|
||||
The **Functional API** allows you to add LangGraph's key features — [persistence](./persistence.md), [memory](../how-tos/memory/add-memory.md), [human-in-the-loop](./human_in_the_loop.md), and [streaming](./streaming.md) — to your applications with minimal changes to your existing code.
|
||||
|
||||
It is designed to integrate these features into existing code that may use standard language primitives for branching and control flow, such as `if` statements, `for` loops, and function calls. Unlike many data orchestration frameworks that require restructuring code into an explicit pipeline or DAG, the Functional API allows you to incorporate these capabilities without enforcing a rigid execution model.
|
||||
It is designed to integrate these features into existing code that may use standard language primitives for branching and control flow, such as `if` statements, `for` loops, and function calls. Unlike many data orchestration frameworks that require restructuring code into an explicit pipeline or DAG, the Functional API allows you to incorporate these capabilities without enforcing a rigid execution model.
|
||||
|
||||
The Functional API uses two key building blocks:
|
||||
The Functional API uses two key building blocks:
|
||||
|
||||
- **`@entrypoint`** – Marks a function as the starting point of a workflow, encapsulating logic and managing execution flow, including handling long-running tasks and interrupts.
|
||||
:::python
|
||||
|
||||
- **`@entrypoint`** – Marks a function as the starting point of a workflow, encapsulating logic and managing execution flow, including handling long-running tasks and interrupts.
|
||||
- **`@task`** – Represents a discrete unit of work, such as an API call or data processing step, that can be executed asynchronously within an entrypoint. Tasks return a future-like object that can be awaited or resolved synchronously.
|
||||
:::
|
||||
|
||||
:::js
|
||||
|
||||
- **`entrypoint`** – An entrypoint encapsulates workflow logic and manages execution flow, including handling long-running tasks and interrupts.
|
||||
- **`task`** – Represents a discrete unit of work, such as an API call or data processing step, that can be executed asynchronously within an entrypoint. Tasks return a future-like object that can be awaited or resolved synchronously.
|
||||
:::
|
||||
|
||||
This provides a minimal abstraction for building workflows with state management and streaming.
|
||||
|
||||
@@ -33,17 +42,17 @@ Here are some key differences:
|
||||
- **Checkpointing**: Both APIs generate and use checkpoints. In the **Graph API** a new checkpoint is generated after every [superstep](./low_level.md). In the **Functional API**, when tasks are executed, their results are saved to an existing checkpoint associated with the given entrypoint instead of creating a new checkpoint.
|
||||
- **Visualization**: The Graph API makes it easy to visualize the workflow as a graph which can be useful for debugging, understanding the workflow, and sharing with others. The Functional API does not support visualization as the graph is dynamically generated during runtime.
|
||||
|
||||
|
||||
## Example
|
||||
|
||||
Below we demonstrate a simple application that writes an essay and [interrupts](human_in_the_loop.md) to request human review.
|
||||
|
||||
:::python
|
||||
|
||||
```python
|
||||
from langgraph.checkpoint.memory import InMemorySaver
|
||||
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."""
|
||||
@@ -70,12 +79,50 @@ def workflow(topic: str) -> dict:
|
||||
}
|
||||
```
|
||||
|
||||
:::
|
||||
|
||||
:::js
|
||||
|
||||
```typescript
|
||||
import { MemorySaver, entrypoint, task, interrupt } from "@langchain/langgraph";
|
||||
|
||||
const writeEssay = task("writeEssay", async (topic: string) => {
|
||||
// A placeholder for a long-running task.
|
||||
await new Promise((resolve) => setTimeout(resolve, 1000));
|
||||
return `An essay about topic: ${topic}`;
|
||||
});
|
||||
|
||||
const workflow = entrypoint(
|
||||
{ checkpointer: new MemorySaver(), name: "workflow" },
|
||||
async (topic: string) => {
|
||||
const essay = await writeEssay(topic);
|
||||
const isApproved = 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, // 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, // The essay that was generated
|
||||
isApproved, // Response from HIL
|
||||
};
|
||||
}
|
||||
);
|
||||
```
|
||||
|
||||
:::
|
||||
|
||||
??? example "Detailed Explanation"
|
||||
|
||||
This workflow will write an essay about the topic "cat" and then pause to get a review from a human. The workflow can be interrupted for an indefinite amount of time until a review is provided.
|
||||
|
||||
When the workflow is resumed, it executes from the very start, but because the result of the `write_essay` task was already saved, the task result will be loaded from the checkpoint instead of being recomputed.
|
||||
When the workflow is resumed, it executes from the very start, but because the result of the `writeEssay` task was already saved, the task result will be loaded from the checkpoint instead of being recomputed.
|
||||
|
||||
:::python
|
||||
```python
|
||||
import time
|
||||
import uuid
|
||||
@@ -147,18 +194,104 @@ def workflow(topic: str) -> dict:
|
||||
```
|
||||
|
||||
The workflow has been completed and the review has been added to the essay.
|
||||
:::
|
||||
|
||||
:::js
|
||||
```typescript
|
||||
import { v4 as uuidv4 } from "uuid";
|
||||
import { MemorySaver, entrypoint, task, interrupt } from "@langchain/langgraph";
|
||||
|
||||
const writeEssay = task("writeEssay", async (topic: string) => {
|
||||
// This is a placeholder for a long-running task.
|
||||
await new Promise(resolve => setTimeout(resolve, 1000));
|
||||
return `An essay about topic: ${topic}`;
|
||||
});
|
||||
|
||||
const workflow = entrypoint(
|
||||
{ checkpointer: new MemorySaver(), name: "workflow" },
|
||||
async (topic: string) => {
|
||||
const essay = await writeEssay(topic);
|
||||
const isApproved = 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, // 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, // The essay that was generated
|
||||
isApproved, // Response from HIL
|
||||
};
|
||||
}
|
||||
);
|
||||
|
||||
const threadId = uuidv4();
|
||||
|
||||
const config = {
|
||||
configurable: {
|
||||
thread_id: threadId
|
||||
}
|
||||
};
|
||||
|
||||
for await (const item of workflow.stream("cat", config)) {
|
||||
console.log(item);
|
||||
}
|
||||
```
|
||||
|
||||
```console
|
||||
{ writeEssay: 'An essay about topic: cat' }
|
||||
{
|
||||
__interrupt__: [{
|
||||
value: { essay: 'An essay about topic: cat', action: 'Please approve/reject the essay' },
|
||||
resumable: true,
|
||||
ns: ['workflow:f7b8508b-21c0-8b4c-5958-4e8de74d2684'],
|
||||
when: 'during'
|
||||
}]
|
||||
}
|
||||
```
|
||||
|
||||
An essay has been written and is ready for review. Once the review is provided, we can resume the workflow:
|
||||
|
||||
```typescript
|
||||
import { Command } from "@langchain/langgraph";
|
||||
|
||||
// Get review from a user (e.g., via a UI)
|
||||
// In this case, we're using a bool, but this can be any json-serializable value.
|
||||
const humanReview = true;
|
||||
|
||||
for await (const item of workflow.stream(new Command({ resume: humanReview }), config)) {
|
||||
console.log(item);
|
||||
}
|
||||
```
|
||||
|
||||
```console
|
||||
{ workflow: { essay: 'An essay about topic: cat', isApproved: true } }
|
||||
```
|
||||
|
||||
The workflow has been completed and the review has been added to the essay.
|
||||
:::
|
||||
|
||||
## Entrypoint
|
||||
|
||||
The [`@entrypoint`][langgraph.func.entrypoint] decorator can be used to create a workflow from a function. It encapsulates workflow logic and manages execution flow, including handling *long-running tasks* and [interrupts](./human_in_the_loop.md).
|
||||
:::python
|
||||
The @[`@entrypoint`][entrypoint] decorator can be used to create a workflow from a function. It encapsulates workflow logic and manages execution flow, including handling _long-running tasks_ and [interrupts](./human_in_the_loop.md).
|
||||
:::
|
||||
|
||||
:::js
|
||||
The @[`entrypoint`][entrypoint] function can be used to create a workflow from a function. It encapsulates workflow logic and manages execution flow, including handling _long-running tasks_ and [interrupts](./human_in_the_loop.md).
|
||||
:::
|
||||
|
||||
### Definition
|
||||
|
||||
An **entrypoint** is defined by decorating a function with the `@entrypoint` decorator.
|
||||
:::python
|
||||
An **entrypoint** is defined by decorating a function with the `@entrypoint` decorator.
|
||||
|
||||
The function **must accept a single positional argument**, which serves as the workflow input. If you need to pass multiple pieces of data, use a dictionary as the input type for the first argument.
|
||||
|
||||
Decorating a function with an `entrypoint` produces a [`Pregel`][langgraph.pregel.Pregel.stream] instance which helps to manage the execution of the workflow (e.g., handles streaming, resumption, and checkpointing).
|
||||
Decorating a function with an `entrypoint` produces a @[`Pregel`][Pregel.stream] instance which helps to manage the execution of the workflow (e.g., handles streaming, resumption, and checkpointing).
|
||||
|
||||
You will usually want to pass a **checkpointer** to the `@entrypoint` decorator to enable persistence and use features like **human-in-the-loop**.
|
||||
|
||||
@@ -185,22 +318,48 @@ You will usually want to pass a **checkpointer** to the `@entrypoint` decorator
|
||||
# some logic that may involve long-running tasks like API calls,
|
||||
# and may be interrupted for human-in-the-loop
|
||||
...
|
||||
return result
|
||||
return result
|
||||
```
|
||||
|
||||
:::
|
||||
|
||||
:::js
|
||||
An **entrypoint** is defined by calling the `entrypoint` function with configuration and a function.
|
||||
|
||||
The function **must accept a single positional argument**, which serves as the workflow input. If you need to pass multiple pieces of data, use an object as the input type for the first argument.
|
||||
|
||||
Creating an entrypoint with a function produces a workflow instance which helps to manage the execution of the workflow (e.g., handles streaming, resumption, and checkpointing).
|
||||
|
||||
You will often want to pass a **checkpointer** to the `entrypoint` function to enable persistence and use features like **human-in-the-loop**.
|
||||
|
||||
```typescript
|
||||
import { entrypoint } from "@langchain/langgraph";
|
||||
|
||||
const myWorkflow = entrypoint(
|
||||
{ checkpointer, name: "workflow" },
|
||||
async (someInput: Record<string, any>): Promise<number> => {
|
||||
// some logic that may involve long-running tasks like API calls,
|
||||
// and may be interrupted for human-in-the-loop
|
||||
return result;
|
||||
}
|
||||
);
|
||||
```
|
||||
|
||||
:::
|
||||
|
||||
!!! important "Serialization"
|
||||
|
||||
The **inputs** and **outputs** of entrypoints must be JSON-serializable to support checkpointing. Please see the [serialization](#serialization) section for more details.
|
||||
|
||||
:::python
|
||||
|
||||
### Injectable parameters
|
||||
|
||||
When declaring an `entrypoint`, you can request access to additional parameters that will be injected automatically at run time. These parameters include:
|
||||
|
||||
|
||||
| Parameter | Description |
|
||||
|--------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------|
|
||||
| **previous** | Access the state associated with the previous `checkpoint` for the given thread. See [short-term-memory](#short-term-memory). |
|
||||
| ------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
|
||||
| **previous** | Access the state associated with the previous `checkpoint` for the given thread. See [short-term-memory](#short-term-memory). |
|
||||
| **store** | An instance of [BaseStore][langgraph.store.base.BaseStore]. Useful for [long-term memory](../how-tos/use-functional-api.md#long-term-memory). |
|
||||
| **writer** | Use to access the StreamWriter when working with Async Python < 3.11. See [streaming with functional API for details](../how-tos/use-functional-api.md#streaming). |
|
||||
| **config** | For accessing run time configuration. See [RunnableConfig](https://python.langchain.com/docs/concepts/runnables/#runnableconfig) for information. |
|
||||
@@ -222,7 +381,7 @@ When declaring an `entrypoint`, you can request access to additional parameters
|
||||
@entrypoint(
|
||||
checkpointer=checkpointer, # Specify the checkpointer
|
||||
store=in_memory_store # Specify the store
|
||||
)
|
||||
)
|
||||
def my_workflow(
|
||||
some_input: dict, # The input (e.g., passed via `invoke`)
|
||||
*,
|
||||
@@ -233,9 +392,12 @@ When declaring an `entrypoint`, you can request access to additional parameters
|
||||
) -> ...:
|
||||
```
|
||||
|
||||
:::
|
||||
|
||||
### Executing
|
||||
|
||||
Using the [`@entrypoint`](#entrypoint) yields a [`Pregel`][langgraph.pregel.Pregel.stream] object that can be executed using the `invoke`, `ainvoke`, `stream`, and `astream` methods.
|
||||
:::python
|
||||
Using the [`@entrypoint`](#entrypoint) yields a @[`Pregel`][Pregel.stream] object that can be executed using the `invoke`, `ainvoke`, `stream`, and `astream` methods.
|
||||
|
||||
=== "Invoke"
|
||||
|
||||
@@ -260,7 +422,7 @@ Using the [`@entrypoint`](#entrypoint) yields a [`Pregel`][langgraph.pregel.Preg
|
||||
```
|
||||
|
||||
=== "Stream"
|
||||
|
||||
|
||||
```python
|
||||
config = {
|
||||
"configurable": {
|
||||
@@ -285,9 +447,42 @@ Using the [`@entrypoint`](#entrypoint) yields a [`Pregel`][langgraph.pregel.Preg
|
||||
print(chunk)
|
||||
```
|
||||
|
||||
:::
|
||||
|
||||
:::js
|
||||
Using the [`entrypoint`](#entrypoint) function will return an object that can be executed using the `invoke` and `stream` methods.
|
||||
|
||||
=== "Invoke"
|
||||
|
||||
```typescript
|
||||
const config = {
|
||||
configurable: {
|
||||
thread_id: "some_thread_id"
|
||||
}
|
||||
};
|
||||
await myWorkflow.invoke(someInput, config); // Wait for the result
|
||||
```
|
||||
|
||||
=== "Stream"
|
||||
|
||||
```typescript
|
||||
const config = {
|
||||
configurable: {
|
||||
thread_id: "some_thread_id"
|
||||
}
|
||||
};
|
||||
|
||||
for await (const chunk of myWorkflow.stream(someInput, config)) {
|
||||
console.log(chunk);
|
||||
}
|
||||
```
|
||||
|
||||
:::
|
||||
|
||||
### Resuming
|
||||
|
||||
Resuming an execution after an [interrupt][langgraph.types.interrupt] can be done by passing a **resume** value to the [Command][langgraph.types.Command] primitive.
|
||||
:::python
|
||||
Resuming an execution after an @[interrupt][interrupt] can be done by passing a **resume** value to the @[Command] primitive.
|
||||
|
||||
=== "Invoke"
|
||||
|
||||
@@ -299,7 +494,7 @@ Resuming an execution after an [interrupt][langgraph.types.interrupt] can be don
|
||||
"thread_id": "some_thread_id"
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
my_workflow.invoke(Command(resume=some_resume_value), config)
|
||||
```
|
||||
|
||||
@@ -313,7 +508,7 @@ Resuming an execution after an [interrupt][langgraph.types.interrupt] can be don
|
||||
"thread_id": "some_thread_id"
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
await my_workflow.ainvoke(Command(resume=some_resume_value), config)
|
||||
```
|
||||
|
||||
@@ -327,7 +522,7 @@ Resuming an execution after an [interrupt][langgraph.types.interrupt] can be don
|
||||
"thread_id": "some_thread_id"
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
for chunk in my_workflow.stream(Command(resume=some_resume_value), config):
|
||||
print(chunk)
|
||||
```
|
||||
@@ -347,8 +542,51 @@ Resuming an execution after an [interrupt][langgraph.types.interrupt] can be don
|
||||
print(chunk)
|
||||
```
|
||||
|
||||
**Resuming after an error**
|
||||
:::
|
||||
|
||||
:::js
|
||||
Resuming an execution after an @[interrupt][interrupt] can be done by passing a **resume** value to the @[`Command`][Command] primitive.
|
||||
|
||||
=== "Invoke"
|
||||
|
||||
```typescript
|
||||
import { Command } from "@langchain/langgraph";
|
||||
|
||||
const config = {
|
||||
configurable: {
|
||||
thread_id: "some_thread_id"
|
||||
}
|
||||
};
|
||||
|
||||
await myWorkflow.invoke(new Command({ resume: someResumeValue }), config);
|
||||
```
|
||||
|
||||
=== "Stream"
|
||||
|
||||
```typescript
|
||||
import { Command } from "@langchain/langgraph";
|
||||
|
||||
const config = {
|
||||
configurable: {
|
||||
thread_id: "some_thread_id"
|
||||
}
|
||||
};
|
||||
|
||||
const stream = await myWorkflow.stream(
|
||||
new Command({ resume: someResumableValue }),
|
||||
config,
|
||||
)
|
||||
|
||||
for await (const chunk of stream) {
|
||||
console.log(chunk);
|
||||
}
|
||||
```
|
||||
|
||||
:::
|
||||
|
||||
:::python
|
||||
|
||||
**Resuming after an error**
|
||||
|
||||
To resume after an error, run the `entrypoint` with a `None` and the same **thread id** (config).
|
||||
|
||||
@@ -363,7 +601,7 @@ This assumes that the underlying **error** has been resolved and execution can p
|
||||
"thread_id": "some_thread_id"
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
my_workflow.invoke(None, config)
|
||||
```
|
||||
|
||||
@@ -376,7 +614,7 @@ This assumes that the underlying **error** has been resolved and execution can p
|
||||
"thread_id": "some_thread_id"
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
await my_workflow.ainvoke(None, config)
|
||||
```
|
||||
|
||||
@@ -389,7 +627,7 @@ This assumes that the underlying **error** has been resolved and execution can p
|
||||
"thread_id": "some_thread_id"
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
for chunk in my_workflow.stream(None, config):
|
||||
print(chunk)
|
||||
```
|
||||
@@ -408,10 +646,49 @@ This assumes that the underlying **error** has been resolved and execution can p
|
||||
print(chunk)
|
||||
```
|
||||
|
||||
:::
|
||||
|
||||
:::js
|
||||
|
||||
**Resuming after an error**
|
||||
|
||||
To resume after an error, run the `entrypoint` with `null` and the same **thread id** (config).
|
||||
|
||||
This assumes that the underlying **error** has been resolved and execution can proceed successfully.
|
||||
|
||||
=== "Invoke"
|
||||
|
||||
```typescript
|
||||
const config = {
|
||||
configurable: {
|
||||
thread_id: "some_thread_id"
|
||||
}
|
||||
};
|
||||
|
||||
await myWorkflow.invoke(null, config);
|
||||
```
|
||||
|
||||
=== "Stream"
|
||||
|
||||
```typescript
|
||||
const config = {
|
||||
configurable: {
|
||||
thread_id: "some_thread_id"
|
||||
}
|
||||
};
|
||||
|
||||
for await (const chunk of myWorkflow.stream(null, config)) {
|
||||
console.log(chunk);
|
||||
}
|
||||
```
|
||||
|
||||
:::
|
||||
|
||||
### Short-term memory
|
||||
|
||||
When an `entrypoint` is defined with a `checkpointer`, it stores information between successive invocations on the same **thread id** in [checkpoints](persistence.md#checkpoints).
|
||||
When an `entrypoint` is defined with a `checkpointer`, it stores information between successive invocations on the same **thread id** in [checkpoints](persistence.md#checkpoints).
|
||||
|
||||
:::python
|
||||
This allows accessing the state from the previous invocation using the `previous` parameter.
|
||||
|
||||
By default, the `previous` parameter is the return value of the previous invocation.
|
||||
@@ -432,9 +709,40 @@ my_workflow.invoke(1, config) # 1 (previous was None)
|
||||
my_workflow.invoke(2, config) # 3 (previous was 1 from the previous invocation)
|
||||
```
|
||||
|
||||
:::
|
||||
|
||||
:::js
|
||||
This allows accessing the state from the previous invocation using the `getPreviousState` function.
|
||||
|
||||
By default, the `getPreviousState` function returns the return value of the previous invocation.
|
||||
|
||||
```typescript
|
||||
import { entrypoint, getPreviousState } from "@langchain/langgraph";
|
||||
|
||||
const myWorkflow = entrypoint(
|
||||
{ checkpointer, name: "workflow" },
|
||||
async (number: number) => {
|
||||
const previous = getPreviousState<number>() ?? 0;
|
||||
return number + previous;
|
||||
}
|
||||
);
|
||||
|
||||
const config = {
|
||||
configurable: {
|
||||
thread_id: "some_thread_id",
|
||||
},
|
||||
};
|
||||
|
||||
await myWorkflow.invoke(1, config); // 1 (previous was undefined)
|
||||
await myWorkflow.invoke(2, config); // 3 (previous was 1 from the previous invocation)
|
||||
```
|
||||
|
||||
:::
|
||||
|
||||
#### `entrypoint.final`
|
||||
|
||||
[entrypoint.final][langgraph.func.entrypoint.final] is a special primitive that can be returned from an entrypoint and allows **decoupling** the value that is **saved in the checkpoint** from the **return value of the entrypoint**.
|
||||
:::python
|
||||
@[`entrypoint.final`][entrypoint.final] is a special primitive that can be returned from an entrypoint and allows **decoupling** the value that is **saved in the checkpoint** from the **return value of the entrypoint**.
|
||||
|
||||
The first value is the return value of the entrypoint, and the second value is the value that will be saved in the checkpoint. The type annotation is `entrypoint.final[return_type, save_type]`.
|
||||
|
||||
@@ -443,7 +751,7 @@ The first value is the return value of the entrypoint, and the second value is t
|
||||
def my_workflow(number: int, *, previous: Any = None) -> entrypoint.final[int, int]:
|
||||
previous = previous or 0
|
||||
# This will return the previous value to the caller, saving
|
||||
# 2 * number to the checkpoint, which will be used in the next invocation
|
||||
# 2 * number to the checkpoint, which will be used in the next invocation
|
||||
# for the `previous` parameter.
|
||||
return entrypoint.final(value=previous, save=2 * number)
|
||||
|
||||
@@ -457,15 +765,52 @@ my_workflow.invoke(3, config) # 0 (previous was None)
|
||||
my_workflow.invoke(1, config) # 6 (previous was 3 * 2 from the previous invocation)
|
||||
```
|
||||
|
||||
:::
|
||||
|
||||
:::js
|
||||
@[`entrypoint.final`][entrypoint.final] is a special primitive that can be returned from an entrypoint and allows **decoupling** the value that is **saved in the checkpoint** from the **return value of the entrypoint**.
|
||||
|
||||
The first value is the return value of the entrypoint, and the second value is the value that will be saved in the checkpoint.
|
||||
|
||||
```typescript
|
||||
import { entrypoint, getPreviousState } from "@langchain/langgraph";
|
||||
|
||||
const myWorkflow = entrypoint(
|
||||
{ checkpointer, name: "workflow" },
|
||||
async (number: number) => {
|
||||
const previous = getPreviousState<number>() ?? 0;
|
||||
// This will return the previous value to the caller, saving
|
||||
// 2 * number to the checkpoint, which will be used in the next invocation
|
||||
// for the `previous` parameter.
|
||||
return entrypoint.final({
|
||||
value: previous,
|
||||
save: 2 * number,
|
||||
});
|
||||
}
|
||||
);
|
||||
|
||||
const config = {
|
||||
configurable: {
|
||||
thread_id: "1",
|
||||
},
|
||||
};
|
||||
|
||||
await myWorkflow.invoke(3, config); // 0 (previous was undefined)
|
||||
await myWorkflow.invoke(1, config); // 6 (previous was 3 * 2 from the previous invocation)
|
||||
```
|
||||
|
||||
:::
|
||||
|
||||
## Task
|
||||
|
||||
A **task** represents a discrete unit of work, such as an API call or data processing step. It has two key characteristics:
|
||||
|
||||
* **Asynchronous Execution**: Tasks are designed to be executed asynchronously, allowing multiple operations to run concurrently without blocking.
|
||||
* **Checkpointing**: Task results are saved to a checkpoint, enabling resumption of the workflow from the last saved state. (See [persistence](persistence.md) for more details).
|
||||
- **Asynchronous Execution**: Tasks are designed to be executed asynchronously, allowing multiple operations to run concurrently without blocking.
|
||||
- **Checkpointing**: Task results are saved to a checkpoint, enabling resumption of the workflow from the last saved state. (See [persistence](persistence.md) for more details).
|
||||
|
||||
### Definition
|
||||
|
||||
:::python
|
||||
Tasks are defined using the `@task` decorator, which wraps a regular Python function.
|
||||
|
||||
```python
|
||||
@@ -478,21 +823,37 @@ def slow_computation(input_value):
|
||||
return result
|
||||
```
|
||||
|
||||
:::
|
||||
|
||||
:::js
|
||||
Tasks are defined using the `task` function, which wraps a regular function.
|
||||
|
||||
```typescript
|
||||
import { task } from "@langchain/langgraph";
|
||||
|
||||
const slowComputation = task("slowComputation", async (inputValue: any) => {
|
||||
// Simulate a long-running operation
|
||||
return result;
|
||||
});
|
||||
```
|
||||
|
||||
:::
|
||||
|
||||
!!! important "Serialization"
|
||||
|
||||
The **outputs** of tasks must be JSON-serializable to support checkpointing.
|
||||
|
||||
### Execution
|
||||
|
||||
**Tasks** can only be called from within an **entrypoint**, another **task**, or a [state graph node](./low_level.md#nodes).
|
||||
**Tasks** can only be called from within an **entrypoint**, another **task**, or a [state graph node](./low_level.md#nodes).
|
||||
|
||||
Tasks *cannot* be called directly from the main application code.
|
||||
Tasks _cannot_ be called directly from the main application code.
|
||||
|
||||
When you call a **task**, it returns *immediately* with a future object. A future is a placeholder for a result that will be available later.
|
||||
:::python
|
||||
When you call a **task**, it returns _immediately_ with a future object. A future is a placeholder for a result that will be available later.
|
||||
|
||||
To obtain the result of a **task**, you can either wait for it synchronously (using `result()`) or await it asynchronously (using `await`).
|
||||
|
||||
|
||||
=== "Synchronous Invocation"
|
||||
|
||||
```python
|
||||
@@ -510,6 +871,22 @@ To obtain the result of a **task**, you can either wait for it synchronously (us
|
||||
return await slow_computation(some_input) # Await result asynchronously
|
||||
```
|
||||
|
||||
:::
|
||||
|
||||
:::js
|
||||
When you call a **task**, it returns a Promise that can be awaited.
|
||||
|
||||
```typescript
|
||||
const myWorkflow = entrypoint(
|
||||
{ checkpointer, name: "workflow" },
|
||||
async (someInput: number): Promise<number> => {
|
||||
return await slowComputation(someInput);
|
||||
}
|
||||
);
|
||||
```
|
||||
|
||||
:::
|
||||
|
||||
## When to use a task
|
||||
|
||||
**Tasks** are useful in the following scenarios:
|
||||
@@ -519,16 +896,21 @@ To obtain the result of a **task**, you can either wait for it synchronously (us
|
||||
- **Parallel Execution**: For I/O-bound tasks, **tasks** enable parallel execution, allowing multiple operations to run concurrently without blocking (e.g., calling multiple APIs).
|
||||
- **Observability**: Wrapping operations in **tasks** provides a way to track the progress of the workflow and monitor the execution of individual operations using [LangSmith](https://docs.smith.langchain.com/).
|
||||
- **Retryable Work**: When work needs to be retried to handle failures or inconsistencies, **tasks** provide a way to encapsulate and manage the retry logic.
|
||||
|
||||
|
||||
## Serialization
|
||||
|
||||
There are two key aspects to serialization in LangGraph:
|
||||
|
||||
1. `@entrypoint` inputs and outputs must be JSON-serializable.
|
||||
2. `@task` outputs must be JSON-serializable.
|
||||
1. `entrypoint` inputs and outputs must be JSON-serializable.
|
||||
2. `task` outputs must be JSON-serializable.
|
||||
|
||||
These requirements are necessary for enabling checkpointing and workflow resumption. Use python primitives
|
||||
like dictionaries, lists, strings, numbers, and booleans to ensure that your inputs and outputs are serializable.
|
||||
:::python
|
||||
These requirements are necessary for enabling checkpointing and workflow resumption. Use python primitives like dictionaries, lists, strings, numbers, and booleans to ensure that your inputs and outputs are serializable.
|
||||
:::
|
||||
|
||||
:::js
|
||||
These requirements are necessary for enabling checkpointing and workflow resumption. Use primitives like objects, arrays, strings, numbers, and booleans to ensure that your inputs and outputs are serializable.
|
||||
:::
|
||||
|
||||
Serialization ensures that workflow state, such as task results and intermediate values, can be reliably saved and restored. This is critical for enabling human-in-the-loop interactions, fault tolerance, and parallel execution.
|
||||
|
||||
@@ -536,9 +918,9 @@ Providing non-serializable inputs or outputs will result in a runtime error when
|
||||
|
||||
## Determinism
|
||||
|
||||
To utilize features like **human-in-the-loop**, any randomness should be encapsulated inside of **tasks**. This guarantees that when execution is halted (e.g., for human in the loop) and then resumed, it will follow the same *sequence of steps*, even if **task** results are non-deterministic.
|
||||
To utilize features like **human-in-the-loop**, any randomness should be encapsulated inside of **tasks**. This guarantees that when execution is halted (e.g., for human in the loop) and then resumed, it will follow the same _sequence of steps_, even if **task** results are non-deterministic.
|
||||
|
||||
LangGraph achieves this behavior by persisting **task** and [**subgraph**](./subgraphs.md) results as they execute. A well-designed workflow ensures that resuming execution follows the *same sequence of steps*, allowing previously computed results to be retrieved correctly without having to re-execute them. This is particularly useful for long-running **tasks** or **tasks** with non-deterministic results, as it avoids repeating previously done work and allows resuming from essentially the same.
|
||||
LangGraph achieves this behavior by persisting **task** and [**subgraph**](./subgraphs.md) results as they execute. A well-designed workflow ensures that resuming execution follows the _same sequence of steps_, allowing previously computed results to be retrieved correctly without having to re-execute them. This is particularly useful for long-running **tasks** or **tasks** with non-deterministic results, as it avoids repeating previously done work and allows resuming from essentially the same.
|
||||
|
||||
While different runs of a workflow can produce different results, resuming a **specific** run should always follow the same sequence of recorded steps. This allows LangGraph to efficiently look up **task** and **subgraph** results that were executed prior to the graph being interrupted and avoid recomputing them.
|
||||
|
||||
@@ -556,6 +938,7 @@ Encapsulate side effects (e.g., writing to a file, sending an email) in tasks to
|
||||
|
||||
In this example, a side effect (writing to a file) is directly included in the workflow, so it will be executed a second time when resuming the workflow.
|
||||
|
||||
:::python
|
||||
```python
|
||||
@entrypoint(checkpointer=checkpointer)
|
||||
def my_workflow(inputs: dict) -> int:
|
||||
@@ -568,11 +951,31 @@ Encapsulate side effects (e.g., writing to a file, sending an email) in tasks to
|
||||
value = interrupt("question")
|
||||
return value
|
||||
```
|
||||
:::
|
||||
|
||||
:::js
|
||||
```typescript
|
||||
import { entrypoint, interrupt } from "@langchain/langgraph";
|
||||
import fs from "fs";
|
||||
|
||||
const myWorkflow = entrypoint(
|
||||
{ checkpointer, name: "workflow },
|
||||
async (inputs: Record<string, any>) => {
|
||||
// This code will be executed a second time when resuming the workflow.
|
||||
// Which is likely not what you want.
|
||||
fs.writeFileSync("output.txt", "Side effect executed");
|
||||
const value = interrupt("question");
|
||||
return value;
|
||||
}
|
||||
);
|
||||
```
|
||||
:::
|
||||
|
||||
=== "Correct"
|
||||
|
||||
In this example, the side effect is encapsulated in a task, ensuring consistent execution upon resumption.
|
||||
|
||||
:::python
|
||||
```python
|
||||
from langgraph.func import task
|
||||
|
||||
@@ -590,17 +993,43 @@ Encapsulate side effects (e.g., writing to a file, sending an email) in tasks to
|
||||
value = interrupt("question")
|
||||
return value
|
||||
```
|
||||
:::
|
||||
|
||||
:::js
|
||||
```typescript
|
||||
import { entrypoint, task, interrupt } from "@langchain/langgraph";
|
||||
import * as fs from "fs";
|
||||
|
||||
const writeToFile = task("writeToFile", async () => {
|
||||
fs.writeFileSync("output.txt", "Side effect executed");
|
||||
});
|
||||
|
||||
const myWorkflow = entrypoint(
|
||||
{ checkpointer, name: "workflow" },
|
||||
async (inputs: Record<string, any>) => {
|
||||
// The side effect is now encapsulated in a task.
|
||||
await writeToFile();
|
||||
const value = interrupt("question");
|
||||
return value;
|
||||
}
|
||||
);
|
||||
```
|
||||
:::
|
||||
|
||||
### Non-deterministic control flow
|
||||
|
||||
Operations that might give different results each time (like getting current time or random numbers) should be encapsulated in tasks to ensure that on resume, the same result is returned.
|
||||
|
||||
* In a task: Get random number (5) → interrupt → resume → (returns 5 again) → ...
|
||||
* Not in a task: Get random number (5) → interrupt → resume → get new random number (7) → ...
|
||||
- In a task: Get random number (5) → interrupt → resume → (returns 5 again) → ...
|
||||
- Not in a task: Get random number (5) → interrupt → resume → get new random number (7) → ...
|
||||
|
||||
This is especially important when using **human-in-the-loop** workflows with multiple interrupts calls. LangGraph keeps a list
|
||||
of resume values for each task/entrypoint. When an interrupt is encountered, it's matched with the corresponding resume value.
|
||||
This matching is strictly **index-based**, so the order of the resume values should match the order of the interrupts.
|
||||
:::python
|
||||
This is especially important when using **human-in-the-loop** workflows with multiple interrupts calls. LangGraph keeps a list of resume values for each task/entrypoint. When an interrupt is encountered, it's matched with the corresponding resume value. This matching is strictly **index-based**, so the order of the resume values should match the order of the interrupts.
|
||||
:::
|
||||
|
||||
:::js
|
||||
This is especially important when using **human-in-the-loop** workflows with multiple interrupt calls. LangGraph keeps a list of resume values for each task/entrypoint. When an interrupt is encountered, it's matched with the corresponding resume value. This matching is strictly **index-based**, so the order of the resume values should match the order of the interrupts.
|
||||
:::
|
||||
|
||||
If order of execution is not maintained when resuming, one `interrupt` call may be matched with the wrong `resume` value, leading to incorrect results.
|
||||
|
||||
@@ -610,6 +1039,7 @@ Please read the section on [determinism](#determinism) for more details.
|
||||
|
||||
In this example, the workflow uses the current time to determine which task to execute. This is non-deterministic because the result of the workflow depends on the time at which it is executed.
|
||||
|
||||
:::python
|
||||
```python
|
||||
from langgraph.func import entrypoint
|
||||
|
||||
@@ -618,24 +1048,51 @@ Please read the section on [determinism](#determinism) for more details.
|
||||
t0 = inputs["t0"]
|
||||
# highlight-next-line
|
||||
t1 = time.time()
|
||||
|
||||
|
||||
delta_t = t1 - t0
|
||||
|
||||
|
||||
if delta_t > 1:
|
||||
result = slow_task(1).result()
|
||||
value = interrupt("question")
|
||||
else:
|
||||
result = slow_task(2).result()
|
||||
value = interrupt("question")
|
||||
|
||||
|
||||
return {
|
||||
"result": result,
|
||||
"value": value
|
||||
}
|
||||
```
|
||||
:::
|
||||
|
||||
:::js
|
||||
```typescript
|
||||
import { entrypoint, interrupt } from "@langchain/langgraph";
|
||||
|
||||
const myWorkflow = entrypoint(
|
||||
{ checkpointer, name: "workflow" },
|
||||
async (inputs: { t0: number }) => {
|
||||
const t1 = Date.now();
|
||||
|
||||
const deltaT = t1 - inputs.t0;
|
||||
|
||||
if (deltaT > 1000) {
|
||||
const result = await slowTask(1);
|
||||
const value = interrupt("question");
|
||||
return { result, value };
|
||||
} else {
|
||||
const result = await slowTask(2);
|
||||
const value = interrupt("question");
|
||||
return { result, value };
|
||||
}
|
||||
}
|
||||
);
|
||||
```
|
||||
:::
|
||||
|
||||
=== "Correct"
|
||||
|
||||
:::python
|
||||
In this example, the workflow uses the input `t0` to determine which task to execute. This is deterministic because the result of the workflow depends only on the input.
|
||||
|
||||
```python
|
||||
@@ -654,19 +1111,48 @@ Please read the section on [determinism](#determinism) for more details.
|
||||
t0 = inputs["t0"]
|
||||
# highlight-next-line
|
||||
t1 = get_time().result()
|
||||
|
||||
|
||||
delta_t = t1 - t0
|
||||
|
||||
|
||||
if delta_t > 1:
|
||||
result = slow_task(1).result()
|
||||
value = interrupt("question")
|
||||
else:
|
||||
result = slow_task(2).result()
|
||||
value = interrupt("question")
|
||||
|
||||
|
||||
return {
|
||||
"result": result,
|
||||
"value": value
|
||||
}
|
||||
```
|
||||
:::
|
||||
|
||||
:::js
|
||||
In this example, the workflow uses the input `t0` to determine which task to execute. This is deterministic because the result of the workflow depends only on the input.
|
||||
|
||||
```typescript
|
||||
import { entrypoint, task, interrupt } from "@langchain/langgraph";
|
||||
|
||||
const getTime = task("getTime", () => Date.now());
|
||||
|
||||
const myWorkflow = entrypoint(
|
||||
{ checkpointer, name: "workflow" },
|
||||
async (inputs: { t0: number }): Promise<any> => {
|
||||
const t1 = await getTime();
|
||||
|
||||
const deltaT = t1 - inputs.t0;
|
||||
|
||||
if (deltaT > 1000) {
|
||||
const result = await slowTask(1);
|
||||
const value = interrupt("question");
|
||||
return { result, value };
|
||||
} else {
|
||||
const result = await slowTask(2);
|
||||
const value = interrupt("question");
|
||||
return { result, value };
|
||||
}
|
||||
}
|
||||
);
|
||||
```
|
||||
:::
|
||||
|
||||
Reference in New Issue
Block a user