LangGraph

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

LangGraph を使う理由

一部の AI workflow は単一の prompt ではありません。分岐し、loop し、tool を呼び出し、場合によっては続行する前に人間の承認を待つために停止する必要があります。これを 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 が付与された method は、id の下に graph を登録します。未コンパイルの StateGraph を返すと、Squid がそれを compile します。
Checkpointing。 Squid が checkpointer を提供するため、graph の state は call 間で保持され、interrupt 後も維持されます。State は thread id ごとにグループ化されます。
Client API。 squid.langGraph(graphId) は、security rules によって保護された任意の Squid client から graph を invoke、resume、inspect します。

LangGraph を使うタイミング

Use Case	Recommendation
分岐、loop、人間の承認を伴う 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 を使用

仕組み

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

Quick Start

前提条件

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

Step 1: graph を宣言する

method に @langgraph を付与し、未コンパイルの 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

Step 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 の保護を参照してください。

Step 3: backend を deploy する

squid deploy

Step 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

Core Concepts

Threads

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

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

Run results

invoke、resume、getState はすべて同じ形を返して解決されます。

Field	Description
`ok`	成功時は常に `true`。
`threadId`	run が実行された thread。
`status`	graph が完了した場合は `'completed'`、resume を待っている場合は `'interrupted'`。`getState` はこの field を返しません。
`state`	最終的な（または paused 状態の）state values、または `null`。
`next`	次に実行される node。graph が完了している場合は空です。

Interrupt and resume

node 内で LangGraph の interrupt() を呼び出すと、graph を一時停止して decision を caller に返します。interrupt() に渡した値は caller に公開され、caller が後で resume() に渡した値が、その 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 は空の next とともに completed state を返します。

Long runs

graph execution は backend IPC timeout によって上限が設定されており、この page のすべての method で現在 4 分です。

invoke と resume は graph が停止するまで promise を保持します。promise を保持するのが扱いづらいほど時間がかかる graph では、invokeAsync と resumeAsync を使用してください。これらはすぐに job id を返し、その 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 を引き上げるものではありません。

Inspecting 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 は、@secure_langgraph rule が許可した graph にのみ到達できます。1 つの graph を保護するには graph id を指定して rule を宣言し、すべての graph を対象にするには引数なしで宣言します。

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	Description
`operation`	`'invoke'`、`'resume'`、`'getState'`、または `'deleteThread'`。
`graphId`	操作対象の graph。
`threadId`	operation が指定する場合の target thread。
`input`	`invoke` に渡される initial state。
`resumePayload`	`resume` に渡される payload。

operation を使用して、破壊的な call に対して read より高い基準を適用します。

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 は、より狭い per-graph rule を広げることはありません。graph が独自の @secure_langgraph('graph-id') rule を持つ場合、その rule も pass する必要があります。そのため、invoke のみを許可する graph-specific rule は、global rule が True を返していても getState と deleteThread を拒否します。

security rules に関する一般的な guidance については、Security rules を参照してください。

Error Handling

Error	Cause	Fix
`... not found`	その graph id の下に `@langgraph` method が登録されていない、またはそれを保持する backend が deploy されていない。	id string を確認し、backend を redeploy します。
`must return an uncompiled StateGraph`	`@langgraph` method が別のもの（多くの場合 compiled graph）を返した。	`StateGraph` 自体を返します。`.compile()` は呼び出さないでください。
node 内で発生した exception	graph 自体が失敗した。	message は caller に伝播します。想定される場合は node 内で処理してください。
`UNAUTHORIZED`	この caller に対する operation を許可する `@secure_langgraph` rule がない。	rule を追加または拡張します。LangGraph の保護を参照してください。

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

Best Practices

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

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

LangGraph を使う理由​

概要​

LangGraph を使うタイミング​

仕組み​

Quick Start​

前提条件​

Step 1: graph を宣言する​

Step 2: client が呼び出せるようにする​

Step 3: backend を deploy する​

Step 4: client から graph を invoke する​

Core Concepts​

Threads​

Run results​

Interrupt and resume​

Long runs​

Inspecting state​

LangGraph の保護​

Error Handling​

Best Practices​