LangGraph

Python backend で LangGraph graph を定義し、任意の Squid client から呼び出します。

LangGraph を使用する理由

一部の AI workflow は単一の prompt では完結しません。branch、loop、tool call を行い、場合によっては継続前に人間の承認を待つために停止する必要があります。これを agent call の chain として表現する場合、state machine、persistence、resume logic を自身で管理することになります。

LangGraph は、これを persisted state を持つ node の graph としてすでにモデル化しています。Squid は Python backend 内で graph を実行し、state を checkpoint し、すべての Squid client に公開します。

Backend code

# Declare the graph once in your Python backend.
@langgraph('approval-graph')
def define_approval_graph(self) -> StateGraph:
    ...

Client code

// Run it from the client, pause on interrupt, and resume when the user decides.
const started = await squid.langGraph('approval-graph').invoke({ input: { request: 'refund #123' } });
if (started.status === 'interrupted') {
  await squid.langGraph('approval-graph').resume(started.threadId, { approved: true });
}

概要

LangGraph graph は state machine です。node は共有 state object の読み書きを行い、edge が次に実行する処理を決定します。Squid はこれに加えて 3 つの機能を提供します。

• Discovery。 Python backend service 内で @langgraph により decorate された method は、id の下で graph を登録します。uncompiled StateGraph を返すと、Squid がそれを compile します。
• Checkpointing。 Squid が checkpointer を提供するため、graph の state は call 間で保持され、interrupt 後も存続します。state は thread id ごとに group 化されます。
• Client API。 squid.langGraph(graphId) は、security rulesによって保護され、任意の Squid client から graph を invoke、resume、inspect できます。

LangGraph を使用する場合

ユースケース	推奨
branching、loop、または human approval を伴う multi-step workflow	✅ LangGraph
tool と knowledge base を任意で使用して質問に回答する	AI agent を使用
1 つの backend function を agent に公開する	AI functions を使用
schedule に従って code を実行する	Schedulers を使用
graph state なしで client から server-side logic を呼び出す	Executables を使用

仕組み

1. Python backend service が @langgraph('graph-id') で graph を宣言し、uncompiled StateGraph を返します。
2. client が squid.langGraph('graph-id').invoke({ input }) を呼び出します。Squid は、thread に対して backend 内で graph を実行します。thread を指定しない場合は新しい thread を作成します。
3. call は run の status、threadId、state、next node とともに resolve されます。
4. node が LangGraph の interrupt() を呼び出すと、run は status: 'interrupted' で停止し、その state が checkpoint されます。resume(threadId, payload) を呼び出すとその地点から継続し、payload は interrupt() call の戻り値になります。

クイックスタート

前提条件

• Python Squid backend project。Local developmentを参照してください。
• その backend の Python environment にインストールされた langgraph package。
• frontend にインストールされた @squidcloud/client package。

ステップ 1: graph を宣言する

method を @langgraph で decorate し、uncompiled StateGraph を返します。Squid が独自の checkpointer で graph を compile するため、.compile() を呼び出さないでください。

Backend code

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

from squidcloud_backend import SquidService, langgraph


class GraphState(TypedDict):
    # `operator.add` makes `messages` append across nodes instead of being replaced.
    messages: Annotated[list[str], operator.add]


class ExampleService(SquidService):
    @langgraph('simple-graph')
    def define_simple_graph(self) -> StateGraph:
        def first_node(state: GraphState) -> dict:
            incoming = state.get('messages') or ['no input']
            return {'messages': [f'Processed: {incoming[0]}']}

        def second_node(state: GraphState) -> dict:
            return {'messages': ['Graph completed']}

        graph = StateGraph(GraphState)
        graph.add_node('first_node', first_node)
        graph.add_node('second_node', second_node)
        graph.add_edge(START, 'first_node')
        graph.add_edge('first_node', 'second_node')
        graph.add_edge('second_node', END)

        # Return the uncompiled graph. Squid compiles it with its checkpointer.
        return graph

ステップ 2: client に呼び出しを許可する

API key がない場合、security rule が許可しない限り client call は拒否されます。@secure_langgraph rule を追加します。

Backend code

from squidcloud_backend import SquidService, secure_langgraph


class ExampleService(SquidService):
    @secure_langgraph('simple-graph')
    def allow_simple_graph(self, request: dict) -> bool:
        # request['operation'] is 'invoke', 'resume', 'getState', or 'deleteThread'.
        return self.is_authenticated()

完全な request object については、LangGraph の保護を参照してください。

ステップ 3: backend を deploy する

squid deploy

ステップ 4: client から graph を invoke する

Client code

import { Squid } from '@squidcloud/client';

const squid = new Squid({
  appId: 'YOUR_APP_ID',
  region: 'us-east-1.aws',
  environmentId: 'dev',
});

const result = await squid.langGraph('simple-graph').invoke({
  input: { messages: ['Hello LangGraph!'] },
});

console.log(result.status); // 'completed'
console.log(result.threadId); // Server-generated id for this run
console.log(result.state); // { messages: ['Hello LangGraph!', 'Processed: ...', 'Graph completed'] }
console.log(result.next); // [] once the graph has finished

コアコンセプト

Threads

thread は persisted graph state の単位です。すべての run は 1 つの thread に属します。

• threadId を省略すると、Squid は新しい thread を作成し、その id を返します。どちらも省略した 2 つの run は異なる thread を取得します。
• 特定の thread に対して実行するには、threadId を渡します。まだ存在しない thread を渡すと、その id を持つ新しい thread が開始されます。これにより、conversation ID や order ID のように、すでに持っているものを thread の key にできます。
• deleteThread(threadId) は thread の checkpoint を削除します。これは idempotent であるため、存在しない thread を削除しても成功します。削除後、getState は empty state を返し、その id に対する次の invoke または resume は新しく開始します。

Run result

invoke、resume、getState はいずれも同じ shape に resolve されます。

Field	説明
ok	成功時は常に true。
threadId	run が実行された thread。
status	graph が完了した場合は 'completed'、resume 待ちの場合は 'interrupted'。getState はこの field を返しません。
state	final state value または paused state value、あるいは null。
next	次に実行される node。graph が完了すると empty になります。

Interrupt と Resume

node 内で LangGraph の interrupt() を呼び出して graph を pause し、decision を caller に返します。interrupt() に渡した value は caller に公開され、caller が後で resume() に渡した value がその interrupt() call の戻り値になります。

Backend code

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import interrupt

from squidcloud_backend import SquidService, langgraph


class ApprovalState(TypedDict):
    messages: Annotated[list[str], operator.add]
    approved: bool


class ExampleService(SquidService):
    @langgraph('approval-graph')
    def define_approval_graph(self) -> StateGraph:
        def request_approval(state: ApprovalState) -> dict:
            # Execution stops here and the state is checkpointed.
            response = interrupt({'kind': 'approval', 'question': 'Continue processing?'})
            # After resume(), `response` holds the payload the caller sent.
            approved = response.get('approved', False) if isinstance(response, dict) else False
            return {'messages': [f'Approval response: {approved}'], 'approved': approved}

        def finalize(state: ApprovalState) -> dict:
            status = 'approved' if state.get('approved') else 'rejected'
            return {'messages': [f'Graph completed with status: {status}']}

        graph = StateGraph(ApprovalState)
        graph.add_node('request_approval', request_approval)
        graph.add_node('finalize', finalize)
        graph.add_edge(START, 'request_approval')
        graph.add_edge('request_approval', 'finalize')
        graph.add_edge('finalize', END)

        return graph

client から実行します。

Client code

const graph = squid.langGraph('approval-graph');

const started = await graph.invoke({ input: { messages: [], approved: false } });
console.log(started.status); // 'interrupted'
console.log(started.next); // ['request_approval']

// Later, once a human has decided.
const finished = await graph.resume(started.threadId, { approved: true });
console.log(finished.status); // 'completed'
console.log(finished.state?.messages); // [..., 'Graph completed with status: approved']

すでに完了した thread を resume しても error にはなりません。この call は、empty の next とともに completed state を返します。

Long run

このページのすべての method において、graph execution は backend IPC timeout によって制限されます。現在は 4 分です。

invoke と resume は、graph が停止するまで promise を保持します。promise を保持することが扱いにくくなるほど長い graph には、invokeAsync と resumeAsync を使用してください。これらはすぐに job ID を返し、Jobs APIに渡せます。

Client code

const jobId = await squid.langGraph('simple-graph').invokeAsync({
  input: { messages: ['Hello LangGraph!'] },
});

// Retrieve the result whenever you are ready.
const result = await squid.job().awaitJob(jobId);
console.log(result.status);

async variant は caller を待機から解放するだけです。4 分の execution cap は増加しません。

State の確認

Client code

const state = await squid.langGraph('approval-graph').getState(threadId);
console.log(state.state); // Current state values, or null
console.log(state.next); // Nodes that would run next, empty if the graph completed

// Remove the thread's checkpoints when you no longer need them.
await squid.langGraph('approval-graph').deleteThread(threadId);

LangGraph の保護

API key のない client が graph にアクセスできるのは、@secure_langgraph rule が許可する場合のみです。1 つの graph を保護するには graph ID を指定して rule を宣言し、すべての graph を対象にするには argument を指定しません。

Backend code

from squidcloud_backend import SquidService, secure_langgraph


class ExampleService(SquidService):
    # Applies to every graph in the application.
    @secure_langgraph()
    def allow_all_graphs(self, request: dict) -> bool:
        return self.is_authenticated()

rule は以下を含む dict を受け取ります。

Key	説明
operation	'invoke'、'resume'、'getState'、または 'deleteThread'。
graphId	operation の対象となる graph。
threadId	operation が指定する場合の対象 thread。
input	invoke に渡される initial state。
resumePayload	resume に渡される payload。

operation を使用して、read よりも destructive call に高い条件を課します。

Backend code

class ExampleService(SquidService):
    @secure_langgraph('approval-graph')
    def allow_approval_graph(self, request: dict) -> bool:
        auth = self.get_user_auth()
        if auth is None:
            return False
        if request.get('operation') == 'deleteThread':
            # Only the thread's owner may discard its history. This assumes threads are
            # named '<userId>:<something>' when they are created.
            thread_id = request.get('threadId') or ''
            return thread_id.startswith(f"{auth['userId']}:")
        return True

permissive な global rule で、より限定的な graph ごとの rule が緩和されることはありません。graph 独自の @secure_langgraph('graph-id') rule がある場合、その rule も通過する必要があります。したがって、global rule が True を返していても、invoke のみを許可する graph-specific rule は getState と deleteThread を拒否します。

security rule の一般的なガイダンスについては、Security rulesを参照してください。

Error Handling

Error	原因	修正方法
... not found	その graph ID で登録された @langgraph method がない、またはそれを保持する backend が deploy されていない。	ID string を確認し、backend を再 deploy してください。
must return an uncompiled StateGraph	@langgraph method が別のものを返した。多くの場合は compiled graph。	StateGraph 自体を返します。.compile() を呼び出さないでください。
node 内で exception が発生した	graph 自体が失敗した。	message は caller に propagate されます。予期される場合は node 内で処理してください。
UNAUTHORIZED	この caller に対し、どの @secure_langgraph rule も operation を許可しなかった。	rule を追加または拡張します。LangGraph の保護を参照してください。

invoke({}) と invoke({ input: {} }) はどちらも valid です。seed state を必要としない graph は、argument なしで開始できます。

ベストプラクティス

• graph を自身で compile しないでください。 Squid が thread と interrupt を機能させる checkpointer を使用して compile します。
• 意味のある値を thread の key にしてください。 conversation ID や order ID などの独自の threadId を渡すと、生成された ID を別途保存しなくても graph を resume できます。
• 完了した thread は削除してください。 deleteThread によって削除されるまで checkpoint は保持されます。
• node state を JSON-serializable に保ってください。 state は checkpoint のために process boundary をまたぎます。
• accumulating field には reducer を使用してください。 Annotated[list[str], operator.add] は node 間で append しますが、plain list[str] は最後に書き込んだ node の値に置き換えられます。
• graph ごとだけでなく operation ごとに保護してください。 deleteThread と resume は state を変更しますが、getState は読み取りのみです。