Functional API 允许你以最小的代码改动，为你的应用程序添加 LangGraph 的关键特性（持久化、记忆、人机协同和流式处理）。它旨在将这些特性集成到现有代码中，这些代码可能使用标准语言原语进行分支和控制流，例如 if 语句、for 循环和函数调用。与许多需要将代码重构为显式管道或 DAG 的数据编排框架不同，Functional API 允许你在不强制使用刚性执行模型的情况下，集成这些能力。Functional API 使用两个关键构建块：
from langgraph.checkpoint.memory import InMemorySaverfrom langgraph.func import entrypoint, taskfrom langgraph.types import interrupt@taskdef write_essay(topic: str) -> str: """Write an essay about the given topic.""" time.sleep(1) # A placeholder for a long-running task. return f"An essay about topic: {topic}"@entrypoint(checkpointer=InMemorySaver())def workflow(topic: str) -> dict: """A simple workflow that writes an essay and asks for a review.""" essay = write_essay("cat").result() is_approved = interrupt({ # Any json-serializable payload provided to interrupt as argument. # It will be surfaced on the client side as an Interrupt when streaming data # from the workflow. "essay": essay, # The essay we want reviewed. # We can add any additional information that we need. # For example, introduce a key called "action" with some instructions. "action": "Please approve/reject the essay", }) return { "essay": essay, # The essay that was generated "is_approved": is_approved, # Response from HIL }
import timefrom langchain_core.utils.uuid import uuid7from langgraph.func import entrypoint, taskfrom langgraph.types import interruptfrom langgraph.checkpoint.memory import InMemorySaver@taskdef write_essay(topic: str) -> str: """Write an essay about the given topic.""" time.sleep(1) # This is a placeholder for a long-running task. return f"An essay about topic: {topic}"@entrypoint(checkpointer=InMemorySaver())def workflow(topic: str) -> dict: """A simple workflow that writes an essay and asks for a review.""" essay = write_essay("cat").result() is_approved = interrupt( { # Any json-serializable payload provided to interrupt as argument. # It will be surfaced on the client side as an Interrupt when streaming data # from the workflow. "essay": essay, # The essay we want reviewed. # We can add any additional information that we need. # For example, introduce a key called "action" with some instructions. "action": "Please approve/reject the essay", } ) return { "essay": essay, # The essay that was generated "is_approved": is_approved, # Response from HIL }thread_id = str(uuid7())config = {"configurable": {"thread_id": thread_id}}for item in workflow.stream("cat", config): print(item)# > {'write_essay': 'An essay about topic: cat'}# > {# > '__interrupt__': (# > Interrupt(# > value={# > 'essay': 'An essay about topic: cat',# > 'action': 'Please approve/reject the essay'# > },# > id='b9b2b9d788f482663ced6dc755c9e981'# > ),# > )# > }
文章已经写好，可以审阅了。一旦提供审阅，我们就可以恢复工作流：
from langgraph.types import Command# 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.human_review = Truefor item in workflow.stream(Command(resume=human_review), config): print(item)
{'workflow': {'essay': 'An essay about topic: cat', 'is_approved': False}}
from langgraph.func import entrypoint@entrypoint(checkpointer=checkpointer)def my_workflow(some_input: dict) -> int: # some logic that may involve long-running tasks like API calls, # and may be interrupted for human-in-the-loop. ... return result
from langgraph.func import entrypoint@entrypoint(checkpointer=checkpointer)async def my_workflow(some_input: dict) -> int: # some logic that may involve long-running tasks like API calls, # and may be interrupted for human-in-the-loop ... return result
@entrypoint(checkpointer=checkpointer)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 # for the `previous` parameter. return entrypoint.final(value=previous, save=2 * number)config = { "configurable": { "thread_id": "1" }}my_workflow.invoke(3, config) # 0 (previous was None)my_workflow.invoke(1, config) # 6 (previous was 3 * 2 from the previous invocation)
幂等性确保多次运行相同的操作会产生相同的结果。这有助于防止在某个步骤因故障而重新运行时出现重复的 API 调用和冗余处理。始终将 API 调用放在任务函数中进行检查点，并将它们设计为幂等的，以防重新执行。
这对于导致数据写入的操作尤其重要。
当工作流恢复时，LangGraph 从检查点重放已完成的任务结果。一个已开始但未完成的任务可能会在该次恢复时再次运行，因此请将副作用设计为幂等的。使用幂等键或验证现有结果以避免意外的重复。
@entrypoint(checkpointer=checkpointer)def my_workflow(inputs: dict) -> int: # This code will be executed a second time when resuming the workflow. # Which is likely not what you want. with open("output.txt", "w") as f: f.write("Side effect executed") value = interrupt("question") return value
在此示例中，副作用被封装在任务中，确保了恢复时执行的一致性。
from langgraph.func import task@taskdef write_to_file(): with open("output.txt", "w") as f: f.write("Side effect executed")@entrypoint(checkpointer=checkpointer)def my_workflow(inputs: dict) -> int: # The side effect is now encapsulated in a task. write_to_file().result() value = interrupt("question") return value
