关于 Functional API 的概念性介绍,请参阅 Functional API。
创建简单工作流
在定义entrypoint 时,输入仅限于函数的第一个参数。要传入多个输入,可以使用字典。
Copy
@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})
扩展示例:简单工作流
扩展示例:简单工作流
Copy
import uuid
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(uuid.uuid4())}}
result = workflow.invoke({"number": 7}, config=config)
print(result)
扩展示例:使用 LLM 撰写文章
扩展示例:使用 LLM 撰写文章
本示例演示了
@task 和 @entrypoint 装饰器的用法。由于提供了 checkpointer,工作流的结果将被持久化到 checkpointer 中。Copy
import uuid
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(uuid.uuid4())}}
result = workflow.invoke("the history of flight", config=config)
print(result)
并行执行
通过并发调用任务并等待结果,可以实现任务的并行执行。这对于提升 IO 密集型任务(例如调用 LLM API)的性能非常有用。Copy
@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]
扩展示例:并行 LLM 调用
扩展示例:并行 LLM 调用
本示例演示了如何使用 本示例利用 LangGraph 的并发模型缩短执行时间,尤其适用于涉及 LLM 补全等 I/O 操作的任务。
@task 并行运行多个 LLM 调用。每次调用针对不同主题生成一个段落,结果合并为单一文本输出。Copy
import uuid
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(uuid.uuid4())}}
result = workflow.invoke(["quantum computing", "climate change", "history of aviation"], config=config)
print(result)
调用图
Functional API 与 Graph API 共享相同的底层运行时,可在同一应用中混合使用。Copy
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
}
扩展示例:从 Functional API 调用简单图
扩展示例:从 Functional API 调用简单图
Copy
import uuid
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(uuid.uuid4())}}
print(workflow.invoke(5, config=config)) # Output: {'bar': 10}
调用其他 entrypoint
你可以在 entrypoint 或 task 内部调用其他 entrypoint。Copy
@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
扩展示例:调用另一个 entrypoint
扩展示例:调用另一个 entrypoint
Copy
import uuid
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(uuid.uuid4())}}
print(main.invoke({"x": 6, "y": 7}, config=config)) # Output: {'product': 42}
流式输出
Functional API 使用与 Graph API 相同的流式机制。请阅读 流式输出指南 了解更多详情。 以下示例演示了如何使用流式 API 同时流式传输更新和自定义数据。Copy
from langgraph.func import entrypoint
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.config import get_stream_writer
checkpointer = InMemorySaver()
@entrypoint(checkpointer=checkpointer)
def main(inputs: dict) -> int:
writer = get_stream_writer()
writer("Started processing")
result = inputs["x"] * 2
writer(f"Result is {result}")
return result
config = {"configurable": {"thread_id": "abc"}}
for mode, chunk in main.stream(
{"x": 5},
stream_mode=["custom", "updates"],
config=config
):
print(f"{mode}: {chunk}")
- 从
langgraph.config导入get_stream_writer。 - 在 entrypoint 内部获取一个流写入器实例。
- 在计算开始前发送自定义数据。
- 计算结果后再发送一条自定义消息。
- 使用
.stream()处理流式输出。 - 指定要使用的流式模式。
Copy
('updates', {'add_one': 2})
('updates', {'add_two': 3})
('custom', 'hello')
('custom', 'world')
('updates', {'main': 5})
Python < 3.11 下的异步代码
如果使用 Python < 3.11 并编写异步代码,
get_stream_writer 将无法正常工作。请直接使用 StreamWriter 类。详见 Python < 3.11 下的异步代码。Copy
from langgraph.types import StreamWriter
@entrypoint(checkpointer=checkpointer)
async def main(inputs: dict, writer: StreamWriter) -> int:
...
重试策略
Copy
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)
Copy
'OK'
缓存任务
Copy
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))
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}
for chunk in main.stream({"x": 5}, stream_mode="updates"):
print(chunk)
#> {'slow_add': 10}
#> {'slow_add': 10, '__metadata__': {'cached': True}}
#> {'main': {'result1': 10, 'result2': 10}}
ttl以秒为单位指定,缓存将在该时间后失效。
错误后恢复执行
Copy
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,因为其结果已保存在 checkpoint 中。
Copy
main.invoke(None, config=config)
Copy
'Ran slow task.'
人机协作
Functional API 支持使用interrupt 函数和 Command 原语实现人机协作工作流。
基本人机协作工作流
我们将创建三个 task:- 追加
"bar"。 - 暂停等待人工输入,恢复时追加人工输入。
- 追加
"qux"。
Copy
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"
Copy
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
step_1)的结果会被持久化,因此在 interrupt 之后不会重新执行。
让我们发送一个查询字符串:
Copy
config = {"configurable": {"thread_id": "1"}}
for event in graph.stream("foo", config):
print(event)
print("\n")
step_1 之后因 interrupt 而暂停。interrupt 提供了恢复运行的说明。要恢复执行,需发出一个包含 human_feedback 任务所期望数据的 Command。
Copy
# Continue execution
for event in graph.stream(Command(resume="baz"), config):
print(event)
print("\n")
审阅工具调用
为了在执行前审阅工具调用,我们添加一个review_tool_call 函数,该函数调用 interrupt。调用该函数时,执行将暂停,直到我们发出恢复命令。
给定一个工具调用,我们的函数将 interrupt 以供人工审阅。此时我们可以:
- 接受工具调用
- 修改工具调用并继续
- 生成自定义工具消息(例如,指示模型重新格式化其工具调用)
Copy
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"]
)
ToolMessage。先前任务(本例中为初始模型调用)的结果会被持久化,因此在 interrupt 之后不会重新执行。
Copy
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)
短期记忆
短期记忆允许在同一 thread id 的不同调用之间存储信息。详情请参阅 短期记忆。管理 checkpoint
你可以查看和删除 checkpointer 存储的信息。查看线程状态
Copy
config = {
"configurable": {
"thread_id": "1",
# optionally provide an ID for a specific checkpoint,
# otherwise the latest checkpoint is shown
# "checkpoint_id": "1f029ca3-1f5b-6704-8004-820c16b69a5a" #
}
}
graph.get_state(config)
Copy
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=()
)
查看线程历史
Copy
config = {
"configurable": {
"thread_id": "1"
}
}
list(graph.get_state_history(config))
Copy
[
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=()
)
]
解耦返回值与保存值
使用entrypoint.final 可以将返回给调用者的值与持久化到 checkpoint 中的值解耦。这在以下场景中非常有用:
- 你希望返回一个计算结果(例如摘要或状态),但保存一个不同的内部值供下次调用使用。
- 你需要控制下次运行时传入 previous 参数的内容。
Copy
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
聊天机器人示例
以下是一个使用 Functional API 和InMemorySaver checkpointer 的简单聊天机器人示例。
该机器人能够记住之前的对话,并从上次结束的地方继续。
Copy
from langchain.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"}
for chunk in workflow.stream([input_message], config, stream_mode="values"):
chunk.pretty_print()
input_message = {"role": "user", "content": "what's my name?"}
for chunk in workflow.stream([input_message], config, stream_mode="values"):
chunk.pretty_print()
长期记忆
长期记忆 允许在不同 thread id 之间存储信息。这对于在一次对话中学习用户的相关信息并在另一次对话中使用非常有用。工作流
- 参阅 工作流与 agent 指南,了解更多使用 Functional API 构建工作流的示例。
与其他库集成
- 使用 Functional API 为其他框架添加 LangGraph 特性:为不内置持久化、记忆和流式输出功能的 agent 框架添加 LangGraph 特性。
连接这些文档 到 Claude、VSCode 等,通过 MCP 获取实时答案。

