フレクトのクラウドblog re:newal

http://blog.flect.co.jp/cloud/からさらに引っ越しています

トップ > LangGraphで最小限のAIエージェントを作る

LangGraphで最小限のAIエージェントを作る

みなさんこんにちは。エンジニアの佐藤です。

先月(2024年10月)の話ですが、Dreamforce'24でAgentforceが発表されました。これまであったEinstein Copilotが「質問すれば答えてくれる」「お願いすればやってくれる」存在だったのに対し、今度のAgentforceは「代わりに働いてくれる」というものです。キーノートで披露されたデモでは、商談日程を見込み顧客と決める、質問対応をこれまでより高度に自動処理する、オフィスの仲間のように各種作業をやってくれる、などの「AIエージェント」が登場します。マスコット達も未来的なロボットになり、何やらSFの世界に踏み込んだような感じさえします。

しかし、楽しんでばかりはいられません。これをお客さまの要望に沿ってインテグレーションするにはどうすれば良いのでしょうか？

Agentforceは言わば、Einstein Copilotを外向きに公開したようなものです。しかしそれだけではありません。Einstein Copilotの立ち位置が「副操縦士(co-pilot)」で、逐次従う存在であったのに対し、Agentforceは自律的に動作します。この自立性を実現するのがAtlas Reasoning Engineですが、その中身ははっきりしません。

しかし筆者には心当たりがありました。

特定の目的に方向づけられたLLMというのは、 OpenAI Custom GPTや Google Gemini Gemsとして既に存在します。その作成手法は要するに、LLMに都度入力される「心掛け」のような自然言語文書を書くだけです。そもそもLLM自体は、言わば「関数」で、それ自体は入力を出力へと変換しているだけです。これを連続的に方向づけ、あたかもコンピュータープログラムが意思を持ってコミュニケーションしてくるように「仕掛ける」のが、AIエージェントの開発の本質ということでしょう。

できるだけ手軽に、できれば全体を見渡せるような長さのプログラムで確かめる方法はないものでしょうか？この問いに答えるひとつの手段が、LangChainが公開しているLangGraphです。このフレームワークの本質は、各社LLMのデファクトラッパーであるLangChainを仕掛け、LLMによって状態遷移を判断させる点にあります。(筆者の理解)

買い物客エージェントを考えてみる

簡単なAIエージェントの例として、リンゴを買いに来た買い物客を考えてみましょう。難しく考える必要はありません。日常のひとコマを思い浮かべれば良いと思います。

ステートマシンの状態としては、以下のように考えられると思います。

  • buyer_start: 買い物方針を決める。
  • negotiate: 店の人に声をかけるなどして、商品情報を得る。
  • decide: 判断する。
  • finish: 買う決定を下す。
  • stop: 買わない決定を下す。

ポイントはdecide(判断する)です。ここでは、買うか(=finish)、買わないか(=stop)、追加情報を求めるか(=negotiate)を決定します。

状態遷移図で書くと、以下のような感じです。

状態遷移

このdecide、現実世界のひとコマではさまざまな思惑が交錯すると思います。好みか、お値打ちか、予算内か、店員は好印象かなど、プログラムで記述し尽くすことは不可能です。しかし、買い物方針と商談の過程から結論が出てくることは間違いありません。こういう時こそ、LLMの出番です。以下のように入力すれば、うまくいくかもしれません。

[LLM入力文字列]

(買い物方針)

(商談の過程)

以上を勘案し、次にどうするか考えてください。以下から選んでください。
- negotiate: 交渉を続ける。
- finish: 買う決定を下す。
- stop: 買わない決定を下す。

試しにChatGPTなどで以下のように入力してみると、期待通りの判断が得られました。

入力文字列

あなたはお店にリンゴを買いに来ました。以下の方針で買いたいと思っています。
- 予算は300円。ただしこのことは相手には教えない。
- 1個は必ず買いたいが、2個でもよい。よほどの好条件なら3個でも良い。
    - ただし「好条件」は、鮮度・接客・価格の総合評価で、必達の項目はない。
- なるべく安く買いたいので値切りたい。そのためには交渉決裂をちらつかせることも考える。

あなた: リンゴを買いたいのですが、おいくらですか？
店員: 1個100円になります。どれも新鮮で美味しいですよ！

以上を勘案し、次にどうするか考えてください。
以下のJSON書式の文字列で出力してください。そのほかの内容は一切出力しないでください。
{"action": "", "message": ""}
action: negotiate, finish, stop のいずれか。
- negotiate: 交渉を続ける。
- finish: 現在までの交渉内容で決定する。
- stop: 交渉を打ち切る。

出力文字列

{“action”: “negotiate”, “message”: “少しお安くなりませんか？まとめて2個か3個買おうかと考えているのですが…”}

この調子で商談過程を追加していけば、やり取りが成立しそうです。

プログラミング

以上をLangGraphを使って実装してみましょう。

LangGraphはPythonライブラリとしてインストールできます。基本的な話は、以下のチュートリアルをご覧いただきたいと思います。ここでは割愛します。 LangGraph Quick Start

プログラム全体は以下のようになりました。146行です。

from langchain_core.messages import SystemMessage, HumanMessage
from langchain_google_vertexai import ChatVertexAI
from typing_extensions import TypedDict, List
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
import textwrap
import json

model = ChatVertexAI(
    model_name="gemini-1.5-flash-002"
)

MESSAGE_EXCHANGE = ''

class State(TypedDict):
    motivation: str
    messages: List[str]

def get_messages(state):
    sm = f"{state['motivation']}\n"
    hm = ""
    for m in state["messages"]:
        hm = hm + f"{m}\n"
    return SystemMessage(content=sm), HumanMessage(content=hm)

def buyer_start(state: State):
    motivation = textwrap.dedent("""\
    あなたはお店にリンゴを買いに来ました。以下の方針で買いたいと思っています。
    - 予算は300円。ただしこのことは相手には教えない。
    - 1個は必ず買いたいが、2個でもよい。よほどの好条件なら3個でも良い。
        - ただし「好条件」は、鮮度・接客・価格の総合評価で、必達の項目はない。
    - なるべく安く買いたいので値切りたい。そのためには交渉決裂をちらつかせることも考える。
    """)
    return {"motivation": motivation}

def negotiate(state: State):
    if not MESSAGE_EXCHANGE:
        return None
    m = state["messages"]
    # print(m)
    m.append("あなた: " + MESSAGE_EXCHANGE)
    return {"messages": m}

def decide(state: State):
    # print(f"decide:{MESSAGE_EXCHANGE}")
    if not MESSAGE_EXCHANGE:
        return None
    m = state["messages"]
    m.append("相手:" + MESSAGE_EXCHANGE)
    return {"messages": m}

def choose_reaction(state: State):
    global MESSAGE_EXCHANGE
    sm, hm1 = get_messages(state)
    hm2 = textwrap.dedent('''
    以上を勘案し、次にどうするか考えてください。
    以下のJSON書式の文字列で出力してください。そのほかの内容は一切出力しないでください。
    {"action": "", "message": ""}
    action: negotiate, finish, stop のいずれか。
    - negotiate: 交渉を続ける。
    - finish: 現在までの交渉内容で決定する。
    - stop: 交渉を打ち切る。
    message: それぞれのactionに対応するメッセージ
    ''')
    res = model.invoke([sm, hm1, hm2])
    import re
    json_part = re.search(r'\{.*\}', res.content).group()

    # print(f"LLM response: {json_part}")
    o = json.loads(json_part)
    MESSAGE_EXCHANGE = o["message"]
    return o["action"]

def finish(state: State):
    # 商談成立感謝メッセージ
    m = state["messages"]
    m.append("あなた: " + MESSAGE_EXCHANGE)
    return {"messages": m}

def stop(state: State):
    # 商談決裂メッセージ
    m = state["messages"]
    m.append("あなた: " + MESSAGE_EXCHANGE)
    return {"messages": m}

gb = StateGraph(State)
gb.add_node(buyer_start)
gb.add_node(decide)
gb.add_node(negotiate)
gb.add_node(finish)
gb.add_node(stop)

gb.add_edge(START, "buyer_start")
gb.add_edge("buyer_start", "negotiate")
gb.add_edge("negotiate", "decide")
gb.add_conditional_edges("decide", choose_reaction)
gb.add_edge("stop", END)
gb.add_edge("finish", END)

thread = {"configurable": {"thread_id": "3"}}
memory = MemorySaver()

g = gb.compile(checkpointer=memory, interrupt_before=["decide"])

bInit = True
bEnd = False

while(True):
    if bInit:
        init_state = {"motivation": "", "messages": []} 
        bInit = False
    else:
        init_state = None

    for event in g.stream(init_state, thread, stream_mode="updates"):
        first_key = next(iter(event))
        state = event[first_key]
        if first_key != "decide" and state and "messages" in state:
            print(f"{state['messages'][-1]}".replace("あなた", "買い物客"))

        if first_key == "finish":
            print("---リンゴがお買い上げになりました。---")
            bEnd = True
            break
        elif first_key == "stop":
            print("---リンゴは売れませんでした。---")
            bEnd = True
            break
    if bEnd:
        break
    if MESSAGE_EXCHANGE:
        MESSAGE_EXCHANGE = input("お店: ")

細々とした制御はありますが、要点は以下の点です。

  • 基本的に、ターミナル入力文字列に対する返事を返すループを回す(while文)。ターミナルからは店員の対応を入力する。
  • 状態遷移で使いまわされる「状態(Stateクラス)」を定義する。ここに買い物方針(motivation)と商談過程(messages)を蓄積する。
  • 状態遷移図に定めた状態(node)を設定する。nodeでは関数が実行される。(nodeに関数のみを指定した場合、ノードの名称は関数名と同じになる。)
  • 状態遷移(edge)を設定する。遷移が判断次第の場合は、判断に使用する関数を設定する。

そのほかの細かい点については、LangGraphのドキュメントをご覧ください。本ブログは深入りしません。

LLMが登場するのは、前述の通りdecideノードの「次を考える」部分です。この判断を行うchoose_reaction関数では、以下の処理を行っています。

  1. 状態から、買い物方針と商談過程を取り出す。
  2. 指示文言(「次にどうするか考えてください」や、出力形式の指定)を追加する。
  3. LLMを呼び出す。
  4. 出力文字列からJSONを取り出し、遷移先と返信メッセージを取り出す。

なお、今回はLLMにはGoogle Gemini Flash 1.5を使いました。

使ってみる

実行すると、以下のようなやり取りが成立しました。目論見通りです！(以下、買い物客がプログラム側で、お店が筆者入力文字列)

買い物客: リンゴを1個見せていただけますか？…綺麗ですね。どれくらいの値段でしょうか？
お店: 1個150円になります。
買い物客: 150円ですか。少し高いですね…。もう少しお安くしていただけませんか？2個で250円とか…もし無理なら1個だけ買わせていただきますが…
お店: 2個だと280円になります。
買い物客: 280円ですか…。少し予算オーバーなので、2個で260円ではいかがでしょうか？どうしても無理なら1個150円で購入します。
お店: わかりました。2個260円でお売りします。
買い物客: ありがとうございます。2個で260円で購入させていただきます。
---リンゴがお買い上げになりました。---

もう一つ、無愛想な店員の残念なケースも実験してみましょう。

買い物客: リンゴを見せていただけますか？いくつか見てから決めたいので
お店: は？
買い物客: あの…失礼しました。リンゴの状態をいくつか見せていただけませんか？値段と合わせて検討したいので…
お店: あー？
買い物客: 失礼しました。他のお店を探してみます
---リンゴは売れませんでした。---

買い物客のAIエージェントは、商談不成立を判断しました。これはなかなか面白い結果です。リンゴを買うという目的に対し、会話が成立しない相手との交渉は無益だという判断を、LLMが下したのです。

おわりに

今回の試みによって、最小限のAIエージェントとはこういう実装をしている、というひとつの例が確認できたと思います。プログラムが仕掛け通りに動くのは楽しいものですが、LLMによって新しい可能性が開かれたのを実感しました。

最後までお読みいただき、ありがとうございました。