勉強会「【LangChainゆる勉強会#16】LangGraph Functional APIをさわってみる」を開催しました #StudyCo

import random
import uuid

from langgraph.checkpoint.memory import MemorySaver
from langgraph.func import entrypoint, task
from langgraph.types import Command, interrupt


@task
def generate_random_number() -> int:
    return random.randint(0, 100)


@entrypoint(checkpointer=MemorySaver())
def workflow(topic: str) -> dict:
    random_number = generate_random_number().result()
    is_approved = interrupt({"random_number": random_number})
    return {
        "is_approved": is_approved,
        "random_number": random_number,
    }

ワークフローでは「generate_random_number」というタスクを呼び出しています。

@entrypoint(checkpointer=MemorySaver())
def workflow(topic: str) -> dict:
    random_number = generate_random_number().result()
    is_approved = interrupt({"random_number": random_number})
    return {
        "is_approved": is_approved,
        "random_number": random_number,
    }

Functional APIの公式ドキュメントにも記載されていますが、ワークフローの関数内でランダム性のある処理を行うのはNGです。以下のようなコードにしてしまうと、interrupt（Human-in-the-Loop）の前後でrandom_numberの値が再計算され、異なる値になってしまう可能性があるわけです。

@entrypoint(checkpointer=MemorySaver())
def workflow(topic: str) -> dict:
    random_number = random.randint(0, 100)
    is_approved = interrupt({"random_number": random_number})
    return {
        "is_approved": is_approved,
        "random_number": random_number,
    }

LangGraph Functional APIでは、@entrypointが付けられたワークフローをinterrupt後に再実行すると、@taskをつけた各関数の結果はCheckpointerから得るのだと思われます。

これはHuman-in-the-Loopやエラー時に続きから実行する処理を組み込むためです。そこで、LLMの呼び出しなどは@taskをつけた関数内で実施する必要がある、ということになります。

Functional APIのサンプルコード

LangGraphの公式ドキュメントには、Functional APIのサンプルコードもいくつか追加されています。

たとえば以下は、Functional APIでFunction callingを繰り返すエージェント（LangGraphで言うReActエージェント）を実装する例です。

langchain-ai.github.io

他にもFunctional APIでHandsoff型のマルチエージェントを実装する例も公開されています。

langchain-ai.github.io

Functional APIを使うと、Graph APIよりも通常のプログラミングらしくワークフローを構築できることがわかります。

SubGraphのようなワークフローがうまく動かない例

最後に余談ですが、LangGraphのGraph APIでは、SubGraph中でのinterruptがうまく動作しないという問題があります。

Functional APIもSubGraphのようなことができるのですが、やはりinterruptがうまく動作しないようでした。

@entrypoint()
def sub_workflow(topic: str) -> dict:
        :
    is_approved = interrupt(
        :

@entrypoint(checkpointer=MemorySaver())
def workflow(topic: str) -> dict:
        :
    sub_workflow_result = sub_workflow.invoke({"topic": topic})
        :