LangGraph の Functional APIの概要

Functional APIとは

Functional API は、LangGraph が提供する新しいLLM実行環境を構築するためのAPIです。

これまで、LangGraphは有向非巡回グラフ（DAG）を構築するGraph API を中心に、永続的なメモリ機能やチェックポイント機能 などを追加し、LLMアプリケーション構築のためのエコシステムと連携できるようライブラリを拡張してきました。

Graph APIの詳細については、以下の記事も参考にしてください。

https://zenn.dev/pharmax/articles/8796b892eed183

一方、2025年1月末に発表されたのが「Functional API」は従来の Graph APIとは異なるワークフローを構築 するためのAPIです。

• 公式発表では、従来のDAGベースの機能を「Graph API」、今回新たに発表された機能を「Functional API」と呼んでいます。
• この機能は、2025年2月現在、この機能はベータ版として提供されています。

Functional APIとGraph APIの構文の違い

Functional APIのシンプルなコードは次のようになります。

from langgraph.func import entrypoint, task

@task
def task1(value: str) -> str:
    return f'value: {value}'

@entrypoint()
def workflow(params: dict) -> dict:
    print(params)
    value = task1('1').result()

    return {
        'task1': value,
    }

print(workflow.invoke({'param': 'param1'}))

Functional APIの基本となるコンポーネントは、現時点では次の2つのみです。

コンポーネント	説明
EntryPoint	@entrypointというデコレーターで定義する。ワークフローの実行フローを管理し、その名の通り実行のエントリーポイントとしてinvokeすることができる。
Task	@taskというデコレーターで定義する。API呼び出しやデータ処理など、ワークフロー内の個別の処理単位を表す。非同期実行が可能であり、複数の処理を並行して実行できる。結果はチェックポイントとして保存され、途中から再開が可能。

一方、同等な処理をGraphAPIで実装すると以下のようになります。

from langgraph.graph import StateGraph
from typing_extensions import TypedDict


# Stateの定義
class State(TypedDict):
    params: dict
    task1_result: str

# Nodeの定義
def task1(state: State):
    value = f'value: {state["params"]["value"]}'
    return {'task1_result': value}

# Graphの定義
builder = StateGraph(State)
builder.add_node('task1', task1)

# Edgeの定義
builder.set_entry_point('task1')
builder.set_finish_point('task1')

graph = builder.compile()
print(graph.invoke({'params': {'value': '1'}}))

GraphAPIを使うためには次のコンポーネントが最低限必要です。

コンポーネント	説明
Graph（StateGraph）	エージェントのワークフローをグラフとしてモデル化するためのクラス。ユーザー定義の状態（State）をパラメータとして受け取り、ノードとエッジを組み合わせてワークフローを構築します。
State	アプリケーションの現在のスナップショットを表す共有データ構造。通常、TypedDict や Pydantic の BaseModel などの Python 型で定義され、ワークフロー内のノード間で共有されます。
Node	エージェントのロジック表現する関数。現在の State を入力として受け取り、更新された State を返します。
Edge	現在の State に基づいて、次に実行する Node を決定する関数。条件分岐や固定の遷移を定義することができます。

主な違い

「Graph API」と「Functional API」の主な違いをいくつか紹介します。

① Stateの管理

従来の「Graph API」では、グラフ全体で共通のState（状態）を共有する方式 を採用しています。各Nodeは必ずStateを受け取り、処理の結果として変更されたStateを返す必要がありました。

一方、「Functional API」のTaskは 独立した関数 として実行され、Stateを受け取りません。

「Graph API」の設計における課題の一つは、ステートフルなグラフ構造の中で、共通機能をどのようにコンポーネント化するか という点でした。しかし、「Functional API」では関数単位で管理できるため、状態管理がシンプルになります。

def task1(state: State):
    value = f'value: {state["params"]["value"]}'
    return {'task1_result': value}

「Graph API」のNodeはStateを受け取って更新するStateを返す

@task
def task1(value: str) -> str:
    return f'value: {value}'

「Functional API」のTaskは独立した関数で状態を持たない

② グラフ構造の可視化

「Graph API」で構築したワークフローはDAG（有向非巡回グラフ） であるため、実行前に可視化が可能です。また、グラフ構造が明確に定義されているため、フローの設計や構造変更が容易 なのが特徴です。

一方、「Functional API」は実行フローを動的にプログラムで構成 するため、可視化はサポートされていません。そのため、フローの構造化は実装者自身が設計する必要 があります。

「Graph API」はフローを可視化できる

その他公式で対比の紹介がされているため、参照ください。

Functional APIとGraph APIの構文の互換性

Functional APIもGraph APIも　Pregel という共通機構で動作するため、相互の互換があります。

Graph API -> Functional API

Graph APIのNode内から、Functional APIの Endpoint を呼び出すことができます。

class State(TypedDict):
    input: str
    workflow_result: str


def node(state: State):
    # GraphのNodeからFunctional APIのendpointをinvokeする
    value = workflow.invoke({'param': 'param1'})
    return {'workflow_result': value}


builder = StateGraph(State)
builder.add_node('node', node)

builder.set_entry_point('node')
builder.set_finish_point('node')

graph = builder.compile()

print(graph.invoke({'input': 'graph_input'}))

また、Functional APIの TaskもどうようにGraphのNodeから呼び出し可能です。

@task
def task(value: str) -> str:
    return f'value: {value}'

def node(state: State):
    # taskをnodeから直接呼ぶこともできる
    result = task('1').result()
    return {'workflow_result': result}

class State(TypedDict):
    input: str
    workflow_result: str

builder = StateGraph(State)
builder.add_node('node', node)

builder.set_entry_point('node')
builder.set_finish_point('node')

graph = builder.compile()

Functional API -> Graph API

逆のパターンとして、Functional APIの Taks 内でGraph APIを呼び出すことも可能です。

graph = builder.compile()

@task
def task(value: str) -> str:
    # Functional APIのtaskからGraphをinvokeする
    graph.invoke({'input': value})
    return f'value: {value}'


@entrypoint()
def workflow(params: dict) -> dict:
    value = task('1').result()

    return {
        'task1': value,
    }

print(workflow.invoke({'input': 'graph_input'}))

同様に、Endpoint からもGraphをinvokeすることができます。

graph = builder.compile()

@task
def task(value: str) -> str:
    return f'value: {value}'


@entrypoint()
def workflow(params: dict) -> dict:
    value = task('1').result()
    # EntryPointからも呼び出し可能
    graph.invoke({'input': value})

    return {
        'task1': value,
    }

print(workflow.invoke({'input': 'graph_input'}))

「Functional API」のユースケース

私が考える「Functional API」のユースケースをいくつか紹介します。

シンプルなフローで十分な場合

「Graph API」を用いたワークフローは視覚的に明確であり、複雑で大規模なエージェントフローでも高いメンテナンス性 を維持できます。

しかし、シンプルな処理にもグラフ構築のための前処理が必要となるため、やや冗長に感じることがあります。例えば、LLMを1回呼び出すだけのシンプルなワークフローであれば、「Functional API」を使うほうが効率的 だと考えています。

シンプルな並列処理

「Graph API」で並列処理を行う方法はいくつかあります。（並列なグラフを組み立てる、RunnableParallelを使う など）
しかし、どれも使いやすいとは言えず、それなりに実装の工夫が必要になります。

一方、「Functional API」ではTaskを使うことで並列処理をシンプルに実装でき、トレースなども従来どおり行うことが可能 です。

もちろん、複雑な並列処理については「Graph API」で並列なグラフを構築するほうがメンテナンス性が高いかもしれません。しかし、シンプルな並列処理であれば「Functional API」 のほうが適していると感じます。

ステートレスな機能コンポーネント

「Graph API」で大規模な機能を実装する際、共通処理を別のグラフ（サブグラフ）としてネストすることがあります。
しかし、サブグラフを使うとState管理が複雑化する という課題を感じることがありました。

「Graph API」は基本的にステートフルであり、グラフ全体でStateが共有される仕組み になっています。そのため、グラフが大規模になるほどStateのコントロールが難しくなる傾向があります。

一方、「Functional API」はStateを持たないため、独立したコンポーネント設計がしやすい という利点があります。そのため、共通で使われる機能 については「Functional API」のほうがメンテナンス性が高いのではないかと考えています。

まとめ

「Functional API」を活用することで、従来の「Graph API」では対応が難しかった課題をより柔軟に解決できる可能性があります。

また、NodeとTaskに互換性がある ため、既存の「Graph API」を使用した実装とも組み合わせやすい点も大きな利点です。

「Functional API」は「Graph API」の代替機能ではなく、ワークフロー構築をより柔軟にするための拡張機能 という位置づけにあると考えていて、今後の機能拡充に期待しつつ、積極的に活用していきたいと思います。