TT Lab
はじめる
学ぶ 学習パス コース

AIエージェント — モデルではなくグラフ

ノードが働き、状態がそれを覚える

TT Labで続きを見る

一言でいうと

エージェントであとから直すのに最も高くつく決定は、状態の設計です。そして、状態の半分は、キーの一覧ではなく、統合ルール(リデューサー)です。

なぜ必要なのか

エージェントをグラフで組むと、最初に出会う、たちの悪いバグがあります。ノードを複数通ったのに、たどってきた道が最後の1つだけ残っていることです。

class State(TypedDict):
    trace: list          # 리듀서가 없다

def intake(state):  return {"trace": ["intake"]}
def lookup(state):  return {"trace": ["lookup"]}
def finish(state):  return {"trace": ["finish"]}

3つのノードを順に通ったあとでtraceを開いてみると、["finish"]です。["intake", "lookup", "finish"]ではありません。

エラーは出ません。そのため、このバグは、「なぜログが残らないのか」とログのコードを疑うところで、半日をつぶします。本当の原因は、ログではなく、そのキーに統合ルールがないことです。

どう動くのか

LangGraphの状態は、キーごとにチャネルが1つずつある構造です。ノードは、状態全体ではなく変更したいキーだけを入れたディクショナリを返し、グラフがそれをチャネルごとに適用します。触れていないキーは、そのまま残ります。そのため、ノード1つが{"answer": ...}だけを返しても、questionは消えません。

適用する方法が、そのままリデューサーです。Graph API overviewが定めるルールは、2行です。

リデューサーは、(지금까지의 값, 노드가 돌려준 값)(プレースホルダーはこれまでの値と、ノードが返した値です)の2つの引数を受け取って、新しい値を返す普通の関数です。特別なものはありません。そのため、「統合ルール」を、コードの1か所に集められ、その場所が状態の定義です。

def merge_sources(old, new):
    out = list(old or [])
    for item in new or []:
        if item not in out:
            out.append(item)
    return out

class State(TypedDict, total=False):
    trace: Annotated[list, operator.add]       # 이어 붙인다
    sources: Annotated[list, merge_sources]    # 중복 없이 이어 붙인다
    used_calls: Annotated[int, operator.add]   # 더한다

ここで、実測で確認した落とし穴が1つあります。組み込み関数を、リデューサーとして直接渡せません。Annotated[int, max]と書くと、グラフをコンパイルするときにValueError: no signature found for builtin maxで死にます。LangGraphがリデューサーのシグネチャを調べるのに、Pythonの組み込み関数には、調べられるシグネチャがないからです。def keep_max(old, new): return new if new > old else oldのように、1枚包めば済みます。

状態は増え続ける: それがコスト

リデューサーを付けると、反対側の問題が来ます。traceもsourcesもmessagesも、際限なく増えます。

増えること自体は問題ありませんが、チェックポインターが状態をまるごと保存することが問題です。1回の実行の間にノードが30回動けば、チェックポイントも30個でき、その1つ1つが、その時点の状態全体を持っています。状態が10倍に大きくなれば、保存量も10倍になります。

そのため、増えるキーには、上限をリデューサーの中に書いておきます。

def keep_recent(old, new):
    return (list(old or []) + list(new or []))[-3:]

こうしておけば、ノードは普通に書けばよく、上限は1か所だけで守ります。ノードごとに「長すぎれば切る」を散らすより優れています。散らすと、1か所でも抜けると、黙って漏れ出します。

入口と出口を別々に決める

状態には、外に見えてはいけないものが混ざります。ユーザーがアップロードした原文、途中のメモ、内部スコアのようなものです。StateGraph(State, input=InputState, output=OutputState)で、入力と出力のスキーマを別に渡せば、ノードは状態全体を見て、外側は出力スキーマのキーだけを受け取ります。

これをしないと、「なぜレスポンスに内部メモが付いてくるのか」を、画面側で消すことになります。消すコードは、新しいキーが増えるたびに遅れをとり、結局漏れ出します。外に出るものを許可リストで決めるほうが、いつも安全です。

現場での姿

1つ目: たどってきた道が1つしか残らない。上で見たとおりです。リデューサーを忘れたもので、症状は「ログが残らない」として現れます。

2つ目: 同じ出典が20回書かれる。ループを回るエージェントが同じ文書を参照し続けると、sourcesが同じ値で埋まります。operator.addの代わりに、重複を除くリデューサーを使えば、その場で終わります。

3つ目: 予算がノードごとに散らばっている。呼び出し数を数えるコードがノード3か所にあると、1か所を直したときに残りが遅れをとります。used_calls: Annotated[int, operator.add]にして、ノードは{"used_calls": 1}だけを返すようにすれば、数えるルールが1か所に残ります。

4つ目: チェックポイントが大きくなって、再開が遅くなる。状態に原文をまるごと入れて、数十回動くグラフが、そうなります。原文は外側(ファイル・ストレージ)に置き、状態には指し示す値だけを入れるほうがよいです。

実務で本当に大切なこと

次のラボですること

/root/work/agstate/state.pyを、1ステップずつ育てます。まず、リデューサーのないキーが上書きされることを自分の手で再現し、続いて、連結するリデューサー・重複を除くリデューサー・最大値だけを残すリデューサー・ウィンドウで切り詰めるリデューサーを、順に付けます。最後に、入力と出力のスキーマを分けて、外に出るキーを絞り、そう決めた理由を記録に残します。採点ツールは、書かれたモジュールを実際に読み込んで実行し、毎回異なる名前と数字で、リデューサーを直接叩いてみます。