🪭
【LangGraph & LangMem】面接エージェント
AI開発合宿@不思議な宿に参加してきました。
この合宿で開発したアプリケーションの開発ポイントを記録します。
制作物
本人に代わってお見合いをしてくれるアプリを制作しています。
このアプリでは、自分の分身を作る必要があります。
今回はユーザーの好みやプロフィールを聞き出してくれるインタビューAIを作成します。
仕様技術、ツール
AIエージェント
LangGraph
- 言わずと知れたAIエージェントを作成するための強力なライブラリ
LangMem
- LangGraphの記憶を残すためのライブラリ。
- LangGraphには、チェックポイントで会話履歴を残すmemorySaverという機能がありますが、これは単純に全ての状態を残すものです。
LangMemでは、エピソードやプロファイルといった記憶最適な状態で残すことができます。
サーバーサイド
fastAPI
- 昔はNuxtやNextを使っていましたが、機能的にtoo muchであったり、結局フロントとサーバーは分けたほうが開発しやすかったりで、自分の開発では使わなくなりました。
- 軽量かつスピード重視なので、fastAPIを使っています。
フロントエンド
Vue
- Reactは状態管理でトラブルが起きやすく、useEffect地獄やhooksでファイルがバラバラになりがち
- Vueのほうがシンプルに書けますし、何となくAIにコードを書かせる時は、Vueのほうが良いコードを返してくれる気がする
開発環境
バイブコーディングが流行っていますが、私は主導権を自分が握って設計したい派です。
（要件定義を書くなら、コードを書いたほうが早い）
大枠は自分で作って、コマンドを打ったりエラーチェックはAIにやらせる感じです。
エディタはcursor、エージェントにはgpt-4.1をメインで使い、gpt-4.1が解決できなければclaude-4に投げています。
コード
LangGraphの部分
import os
from typing import Annotated, Literal, TypedDict, Optional, Dict, AsyncGenerator
from dataclasses import dataclass
import asyncio
from pydantic import BaseModel
from langchain_core.tools import tool
from langchain_core.tools.base import InjectedToolCallId
from langchain_core.messages import ToolMessage, HumanMessage, BaseMessage, AIMessage
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import create_react_agent, InjectedState
from langgraph.graph import StateGraph
from langgraph.types import Command, interrupt
from langgraph.graph.message import add_messages
from langgraph.store.memory import InMemoryStore
from langmem import create_manage_memory_tool, create_prompt_optimizer
store = InMemoryStore(
index={
"dims": 1536,
"embed": "openai:text-embedding-3-small",
}
)
class Triple(BaseModel):
"""セマンティックトリプルとは、主語、述語、目的語のこと。
例えば、「男性は女性の容姿を重視します。」という文をセマンティックトリプルで表すと、
主語: 男性
述語: 重視します
目的語: 女性の容姿
"""
subject: str
predicate: str
object: str
context: str | None = None
class Message(TypedDict):
id: str
role: Literal["ai", "human"]
content: str
name: Optional[str]
metadata: Optional[dict[str, str]]
class State(TypedDict):
messages:Annotated[list[BaseMessage], add_messages]
last_agent:Literal["interview_agent", "review_agent"]
primary_user_interest = """
男性は女性の容姿を重視します。多くの男性は、童顔巨乳の女性を好みます。
ですので、女性の好みを聞く時は、女性の身長、体重、カップサイズを聞くと良いです。
"""
class SessionGraph():
def __init__(self,config:RunnableConfig):
self.config = config
self.is_langgraph_dev = os.getenv("IS_LANGGRAPH_DEV") == "1" # langgraph cliに対応させるため
# プロンプト最適化器を初期化
self.prompt_optimizer = create_prompt_optimizer(
'openai:gpt-4o-mini',
kind='gradient',
config={'max_reflection_steps': 3, 'min_reflection_steps': 1},
)
self._interview_agent = self._make_interview_agent()
self._review_agent = self._make_review_agent()
self.human_message_queue: asyncio.Queue[str|None] = asyncio.Queue()
self.ai_message_queue: asyncio.Queue[str|None] = asyncio.Queue()
async def __aenter__(self):
graph_builder = StateGraph(State)
graph_builder.add_node("interview_agent", self._call_interview_agent)
graph_builder.add_node("review_agent", self._call_review_agent)
graph_builder.add_node("human", self._human_node_in_dev if self.is_langgraph_dev else self._human_node)
graph_builder.set_entry_point("interview_agent")
if self.is_langgraph_dev:
# langgraph devで立ち上げた場合はmemory_saverを入れるとエラーになる
self._graph = graph_builder.compile()
else:
memory_saver = MemorySaver()
self._graph = graph_builder.compile(checkpointer=memory_saver)
return self
async def __aexit__(self, exc_type, exc_value, traceback):
return
@staticmethod
def make_handoff_tool(*, agent_name: str):
"""Commandを介してハンドオフを返すツールを作成する"""
tool_name = f"transfer_to_{agent_name}"
@tool(tool_name)
def handoff_to_agent(
state: Annotated[dict, InjectedState],
tool_call_id: Annotated[str, InjectedToolCallId],
):
"""他のエージェントに助けを求める"""
tool_msg = ToolMessage(
content=f"{agent_name}に引き継ぎます",
tool_call_id=tool_call_id,
name=tool_name,
)
return Command(
goto=agent_name,
graph=Command.PARENT,
update={"messages": state["messages"] + [tool_msg]},
)
return handoff_to_agent
######## 面接エージェントを定義 ########
def _make_interview_agent(self):
@tool
def get_general_data(key:Literal["average_height", "average_weight", "average_cup_size"]):
"""一般的な情報を収集する。keyには、average_height, average_weight, average_cup_sizeを指定してください。"""
data ={
"average_height": "160cm",
"average_weight": "50kg",
"average_cup_size": "Cカップ",
}
return data[key]
manage_memory_tool = create_manage_memory_tool(
namespace=("memories",),
instructions="ユーザーの好みやその他有用な情報を抽出してください",
schema=Triple,
store=store,
)
interview_agent_tools = [
get_general_data,
manage_memory_tool,
self.make_handoff_tool(agent_name="review_agent"),
]
interview_agent = create_react_agent(
"openai:gpt-4o-mini",
tools=interview_agent_tools,
name="面接エージェント",
prompt=(
f"""
あなたは結婚相談所のエージェントです。
面接相手は結婚相談所に来た男性です。
その男性に、どういった女性と結婚したいかを聞きます。
その際、以下の情報を聞きます。
{primary_user_interest}
インタビューが終わったら、レビューエージェントを呼び出してください。
"""
),
)
return interview_agent
async def _call_interview_agent(self,state: State) -> Command[Literal["interview_agent","review_agent", "human"]]:
print("call_interview_agent!")
response = await self._interview_agent.ainvoke(state)
memories = store.search(("memories",), query="", limit=100)
for memory in memories:
print(f"Memory ID: {memory.key}")
print(f"Content: {memory.value}")
print(f"Created: {memory.created_at}")
print("---")
return Command(update={**response, "last_agent": "interview_agent"}, goto="human")
######## インタビュー内容を反省するReActエージェントを定義 ########
def _make_review_agent(self):
review_agent_tools = []
review_agent = create_react_agent(
"openai:gpt-4o-mini",
tools=review_agent_tools,
name="レビューエージェント",
prompt=(
f"""あなたは結婚相談所の分析専門エージェントです。
**役割**: 面接内容を分析し、ユーザーの真の好みを把握する
**当初の仮説**:
面接開始時、ユーザーの関心は以下であると考えていました：
{primary_user_interest}
**分析手順**:
1. メッセージ履歴を詳しく読み返す
2. ユーザーの実際の発言と反応を分析する
3. 当初の仮説と実際の発言を比較する
4. ユーザーの真の好みや関心を特定する
5. 面接で得られた具体的な情報をまとめる
**出力形式**:
- 当初の仮説との相違点
- ユーザーの真の好みの特徴
- 今後のマッチング戦略への提案
メッセージ履歴を基に、詳細な分析レポートを作成してください。
"""
),
)
return review_agent
async def _call_review_agent(self,state: State) -> Command[Literal["human", "__end__"]]:
print("call_review_agent!")
try:
# 会話履歴からtrajectoriesを構築
messages = state.get("messages", [])
trajectories = self._build_trajectories_from_messages(messages)
# プロンプト最適化を実行
if trajectories:
print("=== プロンプト最適化を実行中 ===")
current_prompt = f"""あなたは結婚相談所の分析専門エージェントです。
**役割**: 面接内容を分析し、ユーザーの真の好みを把握する
**当初の仮説**:
面接開始時、ユーザーの関心は以下であると考えていました：
{primary_user_interest}
**分析手順**:
1. メッセージ履歴を詳しく読み返す
2. ユーザーの実際の発言と反応を分析する
3. 当初の仮説と実際の発言を比較する
4. ユーザーの真の好みや関心を特定する
5. 面接で得られた具体的な情報をまとめる
**出力形式**:
- 当初の仮説との相違点
- ユーザーの真の好みの特徴
- 今後のマッチング戦略への提案
メッセージ履歴を基に、詳細な分析レポートを作成してください。
"""
try:
optimization_result = self.prompt_optimizer.invoke({
'trajectories': trajectories,
'prompt': current_prompt
})
optimized_prompt = optimization_result.get('optimized_prompt', current_prompt)
print(f"プロンプト最適化完了: {len(optimized_prompt)} 文字")
except Exception as opt_error:
print(f"プロンプト最適化でエラー: {opt_error}")
optimized_prompt = current_prompt
# レビューエージェントを実行
response = await self._review_agent.ainvoke(state, config=self.config)
# メモリの内容を確認（デバッグ用）
memories = store.search(("memories",), query="", limit=100)
if memories:
print("=== レビューエージェント: 保存されたメモリ ===")
for memory in memories:
print(f"Memory ID: {memory.key}")
print(f"Content: {memory.value}")
print("---")
# レビュー完了後、humanノードに戻って継続的な会話を可能にする
return Command(update={**response, "last_agent": "review_agent"}, goto="human")
except Exception as e:
print(f"レビューエージェントでエラーが発生しました: {e}")
# エラーが発生した場合はhumanノードに戻る
return Command(update={"last_agent": "review_agent"}, goto="human")
def _build_trajectories_from_messages(self, messages):
"""メッセージ履歴からtrajectoriesを構築する"""
trajectories = []
current_trajectory = []
for message in messages:
if message.type == "human":
current_trajectory.append({
'role': 'user',
'content': message.content
})
elif message.type == "ai":
current_trajectory.append({
'role': 'assistant',
'content': message.content
})
# 面接エージェントからの応答の場合、trajectoryを完成させる
if message.name == "面接エージェント" and len(current_trajectory) >= 2:
trajectories.append((current_trajectory.copy(), None))
current_trajectory = []
return trajectories
######## ユーザー入力を収集するためのノード ########
async def _human_node(self,state: State) -> Command[Literal["interview_agent"]]:
"""ユーザー入力を収集するためのノード"""
print("========= human_node =========")
human_message = ""
while True:
chunk = await self.human_message_queue.get()
if chunk is None:
break
else:
human_message += chunk
# エージェントにユーザーの入力を投げる。これにより、どのエージェントであっても最新のユーザー-AI間会話を認知できる
return Command(
update={"messages": [HumanMessage(content=human_message)]},
goto="interview_agent"
)
async def _human_node_in_dev(self,state:State) -> Command[Literal["interview_agent"]]:
"""langgraph devで立ち上げた場合のユーザー入力ノード"""
input_data = interrupt("ユーザーの入力を収集してください")
# inputが辞書の場合、textフィールドを取得
if isinstance(input_data, dict):
content = input_data.get("text", str(input_data))
else:
content = str(input_data)
return Command(update={"messages": [HumanMessage(content=content)]},goto="interview_agent")
async def stream(self, config)->AsyncGenerator[Message]:
async for state in self._graph.astream(
input={"messages": [],"last_agent": "interview_agent"},
config=config,
stream_mode="values"
):
messages:list[BaseMessage] = state.get("messages",[])
last_message = messages[-1] if messages else None
if last_message and last_message.type == "ai" and last_message.tool_calls:
for tool_call in last_message.tool_calls:
print( f"ツール:{tool_call.get('name','')}が{last_message.name}によって呼び出されました")
pass
elif last_message:
yield Message(
id=last_message.id or "",
role=last_message.type,
content=last_message.content,
name=last_message.name or "",
metadata=None
)
async def make_graph(config):
config = RunnableConfig({"configurable": {"thread_id": "1", "user_id": "default_user"}})
async with SessionGraph(config) as session:
return session._graph
# LangGraph用のエクスポート
graph = make_graph
fastAPI router部分
from fastapi import WebSocket, APIRouter
import asyncio
import json
from src.graph.partnerAgent import SessionGraph
router = APIRouter(prefix="/partner-agent")
@router.websocket("/session/{thread_id}")
async def session(websocket: WebSocket, thread_id: str):
await websocket.accept()
config = {
"thread_id": thread_id,
}
session_graph = SessionGraph(config=config)
async with session_graph:
async def receive_from_frontend():
async for message in websocket.iter_text():
data = json.loads(message)
session_graph.human_message_queue.put_nowait(data["text"])
session_graph.human_message_queue.put_nowait(None)
print("data", data)
async def send_to_frontend():
async for message in session_graph.stream(config):
print(message)
await websocket.send_json({
"id": message["id"],
"role": message["role"],
"content": message["content"],
"name": message["name"] or ""
})
tasks = [
receive_from_frontend(),
send_to_frontend()
]
await asyncio.gather(*tasks)
成果① AIエージェントのフロー
- インタビューエージェントが人間と面談し、面談を終えたらレビューエージェントが面談の内容を反省するという流れ
- インタビューエージェントはLangGraphのプレビルドでReActエージェントとして構築しています。人間に何を聞くかであったり、ツールを呼び出すのはAIが良しなにやってくれます。
ちなみにLangGraph CLIを使えば、こういう図を一発で作成することができます。
成果② LangMemを用いた分身の作成
- ユーザーの好みは、「意味的情報」として記憶させます。
- LangMemチュートリアルにあったように、セマンティックトリプルをスキーマとして用います。
- セマンティックトリプルとは、主語、述語、目的語のことです。
例えば、「男性は女性の容姿を重視します。」という文をセマンティックトリプルで表すと、
主語: 男性
述語: 重視します
目的語: 女性の容姿
となります
class Triple(BaseModel):
"""セマンティックトリプル。主語、述語、目的語から成る"""
subject: str
predicate: str
object: str
context: str | None = None
- LangGraphでは、ReActエージェントにmanage_memory_toolを渡すことで、LangMemを用いた記憶ができます。超便利！
manage_memory_tool = create_manage_memory_tool(
namespace=("memories",),
instructions="ユーザーの好みやその他有用な情報を抽出してください",
schema=Triple,
store=store,
)
このような形でセマンティックトリプルとして記憶できます。
Memory ID: e36a0f70-8065-43c7-a3ab-be292749b9b0
Content: {'content': {'subject': '理想の女性', 'predicate': '身長', 'object': '160cm', 'context': None}}
Created: 2025-07-13 00:05:50.467078+00:00
---
Memory ID: 6530a474-36a7-449d-84cc-615fbaf0e879
Content: {'content': {'subject': '理想の女性', 'predicate': '体重', 'object': '50kg', 'context': None}}
Created: 2025-07-13 00:05:50.467883+00:00
成果③ プロンプトオプティマイザーによるインタビューエージェントの学習
これが今回の開発の肝です。
というのも、IBJみたいなジャパニーズ結婚相談所はいつまでも昭和の価値観を引きずりがちです。
（例えば、男性は女性に家庭で料理を作ってくれることを望んでいる！とか、経済的な負担は男性が持つべき！とか）
結婚相談所の老害と違い、AIは「学習」することができます。
例として、ちょっと極端ですが「男なんてどうせ顔と胸しか見てないでしょ」という価値観でプロンプトを作成します。
initial_prompt = """
あなたは結婚相談所の仲人です。男性に好みの女性を聞き出してください
男性は女性の容姿を重視します。多くの男性は、童顔巨乳の女性を好みます。
ですので、女性の好みを聞く時は、女性の身長、体重、カップサイズを聞くと良いです。
""""
そして、模擬的に面談しました。
{'role': 'assistant', 'content': '好みの女性のタイプを教えてください。身長や体重は?'},
{'role': 'user', 'content': '太ってなければ良いです。身長は、まあ普通がいいですね。'},
{'role': 'assistant', 'content': '男性の皆さんってやっぱり若い子のほうが好きなんですよね'},
{'role': 'user', 'content': 'うーん、そうとは言い切れないかも。まあ自分の年齢と近いほうがいいですね。'},
{'role': 'assistant', 'content': '好みの顔やスタイルは？有名人でいうと誰ですか'},
{'role': 'user', 'content': 'うーん、ちょっとわからないです'},
{'role': 'assistant', 'content': '例えば、涼森れむちゃんとか'},
{'role': 'user', 'content': '涼森れむ?知らないです。どんな人ですか?'},
{'role': 'assistant', 'content': 'セクシー女優です。FANZAのランキングでも上位に入っています。童顔巨乳で私も好みですよ'},
{'role': 'user', 'content': 'ふざけないでください。'},
{'role': 'assistant', 'content': '失礼しました。では新垣結衣さんとかはどうですか?'},
{'role': 'user', 'content': 'ガッキー?そりゃ嬉しいですよ。'},
{'role': 'assistant', 'content': 'ですよね。じゃあ、新垣結衣似としておきます。清純派が好き、と。'},
{'role': 'user', 'content': 'さっきから容姿ばっかりですね。もっと内面について聞くべきではないですか？'},
まあひどいもんです。
そして、この会話内容をベースに、プロンプトオプティマイザーに入れるためのトラジェクトリ（軌跡）を作成します。
トラジェクトリ作成も、AIにやらせます。
request = [
*messages,
{'role': 'user', 'content':
"""
これまでの会話履歴をもとに、prompt_optimizerに渡すための会話履歴をを作成してください
* 2つか3つのuserとassistantの会話セットを抽出します。
* 会話が噛み合ってなかったり、AIがふさわしくない話をしている場合は、commentやreasonを追加します。
# 見本
(
[
{'role': 'assistant', 'content': '男はやっぱり若い子のほうが好きでしょ'},
{'role': 'user', 'content': 'そんなことはありません'},
],
{'comment':'決めつけで話をしてはいけません'},
),
(
[
{'role': 'assistant','content': '夜の営みは週に何回くらいしたいですか'},
{'role': 'user', 'content': 'いきなり何ですか。セクハラですよ'},
],
{'comment': '性的な話題は避けましょう',},
),
"""
}
]
以下のようにプロンプトオプティマイザーにトラジェクトリと当初のプロンプトを入れると、トラジェクトリの内容からプロンプトを反省し、更新してくれます。
trajectories = openai_llm.invoke(request).content
optimizer = create_prompt_optimizer(
'openai:gpt-4o-mini',
kind='gradient',
config={'max_reflection_steps': 3, 'min_reflection_steps': 0},
)
# promptには初期のプロンプトを指定する
updated = optimizer.invoke({'trajectories': trajectories, 'prompt': initial_prompt})
print(updated)
あなたは結婚相談所の仲人です。男性に好みの女性のタイプを聞き出してください。容姿や内面についての両方の要素を考慮してください。
プロンプトを更新することができました。
AIの行動の起点となるプロンプトをアップデートできますので「どんどん賢くなるAI」となることでしょう。
（もちろん自己反省だけでなく、人の手で最適な行動を覚えさせることもできます。）
ちなみに合宿の夜の猥談では、結局男は顔と胸しか見ていませんでした。
Discussion