Graphを使わないLangGraph？？LangGraphに新しいFunctional APIが追加されました！

LangChainが新しいブログを投稿してました。

冒頭部分に以下の記載があります。

グラフを明示的に定義せずに、ヒューマンインザループ、永続性/メモリ、ストリーミングなどの LangGraph のコア機能を活用したいと思ったことはありませんか?

はい、あります！
グラフが難しいなぁと思ってあまり手を出せずにいました。

続きを翻訳すると、、

PythonとJavaScriptで利用可能なLangGraphのFunctional APIのリリースを発表できることを嬉しく思います。
Functional APIを使用すると、より従来的なプログラミング パラダイムを使用してLangGraph機能を活用できるため、ヒューマンインザループ、短期および長期のメモリ、ストリーミング機能を組み込んだAIワークフローを簡単に構築できます。
Functional API はentrypointとtaskの2つのデコレータで構成されており、標準関数を使用してワークフローを定義し、通常のループと条件を使用して実行フローを制御できます。これにより、コードを再構築することなく、既存のアプリケーションにLangGraphの機能を簡単に導入できます。
このAPIはGraph API (StateGraph)を補完するものであり、両方のAPIが同じ基盤ランタイムを使用するため、併用することができます。これにより、2つのパラダイムを組み合わせて、両方の長所を活用した複雑なワークフローを作成できます。

なんだか、僕のために新しいAPIを用意してくれたようです。やったね。

Graph APIとFunctional APIで同じ機能を実装し、比較しました。

ブログ記事では明記がないものの、ドキュメント上はベータ版扱いのようです。API変更などが発生するかもしれませんのでご注意ください。

題材

Graph APIのクイックスタートにあるヒューマンインザループのコードを題材に比較します。

Graph APIでの実装

ドキュメント通りですが一通り解説します。

まず、ライブラリーをインポートします。

from typing import Annotated

from langchain_aws import ChatBedrockConverse
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.tools import tool
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.types import Command, interrupt
from typing_extensions import TypedDict

Stateを定義します。

class State(TypedDict):
    messages: Annotated[list, add_messages]

ツールを２つ（ヒューマンインザループと検索）を用意し、LLMにバインドします。

@tool
def human_assistance(query: str) -> str:
    """人間からの支援を要請する"""
    human_response = interrupt({"query": query})
    return human_response["data"]


serch_tool = TavilySearchResults(max_results=2)
tools = [serch_tool, human_assistance]

llm = ChatBedrockConverse(model="us.amazon.nova-pro-v1:0")
llm_with_tools = llm.bind_tools(tools)

LLMはAmazon Nova Proを使用しました

メモリーを生成します。

memory = MemorySaver()

ノード（チャットボットノードとツールノードを定義します。

def chatbot(state: State):
    message = llm_with_tools.invoke(state["messages"])
    # ツールの実行中に中断するため、再開時にツールの呼び出しが繰り返されないように、
    # 並列ツール呼び出しを無効にします。
    assert len(message.tool_calls) <= 1
    return {"messages": [message]}

tool_node = ToolNode(tools=tools)

グラフを構築します。

graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", tool_node)
graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,
)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")

graph = graph_builder.compile(checkpointer=memory)

今回はまだ単純なグラフと思うのですが、このコードが個人的にとっつきにくさ満点です🙈

構築したグラフはこんな感じです。

実行

実行します。（ヒューマンインザループツールを呼び出すような入力を行います。）

user_input = (
    "AI エージェントの構築には専門家の指導が必要です。サポートをお願いできますか?"
)
config = {"configurable": {"thread_id": "1"}}

events = graph.stream(
    {"messages": [{"role": "user", "content": user_input}]},
    config,
    stream_mode="values",
)

for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

出力

================================ Human Message =================================

AI エージェントの構築には専門家の指導が必要です。サポートをお願いできますか?
================================== Ai Message ==================================

[{'type': 'text', 'text': '<thinking> The user is asking for assistance in building an AI agent, which likely involves technical guidance and expertise. Since this is a specialized request, I will use the `human_assistance` tool to seek help from a human expert. </thinking>\n'}, {'type': 'tool_use', 'name': 'human_assistance', 'input': {'query': 'AIエージェントの構築に関する専門家の指導を求めています。サポートをお願いできますか？'}, 'id': 'tooluse_kj8mbrs1QV2NZ18pGnxKKA'}]
Tool Calls:
  human_assistance (tooluse_kj8mbrs1QV2NZ18pGnxKKA)
 Call ID: tooluse_kj8mbrs1QV2NZ18pGnxKKA
  Args:
    query: AIエージェントの構築に関する専門家の指導を求めています。サポートをお願いできますか？

human_assistanceツールの呼び出しで停止しますので、human_assistanceツールへの返答を行います。

human_response = (
    "私たち専門家がお手伝いします! エージェントを構築するには、LangGraph をチェックすることをお勧めします。"
    "単純な自律エージェントよりもはるかに信頼性が高く、拡張性があります。"
)

human_command = Command(resume={"data": human_response})

events = graph.stream(human_command, config, stream_mode="values")
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

出力

================================== Ai Message ==================================

[{'type': 'text', 'text': '<thinking> The user is asking for assistance in building an AI agent, which likely involves technical guidance and expertise. Since this is a specialized request, I will use the `human_assistance` tool to seek help from a human expert. </thinking>\n'}, {'type': 'tool_use', 'name': 'human_assistance', 'input': {'query': 'AIエージェントの構築に関する専門家の指導を求めています。サポートをお願いできますか？'}, 'id': 'tooluse_kj8mbrs1QV2NZ18pGnxKKA'}]
Tool Calls:
  human_assistance (tooluse_kj8mbrs1QV2NZ18pGnxKKA)
 Call ID: tooluse_kj8mbrs1QV2NZ18pGnxKKA
  Args:
    query: AIエージェントの構築に関する専門家の指導を求めています。サポートをお願いできますか？
================================= Tool Message =================================
Name: human_assistance

私たち専門家がお手伝いします! エージェントを構築するには、LangGraph をチェックすることをお勧めします。単純な自律エージェントよりもはるかに信頼性が高く、拡張性があります。
================================== Ai Message ==================================

エージェントの構築には専門家の指導が必要であることは理解しています。幸いにも、専門家からの助言があります。エージェントを構築するには、LangGraph を使用することをお勧めします。LangGraph は、単純な自律エージェントよりもはるかに信頼性が高く、拡張性があります。LangGraph の詳細については、ドキュメントやオンラインリソースを参照することをお勧めします。

human_assistanceツールの実行結果を受けて、AIが回答を生成してくれました。

Functional APIでの実装

Functional APIで、上記と同じ処理を実装してみます。

ライブラリーをインポートします。

from langchain_aws import ChatBedrockConverse
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.messages import BaseMessage, ToolMessage
from langchain_core.tools import tool
from langgraph.checkpoint.memory import MemorySaver
from langgraph.func import entrypoint, task
from langgraph.graph.message import add_messages
from langgraph.types import Command, interrupt

Functional APIでは、Stateは登場しません。

ツールを２つ（ヒューマンインザループと検索）を用意し、LLMにバインドします。先ほどと同じコードです。

@tool
def human_assistance(query: str) -> str:
    """人間からの支援を要請する"""
    human_response = interrupt({"query": query})
    return human_response["data"]


serch_tool = TavilySearchResults(max_results=2)
tools = [serch_tool, human_assistance]

llm = ChatBedrockConverse(model="us.amazon.nova-pro-v1:0")
llm_with_tools = llm.bind_tools(tools)

ノードの代わりにタスクを定義します。@taskデコレータを付与することでタスクとして定義します。
LLMを呼び出すタスクと、ツールを実行するタスクを定義します。

@task
def call_model(messages: list[BaseMessage]):

    message = llm_with_tools.invoke(messages)
    # ツールの実行中に中断するため、再開時にツールの呼び出しが繰り返されないように、
    # 並列ツール呼び出しを無効にします。
    assert len(message.tool_calls) <= 1
    return message

@task
def call_tool(tool_call):
    print(tool_call)
    tools_by_name = {tool.name: tool for tool in tools}
    tool = tools_by_name[tool_call["name"]]
    observation = tool.invoke(tool_call)
    return ToolMessage(content=observation, tool_call_id=tool_call["id"])

ブログ記事では、タスクについて以下のように解説されています。

タスク: API呼び出しやデータ処理ステップなど、エントリポイント内から非同期に実行できる個別の作業単位。タスクを呼び出すと、結果を取得するために待機したり、同期的に解決したりできるfutureのようなオブジェクトが返されます。

メモリーを生成します。（Graph APIと同様）

memory = MemorySaver()

Graph APIでのグラフに相当するワークフローを定義します。ワークフローは@entrypointデコレーターを付与します。

この部分が、冒頭の説明にあった「標準関数を使用してワークフローを定義し、通常のループと条件を使用して実行フローを制御できます」に該当します。

関数は一般的なPythonプログラムとして記述できるので、グラフの記述を新たに覚える必要はありません。

（その代わり、ツール呼び出し部分を愚直に実装する必要があります）

@entrypoint(checkpointer=checkpointer)
def workflow(inputs: list[BaseMessage], *, previous: list[BaseMessage]):
    messages = inputs or []

    if previous is not None:
        messages = add_messages(previous, messages)

    response = call_model(messages).result()
    while True:
        if not response.tool_calls:
            break

        # Execute tools
        tool_result_futures = [
            call_tool(tool_call) for tool_call in response.tool_calls
        ]
        tool_results = [fut.result() for fut in tool_result_futures]

        # Append to message list
        messages = add_messages(messages, [response, *tool_results])

        # Call model again
        response = call_model(messages).result()

    # Generate final response
    messages = add_messages(messages, response)
    return entrypoint.final(value=response, save=messages)

個人的にはこっちのほうがとっつきやすいです。

ブログ記事では、エントリーポイントについて以下のように解説されています。

エントリーポイント: ワークフロー ロジックをカプセル化し、長時間実行されるタスクや割り込みの処理を含む実行フローを管理するワークフローの開始点。

これでフローが完成です。

Functional APIはGraph APIのようにビジュアルでの処理フローを出力する機能はありません。

実行

実行します。呼び出し方はGraph APIをほぼ同じです。

user_input = (
    "AI エージェントの構築には専門家の指導が必要です。サポートをお願いできますか?"
)
config = {"configurable": {"thread_id": "1"}}

events = workflow.stream(
    [{"role": "user", "content": user_input}],
    config,
    stream_mode="values",
)

for event in events:
    event.pretty_print()

出力

{'name': 'human_assistance', 'args': {'query': 'AIエージェントの構築に関する専門家の指導を求めています。サポートをお願いできますか?'}, 'id': 'tooluse_y3GD0BbbRQeAzcpZnPSLnw', 'type': 'tool_call'}

Graph APIと同様、human_assistanceツールの呼び出しで停止しました。返答を送信します。

human_response = (
    "私たち専門家がお手伝いします! エージェントを構築するには、LangGraph をチェックすることをお勧めします。"
    "単純な自律エージェントよりもはるかに信頼性が高く、拡張性があります。"
)

human_command = Command(resume={"data": human_response})

events = workflow.stream(human_command, config, stream_mode="values")

for event in events:
    if event:
        event.pretty_print()

出力

{'name': 'human_assistance', 'args': {'query': 'AIエージェントの構築に関する専門家の指導を求めています。サポートをお願いできますか?'}, 'id': 'tooluse_y3GD0BbbRQeAzcpZnPSLnw', 'type': 'tool_call'}
================================== Ai Message ==================================

専門家からの助言によると、AIエージェントの構築にはLangGraphを使うことをおすすめします。LangGraphは、単純な自律エージェントよりも信頼性が高く、拡張性があります。

LangGraphについて詳しく知りたい場合や、エージェントの構築に関する具体的な質問がある場合は、お気軽にお問い合わせください。

human_assistanceツールの実行結果を受けて、AIが回答を生成してくれました。

まとめ

Graph APIとFunctional APIを比較しました。
Graph APIでは複雑なグラフ構造が実現できる一方、グラフを構築するという行為にあまり慣れていないためか、とっつきにくいと感じていました。Functional APIであれば、通常のプログラムロジックと同じ頭で考えられるので、わかりやすくていいなと思いました。

Functional APIはGraph APIを置き換えるものとかではなく、双方を組み合わせて使えることも明記されています。Functional APIでLangGraphを始めて、複雑なAIエージェントの実装に足を踏み入れたタイミングで、Graph APIを取り入れるようなアプローチが良いのかもしれません。