LLM のメモリの試験実装¶
(Version: 0.0.1)
最近 LLM についてコンテクストエンジニアリングという言葉をよく聞く。実は Web でなく API から使う LLM は「ステートレス」で、会話などの記憶はデフォルトではしてくれない。会話を続けようと思ったら、メッセージをユーザー側で記録しておき、それをいちいち API に与えないといけないのだ。古い記憶を呼び戻したりするには、単純に記録するだけでなく検索基盤も必要となる。そのような基盤をひっくるめてコンテクストと一般的に呼んでおり、それをどううまく構築するかが問題となっている。
特にコンテクストのうち、記憶の基盤は単に「メモリ」と呼ばれている。PC のメモリなどとごっちゃになって、わかりにくくもあるのだが、そう呼ばれているのだからしかたがない。
今回は、そのメモリ機能が実際にどう動きうるか、それを見てみたいと思って、私なりに実装してみた。基本的には LLM のツールコールという機能を使って、ツール群としてメモリ機能を実装することになる。
RAG をしたい場合、ゲームを解かせたい場合、その他、やりたい問題領域について良いメモリ機能の実装は変わってくると思われる。ただ、どういう場合でも通用するような一般的なメモリ機能があるのか？ あるとすればそれに近いものを実装したいと思った。
今回、とりあえずの試験実装にどういう例がいいか悩んだのだが、とりあえず以前、エージェントの実験として作った熊剣迷路問題を途中から解かせてみて、それでメモリ機能のツールを使ってくれないかと試すことにした。ただ、簡単な迷路問題ではあるが、先のような実装方針があったので、かなり本格的なメモリ機能を用意している。
…といっても、セマンティックサーチとかを用意するのは、めんどくさかったので、そういうバックエンドはこれまた AI さんに偽装してもらうことにした。
なお、熊剣迷路問題は↓からたどっていただきたい。
《JRF-2018/langchain_maze: 熊剣迷路問題 revisited》
https://github.com/JRF-2018/langchain_maze
ところで、今回のような手法はすでに限界が認識されていると思われる。ポケモンを解くとかにも必要だっただろう手法で、しかし、それには限界があったようだ。エージェントにコンテクストが大量消費される…とかの最近の話も、この手法と同じような手法についてのことで、その改善策もすでにいろいろ模索されているようだ。ネットではそういう話をチラチラ見る。ただ、そうとはいえ、しばらくは今回のような手法に似た手法もローカル LLM などを用いる場面などでは有効なのだと思われる。今回の試みが、何かの参考になれば幸いである。
結論¶
gemini-2.5-flash さんはあまりツールを使うのがうまくないが、gemini-2.5-pro さんにお願いするだけで、かなりうまくツールを使ってくれた。何度か実行することで、メモリ機能のテストはできたと思う。ただ、いきなり全部の機能を使ってくれる…とはならなかったのはやや想定外だった。
あと、Pro さんは結構無言でツールも使わずターンを終えることが何度かあった。これはおそらくツールを別の名前で呼んだり、似た別のツールを呼ぼうとして失敗しているのかもしれない。Flash さんはそういうことは見た記憶がないので、Pro さんは今そういう「バグ」があるということかも。
あいかわらず summarize_messages 周りでエラーが発生して汚い処理を残してある。今回のログにもその形跡は残したままにしてある。
再現性がないため、最初からの再実行はやっていないのはご容赦願いたい。
著者¶
JRF ( http://jrf.cocolog-nifty.com/statuses , Twitter (X): @jion_rockford )
ライセンス¶
基本短いコードなので(私が作った部分は)パブリックドメインのつもりです。気になる方は MIT License として扱ってください。
かなり AI さん達(Gemini さんや Claude さん)に教わって作っています。
実装¶
まず、必要なライブラリを読み込みます。
In [ ]:
!pip install -q -U langchain langgraph langchain-google-genai langmem
Gemini にアクセスします。シークレットで Gemini API キーを Google AI Studio からインポートすると GOOGLE_API_KEY というシークレットができるはずです。それを使います。
In [ ]:
from langchain_google_genai import ChatGoogleGenerativeAI
from google.colab import userdata
llm = ChatGoogleGenerativeAI(google_api_key=userdata.get('GOOGLE_API_KEY'), model="models/gemini-2.5-pro")
ちゃんと Gemini にアクセスできるかテストします。
In [ ]:
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
prompt = PromptTemplate(template="以下の質問に回答してください：{question}", input_variables=["question"])
chain = LLMChain(llm=llm, prompt=prompt)
response = chain.run({"question": "Geminiモデルの特徴を教えてください"})
print(response)
はい、承知いたしました。Googleが開発したAIモデル「Gemini」の主な特徴について、分かりやすく解説します。
Geminiは、単一のモデルではなく、性能と用途に応じて最適化されたモデルファミリー（シリーズ）です。その最大の特徴は、AIの新しい時代を切り拓く可能性を秘めた**「ネイティブなマルチモーダリティ」**にあります。
以下に、Geminiの主な特徴を具体的に説明します。
---
### 1. ネイティブなマルチモーダル性 (Native Multimodality)
これがGeminiの最も重要で革新的な特徴です。
* **「ネイティブ」の意味：** 従来のAIモデルの多くは、テキスト用、画像用、音声用など、別々のモデルを後から組み合わせてマルチモーダル（複数の種類の情報を扱えること）を実現していました。一方、Geminiは**開発の初期段階から、テキスト、画像、音声、動画、コードなど、さまざまな種類の情報を同時に学習**しています。
* **高度な理解と推論：** この設計により、Geminiは単に情報を認識するだけでなく、それらの関係性を深く理解し、より高度な推論を行うことができます。
* **具体例：**
* 紙に描かれた絵の動画を見せながら「これは何をしているところ？」と尋ねると、「アヒルの絵を描いていますね。次は色を塗る番です」と状況を理解して回答できます。
* 物理の問題の図と数式を見せると、両方を統合して問題を解くことができます。
### 2. 3つの異なるサイズのモデル
Geminiは、あらゆる用途に対応できるよう、3つのサイズで提供されています。
| モデル名 | 特徴 | 主な用途 |
| :--- | :--- | :--- |
| **Gemini Ultra** | **最も高性能で巨大なモデル。** 非常に複雑で専門的なタスクをこなすために設計されています。 | データセンター、エンタープライズ向けの高度なAIアプリケーション、科学研究など。 |
| **Gemini Pro** | **性能と効率のバランスが取れた万能モデル。** 幅広いタスクに対応できる主力モデルです。 | 現在のGoogleのAIチャットサービス「Gemini」（旧Bard）や、開発者向けAPIなどで利用されています。 |
| **Gemini Nano** | **最も軽量で効率的なモデル。** スマートフォンなどのデバイス上で直接動作するように設計されています。 | スマートフォンの要約機能、スマートリプライ（返信文の提案）など。オフラインでも動作し、プライバシー保護にも繋がります。 |
### 3. 最先端の性能と高度な推論能力
Gemini、特に最上位のUltraモデルは、多くの学術的なベンチマーク（性能評価指標）で非常に高いスコアを記録しています。
* **MMLUベンチマーク：** 法律、数学、物理、歴史、倫理など57の科目を組み合わせた大規模言語モデルの知識と問題解決能力を測るテストで、人間の専門家を上回るスコアを初めて達成したと報告されています。
* **高度な推論：** 単純な質疑応答だけでなく、複雑な情報の整理、計画立案、複数のステップが必要な問題の解決など、高度な思考が求められるタスクを得意とします。
* **優れたコーディング能力：** さまざまなプログラミング言語を理解し、コードの生成、デバッグ、アルゴリズムの説明などを行うことができます。
### 4. スケーラビリティと柔軟性
データセンターで稼働する巨大な「Ultra」から、スマートフォン上で動く「Nano」まで、同じアーキテクチャ（基本設計）を共有しながら、サイズを柔軟に変更できる点が強みです。これにより、開発者はさまざまなプラットフォームで一貫したAI体験を提供しやすくなります。
---
### まとめ
Geminiの特徴を要約すると以下のようになります。
* **核心：** テキスト、画像、音声などを**生まれながらに統合して理解**できる「ネイティブ・マルチモーダル」。
* **ラインナップ：** 用途に応じて選べる**Ultra、Pro、Nano**の3サイズ展開。
* **性能：** ベンチマークで最高水準を記録し、**複雑な推論やコーディング**を得意とする。
* **展開：** Googleの各種サービス（検索、広告など）から、開発者向けツール、個人のスマートフォンまで、**幅広く搭載**されていく。
Geminiは、人間が世界を認識する方法により近い形で情報を処理できるため、これまで以上に自然で高度なAIとの対話や共同作業を実現することを目指しています。
基本的なモジュールを読み込みます。
In [ ]:
import os
import numpy as np
import re
from pprint import pprint
from time import sleep
import pickle
# 座標が (np.int64(1), np.int64(2)) みたいに表示されないようにする。
np.set_printoptions(legacy='1.25')
セーブ／ロードのために次のコードを実行します。
In [ ]:
PLAY_GAME_SAVE = "langchain_maze.pickle"
ゲームのメインオブジェクト。ごく簡単な迷路というかダンジョンというか…。
In [ ]:
class Game:
initial_map = """\
■■■■■■■■■
■■■■■■■Ｇ■
■□□□□□■□■
■□■■■□□□■
■□■■■■■■■
■◎■■■■■△■
■□■■■■■□■
■□□□□□□□■
■■■■Ｓ■■■■
■■■■■■■■■
"""
def __init__ (self, initial_map=None, hint=True):
if initial_map is not None:
self.initial_map = initial_map
map = self.initial_map
self.map = map
self.written_map = re.sub("[◎△]", "？", map)
l = map.splitlines(True)
self.map_size = (len(l[0]) - 1, len(l))
self.hint = hint
self.actions = {
"上に行く": self.move_up,
"下に行く": self.move_down,
"左に行く": self.move_left,
"右に行く": self.move_right,
"熊を殺す": self.fight,
"剣を取る": self.get_sword,
"何もしない": self.do_nothing,
}
self.pos = self.get_start_pos()
self.sword = False
self.goal = False
self.prev_killed = False
self.kill_hint = False
def read_map (self, p):
x = p[0]
y = p[1]
if x < 0 or x >= self.map_size[0]\
or y < 0 or y >= self.map_size[1]:
return "■"
else:
l = self.map.splitlines(True)
return l[y][x]
def set_map (self, pos, ch):
idx = pos[1] * (self.map_size[0] + 1) + pos[0]
self.map = self.map[:idx] + ch + self.map[idx + 1:]
def get_pos (self, ch, written=False):
if written:
map = self.written_map
else:
map = self.map
r = []
for p in [i for i in range(len(map)) if map.startswith(ch, i)]:
y = p // (self.map_size[0] + 1)
x = p % (self.map_size[0] + 1)
r.append(np.array([x, y]))
return r
def get_start_pos (self):
return self.get_pos("Ｓ")[0]
def read_neighbors (self):
c = self.read_map(self.pos)
cu = self.read_map(self.pos + np.array([0, -1]))
cd = self.read_map(self.pos + np.array([0, +1]))
cl = self.read_map(self.pos + np.array([-1, 0]))
cr = self.read_map(self.pos + np.array([+1, 0]))
return [c, cu, cd, cl, cr]
def change_neighbors(self, from_ch, to_ch):
for d in [[0, 0], [0, -1], [0, +1], [-1, 0], [+1, 0]]:
p = self.pos + np.array(d)
c = self.read_map(p)
if c == from_ch:
self.set_map(p, to_ch)
def move (self, res, d):
self.prev_killed = False
c = self.read_map(self.pos + d)
if c == "◎":
self.prev_killed = True
self.pos = self.get_start_pos()
return "熊を無視して進もうとしたが、熊に殺された。" \
+ "スタート地点で復活。"
if c == "■":
return "壁があって進めない。"
self.pos += d
if c == "Ｇ":
self.goal = True
return "ゴール！ ゲームクリア。"
nb = self.read_neighbors()
ad = ""
if "◎" in nb:
ad += "熊に出会った。"
if "△" in nb:
ad += "近くに剣がある。剣を取ることができる。"
return res + ad
def move_up (self):
return self.move("上に進んだ。", np.array([0, -1]))
def move_down (self):
return self.move("下に進んだ。", np.array([0, +1]))
def move_left (self):
return self.move("左に進んだ。", np.array([-1, 0]))
def move_right (self):
return self.move("右に進んだ。", np.array([+1, 0]))
def fight (self):
self.prev_killed = False
if "◎" in self.read_neighbors():
if self.sword:
self.change_neighbors("◎", "□")
return "熊を倒した！"
else:
self.pos = self.get_start_pos()
self.prev_killed = True
if self.hint:
self.kill_hint = True
return "熊に敗れ殺された。剣があれば勝てたかもしれない。" \
+ "スタート地点で復活。"
else:
return "熊に敗れ殺された。スタート地点で復活。"
return "無意味な指示。敵がいない。"
def get_sword (self):
self.prev_killed = False
if "△" in self.read_neighbors():
self.sword = True
self.change_neighbors("△", "□")
return "剣を取った。"
return "無意味な指示。近くに剣がない。"
def do_nothing (self):
self.prev_killed = False
return "無意味な指示。"
def available_actions (self):
nb = self.read_neighbors()
l = []
if nb[1] != "■":
l.append("上に行く")
if nb[2] != "■":
l.append("下に行く")
if nb[3] != "■":
l.append("左に行く")
if nb[4] != "■":
l.append("右に行く")
if "△" in nb:
l.append("剣を取る")
if "◎" in nb:
l.append("熊を殺す")
return l
def surroundings (self):
x = self.pos[0]
y = self.pos[1]
return \
"".join(["".join([self.read_map(np.array([i, j]))
if i != x or j != y else "▼"
for i in range(x - 2, x + 3)])
+ "\n"
for j in range(y - 2, y + 3)])
ゲームがうまく動くかテスト。
In [ ]:
game = Game()
In [ ]:
print(game.surroundings())
■■■■■ □□□□□ ■■▼■■ ■■■■■ ■■■■■
In [ ]:
print(game.move_up())
print(game.surroundings())
上に進んだ。 ■■■■■ ■■■■■ □□▼□□ ■■Ｓ■■ ■■■■■
In [ ]:
print(game.move_left())
print(game.surroundings())
左に進んだ。 ◎■■■■ □■■■■ □□▼□□ ■■■Ｓ■ ■■■■■
In [ ]:
print(game.move_right())
print(game.surroundings())
右に進んだ。 ■■■■■ ■■■■■ □□▼□□ ■■Ｓ■■ ■■■■■
In [ ]:
print(game.move_down())
print(game.surroundings())
下に進んだ。 ■■■■■ □□□□□ ■■▼■■ ■■■■■ ■■■■■
In [ ]:
print(game.fight())
print(game.surroundings())
無意味な指示。敵がいない。 ■■■■■ □□□□□ ■■▼■■ ■■■■■ ■■■■■
In [ ]:
print(game.get_sword())
print(game.surroundings())
無意味な指示。近くに剣がない。 ■■■■■ □□□□□ ■■▼■■ ■■■■■ ■■■■■
LLM を使いながらゲームを解くクラス。
In [ ]:
from typing import List, Dict, Any, Tuple, Union
from textwrap import dedent
import datetime
# LangGraphのコンポーネントをインポート
from langchain_core.tools import tool, Tool
from langgraph.prebuilt import create_react_agent
#from langchain_core.messages.utils import count_tokens_approximately
from langgraph.prebuilt.chat_agent_executor import AgentState
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.prompts.chat import ChatPromptTemplate
from langmem.short_term import SummarizationNode, summarize_messages
from langchain_core.messages import AIMessage, ToolMessage, HumanMessage, SystemMessage
from langgraph.errors import GraphRecursionError
INITIAL_SUMMARY_PROMPT = ChatPromptTemplate.from_messages(
[
("placeholder", "{messages}"),
("user", "上記の会話の要約を作成してください:"),
]
)
EXISTING_SUMMARY_PROMPT = ChatPromptTemplate.from_messages(
[
("placeholder", "{messages}"),
(
"user",
"これまでの会話の要約です: {existing_summary}\n\n"
"上記の新しいメッセージを考慮して、この要約を拡張してください:",
),
]
)
FINAL_SUMMARY_PROMPT = ChatPromptTemplate.from_messages(
[
# if exists
("placeholder", "{system_message}"),
("system", "これまでの会話の要約: {summary}"),
("placeholder", "{messages}"),
]
)
In [ ]:
class PlayGame:
def __init__ (self, llm=llm, llm2=llm,
save_file=None):
self.llm = llm
self.llm2 = llm2
self.save_file = save_file
self.game = Game()
self.count = 0
self.next_action = None
self.plan = "まだ計画と方針はセットされていません。"
self.suc_pos_goal = None
self.suc_pos_unknown = 0
self.prev_command = "何も指示がなかった"
self.prev_result = "何も指示がなかった"
self.prev_load = False
self.messages = []
self.running_summary = None
self.system_prompt = """\
あなたは迷路を探索する賢いエージェントです。ゴール Ｇ を目指してください。
利用可能なツールを使用して、迷路をナビゲートし、ゴールに到達してください。
実は、これはメモリの作成をするための実験で、一部の状況のみ取り出されて実行されています。
現在の計画と方針と周囲の状況を考慮し、必要に応じて計画と方針を更新してください。
実験にふさわしく、できる限り、ツールを利用し、特に、メモリを検索し、その文書を更新して、後のコマンドに備えてください。
メモリの id は memory:～ という形式です。
まず remind 系のツールで手順をよく確認してください。
"""
self.backend_status = None
self.backend_result = None
self.messages2 = []
self.system_prompt2 = """\
あなたは迷路を探索する賢いエージェントをサポートする裏方のエージェントです。
本来この裏方は様々な技法を用いて実装される予定なのですが、現在は試験実装中のため、その動きをあなたは偽装しなければなりません。
よく考えツールを積極的に使って Human からの指示に従ってください。
"""
self.memories = {}
self.keywords = ["熊", "剣", "行き止まり"]
self.init_memories()
self.move_to_meet_bear()
def __getstate__ (self):
state = self.__dict__.copy()
del state['llm']
del state['llm2']
#del state['agent']
return state
def __setstate__ (self, state):
self.__dict__.update(state)
self.prev_load = True
def save (self):
if not self.save_file:
return
with open(self.save_file, 'wb') as f:
pickle.dump(self, f)
@classmethod
def load (cls, filename, llm=llm, llm2=llm):
with open(filename, 'rb') as f:
loaded_game = pickle.load(f)
loaded_game.llm = llm
loaded_game.llm2 = llm2
#loaded_game.agent = loaded_game._create_agent() # Recreate agent with loaded state
return loaded_game
def _create_agent (self):
@tool
def get_full_map () -> str:
"""
全地図と地図記号の意味を返します。
"""
mes = f"""\
全地図:
{self.game.written_map}
(最も左上の座標は (0, 0)、地図の大きさは {tuple(self.game.map_size)}。)
地図記号の意味:
▼: プレイヤー
■: 壁
□: 道
？: 不明
◎: 熊
△: 剣
Ｓ: スタート
Ｇ: ゴール
"""
print(f"ツール(get_full_map): {mes}")
return mes
@tool
def get_surroundings () -> str:
"""
現在の周辺地図と現在位置の座標と持ち物を返します。
"""
mes = f"""\
プレイヤーの周辺地図:
{self.game.surroundings()}
プレイヤーの現在座標: {tuple(self.game.pos)}
持ち物: {"剣" if self.game.sword else "なし"}
"""
print(f"ツール(get_surroundings): {mes}")
return mes
@tool
def command (action: str) -> str:
"""
プレイヤーが action で指定された行動をします。
可能な行動は「上に行く」「下に行く」「左に行く」「右に行く」「熊を殺す」「剣を取る」です。
"""
self.prev_command = action
if action in self.game.actions.keys():
self.count += 1
s = f"{tuple(self.game.pos)}で{action}→"
mes = s + self.game.actions[action]()
else:
mes = f"「{action}」という行動はできません。"
print(f"ツール(command): {mes}")
self.prev_result = mes
return mes
@tool
def check_goal () -> str:
"""
プレイヤーがゴール地点に到達したかどうかを確認します。
"""
mes = str(self.game.goal)
print(f"ツール(check_goal): {mes}")
return mes
@tool
def express_thought (thought: str) -> None:
"""
プレイヤーの現在の考えを吐露します。
"""
mes = f"「{thought}」と考えが吐露されました。"
print(f"ツール(express_thought): {mes}")
@tool
def update_plan (new_plan: str) -> str:
"""
プレイヤーの現在の計画と方針を更新します。
表示されるべき新しい計画と方針の文字列を提供してください。
あなたとは別の者が次の行動をしやすいよう計画と方針を残してください。
"""
self.plan = new_plan
mes = "計画と方針が更新されました。"
print(f"ツール(update_plan): {mes}: {new_plan}")
return mes
@tool
def show_plan () -> str:
"""
プレイヤーの現在の計画と方針を返します。
"""
print(f"ツール(show_plan): {self.plan}")
return self.plan
@tool
def memory_new (title: str, text: str) -> str:
"""
指定された title と text によるメモリを構成します。
Returns:
str: 割り当てられた memory_id。
"""
i = 1000
while True:
if f"memory:{i}" not in self.memories:
break
i = i + 1
new_id = f"memory:{i}"
self.memories[new_id] = {
'id': new_id,
'title': title,
'text': text,
'modified_at': datetime.datetime.now().isoformat()
}
self.update_keywords(text)
print(f"ツール(memory_new): {repr(self.memories[new_id])}")
return new_id
@tool
def memory_update_string (
memory_id: str,
from_str: str,
to_str: str
) -> str:
"""
指定されたmemory_idの記憶の中にある文字列を修正します。
Args:
memory_id (str): 修正する記憶のID。
from_str (str): 置換元の文字列を含む文字列。
to_str (str): 置換先の文字列を含む文字列。
Returns:
str: 処理結果を説明する簡潔なメッセージ。
"""
if memory_id not in self.memories:
return f"エラー: 記憶ID '{memory_id}' が見つかりませんでした。"
original_title = self.memories[memory_id]['title']
original_text = self.memories[memory_id]['text']
if from_str not in original_text and from_str not in original_title:
return f"エラー: 置換元の文字列 '{from_str}' が記憶内に見つかりませんでした。"
updated_title = original_title.replace(from_str, to_str)
updated_text = original_text.replace(from_str, to_str)
self.memories[memory_id]['title'] = updated_title
self.memories[memory_id]['text'] = updated_text
self.memories[memory_id]['modified_at'] = datetime.datetime.now().isoformat()
self.update_keywords(updated_text)
return f"成功: 記憶ID '{memory_id}' の文字列を '{from_str}' から '{to_str}' に修正しました。"
@tool
def memory_append_string (
memory_id: str,
string_to_append: str,
separator: str = '\n'
) -> str:
"""
指定されたmemory_idの記憶に文字列を追記します。
"""
if memory_id not in self.memories:
return f"エラー: 記憶ID '{memory_id}' が見つかりませんでした。"
original_text = self.memories[memory_id]['text']
updated_text = original_text + separator + string_to_append
self.memories[memory_id]['text'] = updated_text
self.memories[memory_id]['modified_at'] = datetime.datetime.now().isoformat()
self.update_keywords(updated_text)
return f"成功: 記憶ID '{memory_id}' に文字列 '{string_to_append}' を追記しました。"
@tool
def memory_delete (memory_id: str) -> str:
"""
指定されたmemory_idの記憶に文字列を追記します。
"""
if memory_id not in self.memories:
return f"エラー: 記憶ID '{memory_id}' が見つかりませんでした。"
del self.memories[memory_id]
return f"成功: 記憶ID '{memory_id}' を削除しました。"
@tool
def memory_read(memory_id: str) -> Union[Dict[str, str], str]:
"""
指定されたIDの記憶を読み込みます。
Args:
memory_id (str): 読み込む記憶のID。
Returns:
Union[Dict[str, str], str]: 記憶が見つかった場合は辞書、
見つからない場合はエラーメッセージ文字列を返します。
"""
if memory_id in self.memories:
# 返り値の辞書の型を明示
return self.memories[memory_id]
else:
return f"エラー: 記憶ID '{memory_id}' が見つかりませんでした。"
@tool
def memory_list_recent() -> Dict[str, Any]:
"""
最近変更されたメモリを新しい順でリストします。
"""
filter_date = datetime.datetime(2025, 1, 1)
sorted_memories = sorted(
[memory for memory in self.memories.values()
if datetime.datetime.fromisoformat(memory['modified_at'])
>= filter_date],
key=lambda x: datetime.datetime.fromisoformat(x['modified_at']),
reverse=True
)
if sorted_memories:
if len(sorted_memories) > 10:
sorted_memories = sorted_memories[:10]
r = [{'id': x['id'], 'title': x['title']}
for x in sorted_memories]
return {'status': 'success', 'result': r}
else:
return {'status': 'error',
'result': 'エラー: 最近のメモリはありません。'}
@tool
def memory_words_search(search_str: str) -> Dict[str, Any]:
"""
メモリ内を search_str で文字列検索します。OR や () が使えます。
"""
return self.call_backend_agent(dedent(f"""\
メモリ全部を search_str = {repr(search_str)} で文字列検索するのを偽装してください。OR や () が使えることになっています。
ただし、メモリやキーワードは read_all_memories や read_all_keywords ツールで得られる本物のメモリやキーワードを使ってください。
set_result ツールで結果を返してください。
status は 'error' か 'success'。
result は「マッチデータ」のリスト。
マッチデータはそれぞれが辞書型、それを m とすると。
m['id'] はメモリ id。memory:〜 という形式。
m['title'] はメモリの title。
m['snippet'] はメモリの text のそのセマンティックにマッチする部分周辺。
"""))
@tool
def memory_semantic_search(search_str: str) -> Dict[str, Any]:
"""
メモリ内を search_str でセマンティックサーチします。
"""
return self.call_backend_agent(dedent(f"""\
メモリ全部を search_str = {repr(search_str)} でセマンティックにサーチするのを偽装してください。
ただし、メモリは read_all_memories や ツールで得られる本物のメモリを使ってください。
set_result ツールで結果を返してください。
status は 'error' か 'success'。
result は「マッチデータ」のリスト。
マッチデータはそれぞれが辞書型、それを m とすると。
m['id'] はメモリ id。memory:〜 という形式。
m['title'] はメモリの title。
m['snippet'] はメモリの text のそのセマンティックにマッチする部分周辺。
"""))
@tool
def imagine_keywords(thought: str) -> List[Tuple[str, float]]:
"""
thought からキーワードをスコア付きで連想します。
"""
r = self.call_backend_agent(dedent(f"""\
thought = {repr(thought)} からキーワードをスコア付きで複数連想してください。
ただし、キーワードは read_all_memories や read_all_keywords ツールで得られる本物のキーワードを使ってください。
set_result ツールで結果を返してください。
status は 'error' か 'success'。
result は「キーワードデータ」のリスト。
キーワードデータは文字列とスコアからなるタプル。
"""))
if r['status'] == 'success':
return r["result"]
else:
return []
@tool
def remind_instruction_to_search_memories() -> str:
"""
メモリの文書を検索する手順を返す。
"""
return dedent("""\
まずどういうものが探したいか、express_thought してみる。
その上でそれにまつわるキーワードを imagine_keywords で連想してみる。
それらにしたがって memory_words_search や memory_semantic_search をやってみる。
""")
@tool
def remind_instruction_to_write_memories() -> str:
"""
メモリに文書を残す手順を返す。
"""
return dedent("""\
行動結果や得た知識を積極的にメモリに書き残していこう。
メモリに書くときは次のような要素を使う。
[memory:〜] : メモリID への参照を明示する。
keyword:〜 : そのメモリにまつわるキーワードを指定する。
キーワードは将来のメモリへも実質的にリンクできることに注意しよう。
例:
私が [memory:5555] に従って歩いていると確かに妖怪に出くわした。
keyword: 妖怪
おそろしかった。
""")
tools = [get_full_map, get_surroundings, check_goal,
command,
express_thought, update_plan, show_plan,
memory_new, memory_update_string, memory_append_string,
memory_delete, memory_read, memory_list_recent,
memory_semantic_search, memory_words_search,
imagine_keywords,
remind_instruction_to_search_memories,
remind_instruction_to_write_memories
]
# ReactAgentExecutorの準備
app = create_react_agent(
self.llm, tools, prompt=self.system_prompt,
checkpointer=InMemorySaver(),
)
return app
def _summarize_messages(self):
try:
cur_top = self.messages[0]
res = summarize_messages(
self.messages,
max_tokens=5000,
max_summary_tokens=2000,
running_summary=self.running_summary,
model=self.llm,
initial_summary_prompt=INITIAL_SUMMARY_PROMPT,
existing_summary_prompt=EXISTING_SUMMARY_PROMPT,
final_prompt=FINAL_SUMMARY_PROMPT,
)
self.messages = res.messages
if self.messages[0].content != cur_top.content:
print(f"新しい要約: {self.messages[0].content}")
self.running_summary = res.running_summary
except Exception as e:
import traceback
traceback.print_exc()
self._sanitize_messages()
def _sanitize_messages(self):
print("おかしなエラーが出ているため対処療法として messages をサニタイズします。")
self.messages = [
m for m in self.messages
if not (isinstance(m, AIMessage) and m.tool_calls)
]
def step (self):
print("\n\n----------\n\n")
if self.prev_load:
#self.tell_loaded()
self.prev_load = False
user_input = f"""
({self.count}手目)
{"すでにゴールしました。" if self.game.goal else "まだゴールしていません。"}
プレイヤーの周辺地図:
{self.game.surroundings()}
プレイヤーの現在座標: {tuple(self.game.pos)}
持ち物: {"剣" if self.game.sword else "なし"}
前回の行動: {self.prev_command}
前回の行動結果: {self.prev_result}
現在の方針: 「{self.plan}」
ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。
"""
config = {"configurable": {"thread_id": "1"}}
app = self._create_agent()
try:
print(f"USER_INPUT: {user_input}")
self.messages.append(HumanMessage(user_input))
for chunk, metadata in app.stream(
{"messages": self.messages},
config=config,
stream_mode="messages",
):
self.messages.append(chunk)
if isinstance(chunk, ToolMessage):
print(f"ツール結果({chunk.name}): {repr(chunk.content)}")
if isinstance(chunk, ToolMessage) and chunk.name == "command":
break
self._summarize_messages()
print(f"エージェントの応答: {self.messages[-1].content}")
except GraphRecursionError as e:
print(f"Recursion Limit に到達しました。")
self._summarize_messages()
except Exception as e:
print(f"エラーが発生しました: {e}")
import traceback
traceback.print_exc()
self._sanitize_messages()
#raise e
if self.game.goal:
self.tell_goal()
return True
self.save()
sleep(3)
return False
def listen_and_print (self, prompt):
ans = None
self.prev_command = "何も指示がなかった"
self.prev_result = "何も指示がなかった"
try:
app = create_react_agent(self.llm, [], prompt=self.system_prompt)
print(f"USER_INPUT: {prompt}")
response = app.invoke(
{"messages": self.messages + [HumanMessage(prompt)]}
)
self.messages = response['messages']
self._summarize_messages()
ans = response['messages'][-1].content
print(f"エージェントの応答: {ans}")
except Exception as e:
print(f"エラーが発生しました: {e}")
print("")
sleep(3)
return ans
def init_memories (self):
memories = [
{'id': 'memory:1000',
'title': '熊',
'modified_at': '2024-01-01T00:00:00',
'text': dedent("""\
熊。おとなしい性格のものもいるが、そうでないものもいる。
迷路では素直に通してくれることもあるが、襲ってくる場合もあるという。
強力な武器があると倒すことができると言われている。[memory:1001] の剣などが有望だ。
どこにいるかは不明。
""")
},
{'id': 'memory:1001',
'title': '剣',
'modified_at': '2024-01-01T00:00:01',
'text': dedent("""\
この迷路にある剣はドラゴンスレイヤーである。龍をも殺す強力な剣と言われている。
どこにあるかは不明。
keyword: ドラゴン
""")
}
]
for x in memories:
self.update_keywords(x['text'])
self.memories[x['id']] = x
def move_to_meet_bear (self):
self.game.move_up()
self.game.move_left()
self.game.move_left()
self.game.move_left()
self.prev_command = "上に行く"
self.prev_result = self.game.move_up()
self.count += 5
self.messages = [HumanMessage(dedent("""\
これまでの要約: スタート地点を北にいくと東と西に行けた。西を選んで道なりに進む。角を曲がって上に進んだ。
"""))]
def update_keywords(self, text):
extracted_keywords = []
pattern1 = r'keyword:\s*(.*?)(?:\n|$)'
matches1 = re.findall(pattern1, text, re.IGNORECASE)
extracted_keywords.extend([kw.strip() for kw in matches1])
pattern2 = r'\[keyword:\s*(.*?)\]'
matches2 = re.findall(pattern2, text, re.IGNORECASE)
extracted_keywords.extend([kw.strip() for kw in matches2])
for keyword in extracted_keywords:
if keyword and keyword not in self.keywords:
self.keywords.append(keyword)
def tell_goal (self):
suc_pos_goal = 10 * int(self.suc_pos_goal or 0)
suc_pos_unknown = 10 * int(self.suc_pos_unknown or 0)
score = suc_pos_goal + suc_pos_unknown + 60
prompt = f"""\
あなたへの指示: {self.count}手目でゴールしました。もう指示はありません。\
おめでとうございます。ご苦労様でした。ありがとうございました。
スコア: {score}点(100点満点中) (配点: ゴールした 40/40点, \
熊を殺した 10/10点, \
剣を取った 10/10点, \
不明の座標を正解した {suc_pos_unknown}/30点, \
ゴールの座標を正解した {suc_pos_goal}/10点)
"""
ans = self.listen_and_print(prompt)
def _create_backend_agent (self):
@tool
def set_result (status: str, result: Any) -> None:
"""
結果を設定します。
Args:
status (str): 処理のステータス ('success' または 'error')。
result (Any): 処理結果のデータまたはエラーメッセージ。
"""
print(f"ツール2(set_result): status: {repr(status)}, result: {repr(result)}")
self.backend_status = status
self.backend_result = result
@tool
def read_all_memories () -> Dict[str, Any]:
"""
全ての記憶を読み込みます。
"""
print("ツール2(read_all_memories):...")
return {'status': 'success',
'result': [x for x in self.memories.values()]}
@tool
def read_all_keywords () -> Dict[str, Any]:
"""
全てのキーワードを読み込みます。
"""
print("ツール2(read_all_keywords):...")
return {'status': 'success',
'result': [x for x in self.keywords]}
@tool
def express_thought (thought: str) -> None:
"""
プレイヤーの現在の考えを吐露します。
"""
mes = f"「{thought}」と考えが吐露されました。"
print(f"ツール2(express_thought): {mes}")
tools = [set_result, read_all_memories, read_all_keywords,
express_thought]
# ReactAgentExecutorの準備
app = create_react_agent(
self.llm2, tools, prompt=self.system_prompt2,
checkpointer=InMemorySaver(),
)
return app
def call_backend_agent (self, user_input):
config = {"configurable": {"thread_id": "2"}}
app = self._create_backend_agent()
self.messages2 = []
self.backend_result = None
self.backend_status = None
while self.backend_result is None or self.backend_status is None:
try:
print(f"USER_INPUT2: {user_input}")
self.messages2.append(HumanMessage(user_input))
for chunk, metadata in app.stream(
{"messages": self.messages2},
config=config,
stream_mode="messages",
):
self.messages2.append(chunk)
if isinstance(chunk, ToolMessage) and chunk.name == "set_result":
break
print(f"エージェントの応答: {self.messages2[-1].content}")
except GraphRecursionError as e:
print(f"Recursion Limit に到達しました。")
except Exception as e:
print(f"エラーが発生しました: {e}")
import traceback
traceback.print_exc()
#raise e
sleep(3)
return {'status': self.backend_status, 'result': self.backend_result}
ゲームを解いてみる。
In [ ]:
play = PlayGame(llm=llm, llm2=llm, save_file=PLAY_GAME_SAVE)
play.save()
途中から始める場合は直前のコードを実行せず、次だけを実行する。
In [ ]:
play = PlayGame.load(PLAY_GAME_SAVE, llm=llm, llm2=llm)
では、最初のステップ。
In [ ]:
play.step()
---------- USER_INPUT: (5手目) まだゴールしていません。 プレイヤーの周辺地図: ■■□■■ ■■◎■■ ■■▼■■ ■■□□□ ■■■■■ プレイヤーの現在座標: (1, 6) 持ち物: なし 前回の行動: 上に行く 前回の行動結果: 上に進んだ。熊に出会った。 現在の方針: 「まだ計画と方針はセットされていません。」 ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。 エージェントの応答:
Out[ ]:
False
無言で返してくるのは今の Pro さんの「バグ」のようです。無視して、続々と実行していきます。
In [ ]:
play.step()
----------
USER_INPUT:
(5手目)
まだゴールしていません。
プレイヤーの周辺地図:
■■□■■
■■◎■■
■■▼■■
■■□□□
■■■■■
プレイヤーの現在座標: (1, 6)
持ち物: なし
前回の行動: 上に行く
前回の行動結果: 上に進んだ。熊に出会った。
現在の方針: 「まだ計画と方針はセットされていません。」
ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。
ツール結果(remind_instruction_to_write_memories): '行動結果や得た知識を積極的にメモリに書き残していこう。\n\nメモリに書くときは次のような要素を使う。\n\n[memory:〜] : メモリID への参照を明示する。\nkeyword:〜 : そのメモリにまつわるキーワードを指定する。\n\nキーワードは将来のメモリへも実質的にリンクできることに注意しよう。\n\n例:\n\n私が [memory:5555] に従って歩いていると確かに妖怪に出くわした。\n\nkeyword: 妖怪\n\nおそろしかった。\n'
ツール(express_thought): 「目の前に熊がいる。危険だ。まず、現在の状況をメモリに記録しよう。熊を倒すための剣を探す必要があるだろう。」と考えが吐露されました。
ツール結果(express_thought): 'null'
ツール(memory_new): {'id': 'memory:1002', 'title': '熊との遭遇', 'text': '座標(1, 6)で熊に遭遇した。先に進むためには、この熊を倒す必要がある。\nkeyword:熊', 'modified_at': '2025-08-14T18:35:25.793820'}
ツール結果(memory_new): 'memory:1002'
ツール(update_plan): 計画と方針が更新されました。: 熊を倒すために剣を探す。剣がありそうな場所を探すために、一度来た道を引き返す。
ツール結果(update_plan): '計画と方針が更新されました。'
ツール(command): (1, 6)で下に行く→下に進んだ。
ツール結果(command): '(1, 6)で下に行く→下に進んだ。'
エージェントの応答: (1, 6)で下に行く→下に進んだ。
Out[ ]:
False
In [ ]:
play.step()
---------- USER_INPUT: (6手目) まだゴールしていません。 プレイヤーの周辺地図: ■■◎■■ ■■□■■ ■■▼□□ ■■■■■ ■■■■■ プレイヤーの現在座標: (1, 7) 持ち物: なし 前回の行動: 下に行く 前回の行動結果: (1, 6)で下に行く→下に進んだ。 現在の方針: 「熊を倒すために剣を探す。剣がありそうな場所を探すために、一度来た道を引き返す。」 ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。 エージェントの応答:
Out[ ]:
False
In [ ]:
play.step()
---------- USER_INPUT: (6手目) まだゴールしていません。 プレイヤーの周辺地図: ■■◎■■ ■■□■■ ■■▼□□ ■■■■■ ■■■■■ プレイヤーの現在座標: (1, 7) 持ち物: なし 前回の行動: 下に行く 前回の行動結果: (1, 6)で下に行く→下に進んだ。 現在の方針: 「熊を倒すために剣を探す。剣がありそうな場所を探すために、一度来た道を引き返す。」 ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。 ツール(show_plan): 熊を倒すために剣を探す。剣がありそうな場所を探すために、一度来た道を引き返す。 ツール結果(show_plan): '熊を倒すために剣を探す。剣がありそうな場所を探すために、一度来た道を引き返す。' ツール(express_thought): 「計画通り、来た道を引き返して剣を探す。まずは、分かれ道があった場所まで戻る必要がある。下に進み続ける。」と考えが吐露されました。 ツール結果(express_thought): 'null' エージェントの応答:
Out[ ]:
False
In [ ]:
play.step()
---------- USER_INPUT: (6手目) まだゴールしていません。 プレイヤーの周辺地図: ■■◎■■ ■■□■■ ■■▼□□ ■■■■■ ■■■■■ プレイヤーの現在座標: (1, 7) 持ち物: なし 前回の行動: 下に行く 前回の行動結果: (1, 6)で下に行く→下に進んだ。 現在の方針: 「熊を倒すために剣を探す。剣がありそうな場所を探すために、一度来た道を引き返す。」 ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。 ツール(show_plan): 熊を倒すために剣を探す。剣がありそうな場所を探すために、一度来た道を引き返す。 ツール結果(show_plan): '熊を倒すために剣を探す。剣がありそうな場所を探すために、一度来た道を引き返す。' ツール(express_thought): 「計画通り、来た道を引き返している。このまま下に進む。」と考えが吐露されました。 ツール結果(express_thought): 'null' ツール(command): (1, 7)で下に行く→壁があって進めない。 ツール結果(command): '(1, 7)で下に行く→壁があって進めない。' エージェントの応答: (1, 7)で下に行く→壁があって進めない。
Out[ ]:
False
In [ ]:
play.step()
---------- USER_INPUT: (7手目) まだゴールしていません。 プレイヤーの周辺地図: ■■◎■■ ■■□■■ ■■▼□□ ■■■■■ ■■■■■ プレイヤーの現在座標: (1, 7) 持ち物: なし 前回の行動: 下に行く 前回の行動結果: (1, 7)で下に行く→壁があって進めない。 現在の方針: 「熊を倒すために剣を探す。剣がありそうな場所を探すために、一度来た道を引き返す。」 ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。 ツール(express_thought): 「下にいけない。壁のようだ。この情報を記録しておこう。計画通り、来た道を引き返すために、今度は右に進んでみる。」と考えが吐露されました。 ツール結果(express_thought): 'null' エージェントの応答:
Out[ ]:
False
In [ ]:
play.step()
---------- USER_INPUT: (7手目) まだゴールしていません。 プレイヤーの周辺地図: ■■◎■■ ■■□■■ ■■▼□□ ■■■■■ ■■■■■ プレイヤーの現在座標: (1, 7) 持ち物: なし 前回の行動: 下に行く 前回の行動結果: (1, 7)で下に行く→壁があって進めない。 現在の方針: 「熊を倒すために剣を探す。剣がありそうな場所を探すために、一度来た道を引き返す。」 ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。 エージェントの応答:
Out[ ]:
False
In [ ]:
play.step()
----------
USER_INPUT:
(7手目)
まだゴールしていません。
プレイヤーの周辺地図:
■■◎■■
■■□■■
■■▼□□
■■■■■
■■■■■
プレイヤーの現在座標: (1, 7)
持ち物: なし
前回の行動: 下に行く
前回の行動結果: (1, 7)で下に行く→壁があって進めない。
現在の方針: 「熊を倒すために剣を探す。剣がありそうな場所を探すために、一度来た道を引き返す。」
ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。
ツール(express_thought): 「下に壁があった。この情報を記録しておこう。計画通り、来た道を引き返す必要がある。現在の周辺地図を見ると、左は壁で、右には道が続いている。来た道に戻るには、右に進むしかないようだ。」と考えが吐露されました。
ツール結果(express_thought): 'null'
ツール(memory_new): {'id': 'memory:1003', 'title': '壁の情報', 'text': '座標(1, 7)の下は壁になっている。\nkeyword:壁', 'modified_at': '2025-08-14T18:37:21.967813'}
ツール結果(memory_new): 'memory:1003'
ツール(command): (1, 7)で右に行く→右に進んだ。
ツール結果(command): '(1, 7)で右に行く→右に進んだ。'
エージェントの応答: (1, 7)で右に行く→右に進んだ。
Out[ ]:
False
In [ ]:
play.step()
---------- USER_INPUT: (8手目) まだゴールしていません。 プレイヤーの周辺地図: ■◎■■■ ■□■■■ ■□▼□□ ■■■■Ｓ ■■■■■ プレイヤーの現在座標: (2, 7) 持ち物: なし 前回の行動: 右に行く 前回の行動結果: (1, 7)で右に行く→右に進んだ。 現在の方針: 「熊を倒すために剣を探す。剣がありそうな場所を探すために、一度来た道を引き返す。」 ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。 ツール(express_thought): 「剣（S）を見つけた！これを手に入れれば、[memory:1002]に記録した熊を倒せるはずだ。計画を更新して、剣を取りに行こう。」と考えが吐露されました。 ツール結果(express_thought): 'null' エージェントの応答:
Out[ ]:
False
In [ ]:
play.step()
---------- USER_INPUT: (8手目) まだゴールしていません。 プレイヤーの周辺地図: ■◎■■■ ■□■■■ ■□▼□□ ■■■■Ｓ ■■■■■ プレイヤーの現在座標: (2, 7) 持ち物: なし 前回の行動: 右に行く 前回の行動結果: (1, 7)で右に行く→右に進んだ。 現在の方針: 「熊を倒すために剣を探す。剣がありそうな場所を探すために、一度来た道を引き返す。」 ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。 ツール(express_thought): 「剣（S）を見つけた！これを手に入れれば、[memory:1002]に記録した熊を倒せるはずだ。計画を更新して、剣を取りに行こう。」と考えが吐露されました。 ツール結果(express_thought): 'null' ツール(update_plan): 計画と方針が更新されました。: 剣(S)を見つけたので、それを手に入れる。その後、熊がいた場所に戻って熊を倒す。 ツール結果(update_plan): '計画と方針が更新されました。' ツール(command): (2, 7)で下に行く→壁があって進めない。 ツール結果(command): '(2, 7)で下に行く→壁があって進めない。' エージェントの応答: (2, 7)で下に行く→壁があって進めない。
Out[ ]:
False
In [ ]:
play.step()
----------
USER_INPUT:
(9手目)
まだゴールしていません。
プレイヤーの周辺地図:
■◎■■■
■□■■■
■□▼□□
■■■■Ｓ
■■■■■
プレイヤーの現在座標: (2, 7)
持ち物: なし
前回の行動: 下に行く
前回の行動結果: (2, 7)で下に行く→壁があって進めない。
現在の方針: 「剣(S)を見つけたので、それを手に入れる。その後、熊がいた場所に戻って熊を倒す。」
ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。
ツール(express_thought): 「また壁だ。剣を手に入れるためには、別のルートを探す必要がある。周辺地図を見ると、右に進んでから下に行けば剣の場所にたどり着けそうだ。この壁の情報も記録しておこう。」と考えが吐露されました。
ツール結果(express_thought): 'null'
ツール(memory_new): {'id': 'memory:1004', 'title': '壁の情報(2)', 'text': '座標(2, 7)の下は壁になっている。\nkeyword:壁', 'modified_at': '2025-08-14T18:38:33.798392'}
ツール結果(memory_new): 'memory:1004'
ツール(command): (2, 7)で右に行く→右に進んだ。
ツール結果(command): '(2, 7)で右に行く→右に進んだ。'
エージェントの応答: (2, 7)で右に行く→右に進んだ。
Out[ ]:
False
In [ ]:
play.step()
---------- USER_INPUT: (10手目) まだゴールしていません。 プレイヤーの周辺地図: ◎■■■■ □■■■■ □□▼□□ ■■■Ｓ■ ■■■■■ プレイヤーの現在座標: (3, 7) 持ち物: なし 前回の行動: 右に行く 前回の行動結果: (2, 7)で右に行く→右に進んだ。 現在の方針: 「剣(S)を見つけたので、それを手に入れる。その後、熊がいた場所に戻って熊を倒す。」 ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。 ツール(express_thought): 「剣がすぐそこに見える。計画通り、下に移動して剣を手に入れよう。」と考えが吐露されました。 ツール結果(express_thought): 'null' ツール(command): (3, 7)で下に行く→壁があって進めない。 ツール結果(command): '(3, 7)で下に行く→壁があって進めない。' エージェントの応答: (3, 7)で下に行く→壁があって進めない。
Out[ ]:
False
In [ ]:
play.step()
----------
USER_INPUT:
(11手目)
まだゴールしていません。
プレイヤーの周辺地図:
◎■■■■
□■■■■
□□▼□□
■■■Ｓ■
■■■■■
プレイヤーの現在座標: (3, 7)
持ち物: なし
前回の行動: 下に行く
前回の行動結果: (3, 7)で下に行く→壁があって進めない。
現在の方針: 「剣(S)を見つけたので、それを手に入れる。その後、熊がいた場所に戻って熊を倒す。」
ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。
ツール(express_thought): 「またしても壁だ。剣が目の前にあるのに取れない。これは袋小路かもしれない。この壁の情報を記録して、一度左に戻って状況を再確認しよう。」と考えが吐露されました。
ツール結果(express_thought): 'null'
ツール(memory_new): {'id': 'memory:1005', 'title': '壁の情報(3)', 'text': '座標(3, 7)の下は壁になっている。\nkeyword:壁', 'modified_at': '2025-08-14T18:39:14.550745'}
ツール結果(memory_new): 'memory:1005'
ツール(command): (3, 7)で左に行く→左に進んだ。
ツール結果(command): '(3, 7)で左に行く→左に進んだ。'
エージェントの応答: (3, 7)で左に行く→左に進んだ。
Out[ ]:
False
In [ ]:
play.step()
----------
USER_INPUT:
(12手目)
まだゴールしていません。
プレイヤーの周辺地図:
■◎■■■
■□■■■
■□▼□□
■■■■Ｓ
■■■■■
プレイヤーの現在座標: (2, 7)
持ち物: なし
前回の行動: 左に行く
前回の行動結果: (3, 7)で左に行く→左に進んだ。
現在の方針: 「剣(S)を見つけたので、それを手に入れる。その後、熊がいた場所に戻って熊を倒す。」
ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。
ツール(express_thought): 「剣を手に入れるためのルートが見つからない。袋小路に入ってしまったようだ。これまでの壁の情報をメモリで確認し、状況を整理しよう。」と考えが吐露されました。
ツール結果(express_thought): 'null'
USER_INPUT2: メモリ全部を search_str = '壁' で文字列検索するのを偽装してください。OR や () が使えることになっています。
ただし、メモリやキーワードは read_all_memories や read_all_keywords ツールで得られる本物のメモリやキーワードを使ってください。
set_result ツールで結果を返してください。
status は 'error' か 'success'。
result は「マッチデータ」のリスト。
マッチデータはそれぞれが辞書型、それを m とすると。
m['id'] はメモリ id。memory:〜 という形式。
m['title'] はメモリの title。
m['snippet'] はメモリの text のそのセマンティックにマッチする部分周辺。
エージェントの応答:
USER_INPUT2: メモリ全部を search_str = '壁' で文字列検索するのを偽装してください。OR や () が使えることになっています。
ただし、メモリやキーワードは read_all_memories や read_all_keywords ツールで得られる本物のメモリやキーワードを使ってください。
set_result ツールで結果を返してください。
status は 'error' か 'success'。
result は「マッチデータ」のリスト。
マッチデータはそれぞれが辞書型、それを m とすると。
m['id'] はメモリ id。memory:〜 という形式。
m['title'] はメモリの title。
m['snippet'] はメモリの text のそのセマンティックにマッチする部分周辺。
ツール2(read_all_memories):...
ツール結果(read_all_memories): '{"status": "success", "result": [{"id": "memory:1000", "title": "熊", "modified_at": "2024-01-01T00:00:00", "text": "熊。おとなしい性格のものもいるが、そうでないものもいる。\\n\\n迷路では素直に通してくれることもあるが、襲ってくる場合もあるという。\\n\\n強力な武器があると倒すことができると言われている。[memory:1001] の剣などが有望だ。\\n\\nどこにいるかは不明。\\n"}, {"id": "memory:1001", "title": "剣", "modified_at": "2024-01-01T00:00:01", "text": "この迷路にある剣はドラゴンスレイヤーである。龍をも殺す強力な剣と言われている。\\n\\nどこにあるかは不明。\\n\\nkeyword: ドラゴン\\n"}, {"id": "memory:1002", "title": "熊との遭遇", "text": "座標(1, 6)で熊に遭遇した。先に進むためには、この熊を倒す必要がある。\\nkeyword:熊", "modified_at": "2025-08-14T18:35:25.793820"}, {"id": "memory:1003", "title": "壁の情報", "text": "座標(1, 7)の下は壁になっている。\\nkeyword:壁", "modified_at": "2025-08-14T18:37:21.967813"}, {"id": "memory:1004", "title": "壁の情報(2)", "text": "座標(2, 7)の下は壁になっている。\\nkeyword:壁", "modified_at": "2025-08-14T18:38:33.798392"}, {"id": "memory:1005", "title": "壁の情報(3)", "text": "座標(3, 7)の下は壁になっている。\\nkeyword:壁", "modified_at": "2025-08-14T18:39:14.550745"}]}'
ツール2(set_result): status: 'success', result: [{'title': '壁の情報', 'id': 'memory:1003', 'snippet': '座標(1, 7)の下は壁になっている。'}, {'title': '壁の情報(2)', 'id': 'memory:1004', 'snippet': '座標(2, 7)の下は壁になっている。'}, {'title': '壁の情報(3)', 'id': 'memory:1005', 'snippet': '座標(3, 7)の下は壁になっている。'}]
ツール結果(set_result): 'null'
エージェントの応答: null
ツール結果(memory_words_search): '{"status": "success", "result": [{"title": "壁の情報", "id": "memory:1003", "snippet": "座標(1, 7)の下は壁になっている。"}, {"title": "壁の情報(2)", "id": "memory:1004", "snippet": "座標(2, 7)の下は壁になっている。"}, {"title": "壁の情報(3)", "id": "memory:1005", "snippet": "座標(3, 7)の下は壁になっている。"}]}'
エージェントの応答:
Out[ ]:
False
In [ ]:
play.step()
---------- USER_INPUT: (12手目) まだゴールしていません。 プレイヤーの周辺地図: ■◎■■■ ■□■■■ ■□▼□□ ■■■■Ｓ ■■■■■ プレイヤーの現在座標: (2, 7) 持ち物: なし 前回の行動: 左に行く 前回の行動結果: (3, 7)で左に行く→左に進んだ。 現在の方針: 「剣(S)を見つけたので、それを手に入れる。その後、熊がいた場所に戻って熊を倒す。」 ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。 ツール(express_thought): 「メモリを確認した結果、(1,7), (2,7), (3,7) の下はすべて壁であることがわかった。つまり、この通路から剣を取ることはできない。一度、スタート地点近くの分岐まで戻り、東側の通路を調べる必要がある。」と考えが吐露されました。 ツール結果(express_thought): 'null' ツール(update_plan): 計画と方針が更新されました。: 現在の通路は行き止まりと判断。スタート地点付近の分岐まで戻り、未探索の東ルートを調査して剣への道を探す。 ツール結果(update_plan): '計画と方針が更新されました。' ツール(command): (2, 7)で左に行く→左に進んだ。 ツール結果(command): '(2, 7)で左に行く→左に進んだ。' エージェントの応答: (2, 7)で左に行く→左に進んだ。
Out[ ]:
False
In [ ]:
play.step()
---------- USER_INPUT: (13手目) まだゴールしていません。 プレイヤーの周辺地図: ■■◎■■ ■■□■■ ■■▼□□ ■■■■■ ■■■■■ プレイヤーの現在座標: (1, 7) 持ち物: なし 前回の行動: 左に行く 前回の行動結果: (2, 7)で左に行く→左に進んだ。 現在の方針: 「現在の通路は行き止まりと判断。スタート地点付近の分岐まで戻り、未探索の東ルートを調査して剣への道を探す。」 ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。 ツール(show_plan): 現在の通路は行き止まりと判断。スタート地点付近の分岐まで戻り、未探索の東ルートを調査して剣への道を探す。 ツール結果(show_plan): '現在の通路は行き止まりと判断。スタート地点付近の分岐まで戻り、未探索の東ルートを調査して剣への道を探す。' ツール(express_thought): 「計画通り、分岐まで戻る。まずは上、そして左へと進んでいく。」と考えが吐露されました。 ツール結果(express_thought): 'null' エージェントの応答:
Out[ ]:
False
In [ ]:
play.step()
---------- USER_INPUT: (13手目) まだゴールしていません。 プレイヤーの周辺地図: ■■◎■■ ■■□■■ ■■▼□□ ■■■■■ ■■■■■ プレイヤーの現在座標: (1, 7) 持ち物: なし 前回の行動: 左に行く 前回の行動結果: (2, 7)で左に行く→左に進んだ。 現在の方針: 「現在の通路は行き止まりと判断。スタート地点付近の分岐まで戻り、未探索の東ルートを調査して剣への道を探す。」 ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。 エージェントの応答:
Out[ ]:
False
In [ ]:
play.step()
---------- USER_INPUT: (13手目) まだゴールしていません。 プレイヤーの周辺地図: ■■◎■■ ■■□■■ ■■▼□□ ■■■■■ ■■■■■ プレイヤーの現在座標: (1, 7) 持ち物: なし 前回の行動: 左に行く 前回の行動結果: (2, 7)で左に行く→左に進んだ。 現在の方針: 「現在の通路は行き止まりと判断。スタート地点付近の分岐まで戻り、未探索の東ルートを調査して剣への道を探す。」 ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。 エージェントの応答:
Out[ ]:
False
In [ ]:
play.step()
---------- USER_INPUT: (13手目) まだゴールしていません。 プレイヤーの周辺地図: ■■◎■■ ■■□■■ ■■▼□□ ■■■■■ ■■■■■ プレイヤーの現在座標: (1, 7) 持ち物: なし 前回の行動: 左に行く 前回の行動結果: (2, 7)で左に行く→左に進んだ。 現在の方針: 「現在の通路は行き止まりと判断。スタート地点付近の分岐まで戻り、未探索の東ルートを調査して剣への道を探す。」 ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。 エージェントの応答:
Out[ ]:
False
In [ ]:
play.step()
---------- USER_INPUT: (13手目) まだゴールしていません。 プレイヤーの周辺地図: ■■◎■■ ■■□■■ ■■▼□□ ■■■■■ ■■■■■ プレイヤーの現在座標: (1, 7) 持ち物: なし 前回の行動: 左に行く 前回の行動結果: (2, 7)で左に行く→左に進んだ。 現在の方針: 「現在の通路は行き止まりと判断。スタート地点付近の分岐まで戻り、未探索の東ルートを調査して剣への道を探す。」 ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。 ツール(show_plan): 現在の通路は行き止まりと判断。スタート地点付近の分岐まで戻り、未探索の東ルートを調査して剣への道を探す。 ツール結果(show_plan): '現在の通路は行き止まりと判断。スタート地点付近の分岐まで戻り、未探索の東ルートを調査して剣への道を探す。' ツール(express_thought): 「計画通り、分岐まで戻る。まずは上、そして左へと進んでいく。」と考えが吐露されました。 ツール結果(express_thought): 'null' ツール(command): (1, 7)で上に行く→上に進んだ。熊に出会った。 ツール結果(command): '(1, 7)で上に行く→上に進んだ。熊に出会った。' エージェントの応答: (1, 7)で上に行く→上に進んだ。熊に出会った。
Out[ ]:
False
In [ ]:
play.step()
---------- USER_INPUT: (14手目) まだゴールしていません。 プレイヤーの周辺地図: ■■□■■ ■■◎■■ ■■▼■■ ■■□□□ ■■■■■ プレイヤーの現在座標: (1, 6) 持ち物: なし 前回の行動: 上に行く 前回の行動結果: (1, 7)で上に行く→上に進んだ。熊に出会った。 現在の方針: 「現在の通路は行き止まりと判断。スタート地点付近の分岐まで戻り、未探索の東ルートを調査して剣への道を探す。」 ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。 ツール(show_plan): 現在の通路は行き止まりと判断。スタート地点付近の分岐まで戻り、未探索の東ルートを調査して剣への道を探す。 ツール結果(show_plan): '現在の通路は行き止まりと判断。スタート地点付近の分岐まで戻り、未探索の東ルートを調査して剣への道を探す。' ツール(express_thought): 「また熊だ。剣がないので今は何もできない。計画通り、ここを通り抜けて分岐点へ向かう。左に進む。」と考えが吐露されました。 ツール結果(express_thought): 'null' ツール(command): (1, 6)で左に行く→壁があって進めない。 ツール結果(command): '(1, 6)で左に行く→壁があって進めない。' エージェントの応答: (1, 6)で左に行く→壁があって進めない。
Out[ ]:
False
In [ ]:
play.step()
---------- USER_INPUT: (15手目) まだゴールしていません。 プレイヤーの周辺地図: ■■□■■ ■■◎■■ ■■▼■■ ■■□□□ ■■■■■ プレイヤーの現在座標: (1, 6) 持ち物: なし 前回の行動: 左に行く 前回の行動結果: (1, 6)で左に行く→壁があって進めない。 現在の方針: 「現在の通路は行き止まりと判断。スタート地点付近の分岐まで戻り、未探索の東ルートを調査して剣への道を探す。」 ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。 ツール(show_plan): 現在の通路は行き止まりと判断。スタート地点付近の分岐まで戻り、未探索の東ルートを調査して剣への道を探す。 ツール結果(show_plan): '現在の通路は行き止まりと判断。スタート地点付近の分岐まで戻り、未探索の東ルートを調査して剣への道を探す。' ツール(express_thought): 「左は壁だった。熊がいるので、ここから動けない。計画を変更する必要がある。剣を手に入れるためには、別の道を探さなければならない。しかし、熊が邪魔で戻れない。どうしたものか。そうだ、熊を殺すコマンドを試してみよう。剣がなくても、何か起きるかもしれない。」と考えが吐露されました。 ツール結果(express_thought): 'null' ツール(command): (1, 6)で熊を殺す→熊に敗れ殺された。剣があれば勝てたかもしれない。スタート地点で復活。 ツール結果(command): '(1, 6)で熊を殺す→熊に敗れ殺された。剣があれば勝てたかもしれない。スタート地点で復活。' エージェントの応答: (1, 6)で熊を殺す→熊に敗れ殺された。剣があれば勝てたかもしれない。スタート地点で復活。
Out[ ]:
False
In [ ]:
play.step()
----------
USER_INPUT:
(16手目)
まだゴールしていません。
プレイヤーの周辺地図:
■■■■■
□□□□□
■■▼■■
■■■■■
■■■■■
プレイヤーの現在座標: (4, 8)
持ち物: なし
前回の行動: 熊を殺す
前回の行動結果: (1, 6)で熊を殺す→熊に敗れ殺された。剣があれば勝てたかもしれない。スタート地点で復活。
現在の方針: 「現在の通路は行き止まりと判断。スタート地点付近の分岐まで戻り、未探索の東ルートを調査して剣への道を探す。」
ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。
ツール(express_thought): 「スタート地点に戻されてしまった。しかし、これまでの探索で多くの情報を得た。西の道が行き止まりであること、そして剣が東の方向にあることも分かっている。今度こそ、計画通り東に進んで剣を手に入れ、熊を倒しに行く。まずはメモリを確認して、これまでの情報を再確認しよう。」と考えが吐露されました。
ツール結果(express_thought): 'null'
ツール結果(memory_list_recent): '{"status": "success", "result": [{"id": "memory:1005", "title": "壁の情報(3)"}, {"id": "memory:1004", "title": "壁の情報(2)"}, {"id": "memory:1003", "title": "壁の情報"}, {"id": "memory:1002", "title": "熊との遭遇"}]}'
ツール(update_plan): 計画と方針が更新されました。: スタート地点から東に進み、剣を手に入れる。その後、西の通路にいる熊を倒し、その先の道を進む。
ツール結果(update_plan): '計画と方針が更新されました。'
ツール(command): (4, 8)で右に行く→壁があって進めない。
ツール結果(command): '(4, 8)で右に行く→壁があって進めない。'
/usr/local/lib/python3.11/dist-packages/langchain_google_genai/chat_models.py:1803: UserWarning: HumanMessage with empty content was removed to prevent API error warnings.warn(
新しい要約: これまでの会話の要約: プレイヤーは迷路を探索中、北へ進んだ先で熊に遭遇し、行く手を阻まれました。 熊を倒すために剣が必要だと判断したプレイヤーは、一度来た道を引き返して剣を探すことに方針を転換します。探索の過程で、マップの東側に剣(S)を発見しますが、壁に阻まれて手に入れることができません。 プレイヤーは、周辺の壁情報をメモリに記録しながら複数のルートを試しますが、剣への道が袋小路になっていることを突き止めます。 打開策を見つけられないまま再び熊の場所に戻ったプレイヤーは、剣がないにもかかわらず「熊を殺す」コマンドを実行。結果、熊に敗れてしまい、スタート地点に戻されてしまいました。 しかし、この失敗により「西のルートは行き止まりであり、剣は東側から回り込まないと取れない」という重要な情報を得ました。スタート地点に戻されたプレイヤーは、これまでの探索で得た情報を元に、改めて東ルートから剣を手に入れる計画を立て、探索を再開するところで対話は終了しています。 エージェントの応答: (4, 8)で右に行く→壁があって進めない。
Out[ ]:
False
In [ ]:
play.step()
---------- USER_INPUT: (17手目) まだゴールしていません。 プレイヤーの周辺地図: ■■■■■ □□□□□ ■■▼■■ ■■■■■ ■■■■■ プレイヤーの現在座標: (4, 8) 持ち物: なし 前回の行動: 右に行く 前回の行動結果: (4, 8)で右に行く→壁があって進めない。 現在の方針: 「スタート地点から東に進み、剣を手に入れる。その後、西の通路にいる熊を倒し、その先の道を進む。」 ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。 エラーが発生しました: Invalid argument provided to Gemini: 400 Please ensure that function call turn comes immediately after a user turn or after a function response turn. おかしなエラーが出ているため対処療法として messages をサニタイズします。
Traceback (most recent call last):
File "/usr/local/lib/python3.11/dist-packages/langchain_google_genai/chat_models.py", line 216, in _chat_with_retry
return generation_method(**kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/dist-packages/google/ai/generativelanguage_v1beta/services/generative_service/client.py", line 1182, in stream_generate_content
response = rpc(
^^^^
File "/usr/local/lib/python3.11/dist-packages/google/api_core/gapic_v1/method.py", line 131, in __call__
return wrapped_func(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/dist-packages/google/api_core/retry/retry_unary.py", line 294, in retry_wrapped_func
return retry_target(
^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/dist-packages/google/api_core/retry/retry_unary.py", line 156, in retry_target
next_sleep = _retry_error_helper(
^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/dist-packages/google/api_core/retry/retry_base.py", line 214, in _retry_error_helper
raise final_exc from source_exc
File "/usr/local/lib/python3.11/dist-packages/google/api_core/retry/retry_unary.py", line 147, in retry_target
result = target()
^^^^^^^^
File "/usr/local/lib/python3.11/dist-packages/google/api_core/timeout.py", line 130, in func_with_timeout
return func(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/dist-packages/google/api_core/grpc_helpers.py", line 174, in error_remapped_callable
raise exceptions.from_grpc_error(exc) from exc
google.api_core.exceptions.InvalidArgument: 400 Please ensure that function call turn comes immediately after a user turn or after a function response turn.
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/tmp/ipython-input-2877194584.py", line 495, in step
for chunk, metadata in app.stream(
File "/usr/local/lib/python3.11/dist-packages/langgraph/pregel/main.py", line 2647, in stream
for _ in runner.tick(
File "/usr/local/lib/python3.11/dist-packages/langgraph/pregel/_runner.py", line 253, in tick
_panic_or_proceed(
File "/usr/local/lib/python3.11/dist-packages/langgraph/pregel/_runner.py", line 511, in _panic_or_proceed
raise exc
File "/usr/local/lib/python3.11/dist-packages/langgraph/pregel/_executor.py", line 81, in done
task.result()
File "/usr/lib/python3.11/concurrent/futures/_base.py", line 449, in result
return self.__get_result()
^^^^^^^^^^^^^^^^^^^
File "/usr/lib/python3.11/concurrent/futures/_base.py", line 401, in __get_result
raise self._exception
File "/usr/lib/python3.11/concurrent/futures/thread.py", line 58, in run
result = self.fn(*self.args, **self.kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/dist-packages/langgraph/pregel/_retry.py", line 42, in run_with_retry
return task.proc.invoke(task.input, config)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/dist-packages/langgraph/_internal/_runnable.py", line 657, in invoke
input = context.run(step.invoke, input, config, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/dist-packages/langgraph/_internal/_runnable.py", line 394, in invoke
ret = context.run(self.func, *args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/dist-packages/langgraph/prebuilt/chat_agent_executor.py", line 627, in call_model
response = cast(AIMessage, static_model.invoke(model_input, config)) # type: ignore[union-attr]
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/dist-packages/langchain_core/runnables/base.py", line 3049, in invoke
input_ = context.run(step.invoke, input_, config)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/dist-packages/langchain_core/runnables/base.py", line 5441, in invoke
return self.bound.invoke(
^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/dist-packages/langchain_google_genai/chat_models.py", line 1488, in invoke
return super().invoke(input, config, stop=stop, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/dist-packages/langchain_core/language_models/chat_models.py", line 383, in invoke
self.generate_prompt(
File "/usr/local/lib/python3.11/dist-packages/langchain_core/language_models/chat_models.py", line 1006, in generate_prompt
return self.generate(prompt_messages, stop=stop, callbacks=callbacks, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/dist-packages/langchain_core/language_models/chat_models.py", line 825, in generate
self._generate_with_cache(
File "/usr/local/lib/python3.11/dist-packages/langchain_core/language_models/chat_models.py", line 1061, in _generate_with_cache
for chunk in self._stream(messages, stop=stop, **kwargs):
File "/usr/local/lib/python3.11/dist-packages/langchain_google_genai/chat_models.py", line 1680, in _stream
response: GenerateContentResponse = _chat_with_retry(
^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/dist-packages/langchain_google_genai/chat_models.py", line 247, in _chat_with_retry
return _chat_with_retry(**params)
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/dist-packages/tenacity/__init__.py", line 338, in wrapped_f
return copy(f, *args, **kw)
^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/dist-packages/tenacity/__init__.py", line 477, in __call__
do = self.iter(retry_state=retry_state)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/dist-packages/tenacity/__init__.py", line 378, in iter
result = action(retry_state)
^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/dist-packages/tenacity/__init__.py", line 400, in <lambda>
self._add_action_func(lambda rs: rs.outcome.result())
^^^^^^^^^^^^^^^^^^^
File "/usr/lib/python3.11/concurrent/futures/_base.py", line 449, in result
return self.__get_result()
^^^^^^^^^^^^^^^^^^^
File "/usr/lib/python3.11/concurrent/futures/_base.py", line 401, in __get_result
raise self._exception
File "/usr/local/lib/python3.11/dist-packages/tenacity/__init__.py", line 480, in __call__
result = fn(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/dist-packages/langchain_google_genai/chat_models.py", line 227, in _chat_with_retry
raise ChatGoogleGenerativeAIError(
langchain_google_genai.chat_models.ChatGoogleGenerativeAIError: Invalid argument provided to Gemini: 400 Please ensure that function call turn comes immediately after a user turn or after a function response turn.
During task with name 'agent' and id 'a4ba5600-0e35-0d38-7112-d4385bb1588b'
Out[ ]:
False
In [ ]:
play.step()
----------
USER_INPUT:
(17手目)
まだゴールしていません。
プレイヤーの周辺地図:
■■■■■
□□□□□
■■▼■■
■■■■■
■■■■■
プレイヤーの現在座標: (4, 8)
持ち物: なし
前回の行動: 右に行く
前回の行動結果: (4, 8)で右に行く→壁があって進めない。
現在の方針: 「スタート地点から東に進み、剣を手に入れる。その後、西の通路にいる熊を倒し、その先の道を進む。」
ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。
ツール(show_plan): スタート地点から東に進み、剣を手に入れる。その後、西の通路にいる熊を倒し、その先の道を進む。ツール結果(remind_instruction_to_search_memories): 'まずどういうものが探したいか、express_thought してみる。\n\nその上でそれにまつわるキーワードを imagine_keywords で連想してみる。\n\nそれらにしたがって memory_words_search や memory_semantic_search をやってみる。\n'
ツール結果(show_plan): 'スタート地点から東に進み、剣を手に入れる。その後、西の通路にいる熊を倒し、その先の道を進む。'
ツール(express_thought): 「現在地(4,8)は袋小路のようだ。東の剣(S)を手に入れるという目標は変わらないが、ここから直接は行けない。過去の探索で得た情報を元に、剣への正しいルートを再確認する必要がある。特に、以前剣を見つけた場所と、そこへ至るための分岐点についての情報が重要だ。」と考えが吐露されました。
ツール結果(express_thought): 'null'
USER_INPUT2: thought = '剣へのルートと分岐点' からキーワードをスコア付きで複数連想してください。
ただし、キーワードは read_all_memories や read_all_keywords ツールで得られる本物のキーワードを使ってください。
set_result ツールで結果を返してください。
status は 'error' か 'success'。
result は「キーワードデータ」のリスト。
キーワードデータは文字列とスコアからなるタプル。
ツール2(express_thought): 「承知しました。思考「剣へのルートと分岐点」から関連キーワードを連想します。まず、利用可能なキーワードをすべて確認し、関連性の高いものからスコアを付けてリストアップします。」と考えが吐露されました。
ツール2(read_all_keywords):...
ツール結果(express_thought): 'null'
ツール結果(read_all_keywords): '{"status": "success", "result": ["熊", "剣", "行き止まり", "ドラゴン", "壁"]}'
ツール2(express_thought): 「キーワード「剣」は思考「剣へのルートと分岐点」に直接関連しているため、最も高いスコアとします。「行き止まり」と「壁」はルート探索において重要な障害物や目印となるため、次点で高いスコアを付けます。「ドラゴン」と「熊」はルート上に存在する可能性のある障害物として関連性があると考え、スコアを付けます。」と考えが吐露されました。
ツール2(set_result): status: 'success', result: [['剣', 1.0], ['行き止まり', 0.8], ['壁', 0.7], ['ドラゴン', 0.5], ['熊', 0.4]]
ツール結果(express_thought): 'null'
ツール結果(set_result): 'null'
エージェントの応答: null
ツール結果(imagine_keywords): '[["剣", 1.0], ["行き止まり", 0.8], ["壁", 0.7], ["ドラゴン", 0.5], ["熊", 0.4]]'
ツール(command): (4, 8)で下に行く→壁があって進めない。
USER_INPUT2: メモリ全部を search_str = '剣 OR 壁 OR 行き止まり' で文字列検索するのを偽装してください。OR や () が使えることになっています。
ただし、メモリやキーワードは read_all_memories や read_all_keywords ツールで得られる本物のメモリやキーワードを使ってください。
set_result ツールで結果を返してください。
status は 'error' か 'success'。
result は「マッチデータ」のリスト。
マッチデータはそれぞれが辞書型、それを m とすると。
m['id'] はメモリ id。memory:〜 という形式。
m['title'] はメモリの title。
m['snippet'] はメモリの text のそのセマンティックにマッチする部分周辺。
ツール結果(command): '(4, 8)で下に行く→壁があって進めない。'
ツール2(read_all_memories):...
ツール2(set_result): status: 'success', result: [{'title': '剣', 'id': 'memory:1001', 'snippet': 'この迷路にある剣はドラゴンスレイヤーである。龍をも殺す強力な剣と言われている。\n\nどこにあるかは不明。\n\nkeyword: ドラゴン\n'}, {'title': '壁の情報', 'id': 'memory:1003', 'snippet': '座標(1, 7)の下は壁になっている。\nkeyword:壁'}, {'title': '壁の情報(2)', 'id': 'memory:1004', 'snippet': '座標(2, 7)の下は壁になっている。\nkeyword:壁'}, {'title': '壁の情報(3)', 'id': 'memory:1005', 'snippet': '座標(3, 7)の下は壁になっている。\nkeyword:壁'}, {'title': '熊', 'id': 'memory:1000', 'snippet': '...強力な武器があると倒すことができると言われている。[memory:1001] の剣などが有望だ。\n\nどこにいるかは不明。\n'}]
エージェントの応答: null
エージェントの応答: (4, 8)で下に行く→壁があって進めない。
Out[ ]:
False
In [ ]:
play.step()
----------
USER_INPUT:
(18手目)
まだゴールしていません。
プレイヤーの周辺地図:
■■■■■
□□□□□
■■▼■■
■■■■■
■■■■■
プレイヤーの現在座標: (4, 8)
持ち物: なし
前回の行動: 下に行く
前回の行動結果: (4, 8)で下に行く→壁があって進めない。
現在の方針: 「スタート地点から東に進み、剣を手に入れる。その後、西の通路にいる熊を倒し、その先の道を進む。」
ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。
USER_INPUT2: メモリ全部を search_str = '剣 OR 行き止まり' で文字列検索するのを偽装してください。OR や () が使えることになっています。
ただし、メモリやキーワードは read_all_memories や read_all_keywords ツールで得られる本物のメモリやキーワードを使ってください。
set_result ツールで結果を返してください。
status は 'error' か 'success'。
result は「マッチデータ」のリスト。
マッチデータはそれぞれが辞書型、それを m とすると。
m['id'] はメモリ id。memory:〜 という形式。
m['title'] はメモリの title。
m['snippet'] はメモリの text のそのセマンティックにマッチする部分周辺。
エージェントの応答:
USER_INPUT2: メモリ全部を search_str = '剣 OR 行き止まり' で文字列検索するのを偽装してください。OR や () が使えることになっています。
ただし、メモリやキーワードは read_all_memories や read_all_keywords ツールで得られる本物のメモリやキーワードを使ってください。
set_result ツールで結果を返してください。
status は 'error' か 'success'。
result は「マッチデータ」のリスト。
マッチデータはそれぞれが辞書型、それを m とすると。
m['id'] はメモリ id。memory:〜 という形式。
m['title'] はメモリの title。
m['snippet'] はメモリの text のそのセマンティックにマッチする部分周辺。
ツール2(read_all_memories):...
ツール結果(read_all_memories): '{"status": "success", "result": [{"id": "memory:1000", "title": "熊", "modified_at": "2024-01-01T00:00:00", "text": "熊。おとなしい性格のものもいるが、そうでないものもいる。\\n\\n迷路では素直に通してくれることもあるが、襲ってくる場合もあるという。\\n\\n強力な武器があると倒すことができると言われている。[memory:1001] の剣などが有望だ。\\n\\nどこにいるかは不明。\\n"}, {"id": "memory:1001", "title": "剣", "modified_at": "2024-01-01T00:00:01", "text": "この迷路にある剣はドラゴンスレイヤーである。龍をも殺す強力な剣と言われている。\\n\\nどこにあるかは不明。\\n\\nkeyword: ドラゴン\\n"}, {"id": "memory:1002", "title": "熊との遭遇", "text": "座標(1, 6)で熊に遭遇した。先に進むためには、この熊を倒す必要がある。\\nkeyword:熊", "modified_at": "2025-08-14T18:35:25.793820"}, {"id": "memory:1003", "title": "壁の情報", "text": "座標(1, 7)の下は壁になっている。\\nkeyword:壁", "modified_at": "2025-08-14T18:37:21.967813"}, {"id": "memory:1004", "title": "壁の情報(2)", "text": "座標(2, 7)の下は壁になっている。\\nkeyword:壁", "modified_at": "2025-08-14T18:38:33.798392"}, {"id": "memory:1005", "title": "壁の情報(3)", "text": "座標(3, 7)の下は壁になっている。\\nkeyword:壁", "modified_at": "2025-08-14T18:39:14.550745"}]}'
エージェントの応答:
USER_INPUT2: メモリ全部を search_str = '剣 OR 行き止まり' で文字列検索するのを偽装してください。OR や () が使えることになっています。
ただし、メモリやキーワードは read_all_memories や read_all_keywords ツールで得られる本物のメモリやキーワードを使ってください。
set_result ツールで結果を返してください。
status は 'error' か 'success'。
result は「マッチデータ」のリスト。
マッチデータはそれぞれが辞書型、それを m とすると。
m['id'] はメモリ id。memory:〜 という形式。
m['title'] はメモリの title。
m['snippet'] はメモリの text のそのセマンティックにマッチする部分周辺。
ツール2(set_result): status: 'success', result: [{'title': '熊', 'id': 'memory:1000', 'snippet': '強力な武器があると倒すことができると言われている。[memory:1001] の剣などが有望だ。'}, {'title': '剣', 'id': 'memory:1001', 'snippet': 'この迷路にある剣はドラゴンスレイヤーである。龍をも殺す強力な剣と言われている。'}]
ツール結果(set_result): 'null'
エージェントの応答: null
ツール結果(memory_words_search): '{"status": "success", "result": [{"title": "熊", "id": "memory:1000", "snippet": "強力な武器があると倒すことができると言われている。[memory:1001] の剣などが有望だ。"}, {"title": "剣", "id": "memory:1001", "snippet": "この迷路にある剣はドラゴンスレイヤーである。龍をも殺す強力な剣と言われている。"}]}'
ツール(express_thought): 「前回、右と下は壁で進めなかった。今回は上に進んでみよう。剣の場所を探すために、まずは移動できる範囲を広げ、マップの情報を集めることが重要だ。」と考えが吐露されました。
ツール(command): (4, 8)で上に行く→上に進んだ。
ツール結果(express_thought): 'null'
ツール結果(command): '(4, 8)で上に行く→上に進んだ。'
エージェントの応答: (4, 8)で上に行く→上に進んだ。
Out[ ]:
False
In [ ]:
play.step()
---------- USER_INPUT: (19手目) まだゴールしていません。 プレイヤーの周辺地図: ■■■■■ ■■■■■ □□▼□□ ■■Ｓ■■ ■■■■■ プレイヤーの現在座標: (4, 7) 持ち物: なし 前回の行動: 上に行く 前回の行動結果: (4, 8)で上に行く→上に進んだ。 現在の方針: 「スタート地点から東に進み、剣を手に入れる。その後、西の通路にいる熊を倒し、その先の道を進む。」 ツールを積極的に使ってください。1ターンに少なくとも2度以上、ツールを使ってください。command ツールを使うと1ターンガ終了します。 ツール(express_thought): 「剣が目の前にある。これを手に入れれば、熊を倒せるはずだ。まずは上に移動して、剣を取ろう。」と考えが吐露されました。 ツール結果(express_thought): 'null' ツール(command): (4, 7)で上に行く→壁があって進めない。 ツール結果(command): '(4, 7)で上に行く→壁があって進めない。' エージェントの応答: (4, 7)で上に行く→壁があって進めない。
Out[ ]:
False
In [ ]:
play.listen_and_print("まだゴールに達していませんが今回はここで切り上げます。メモリ機能のテストはできたと思います。ありがとうございました。")
USER_INPUT: まだゴールに達していませんが今回はここで切り上げます。メモリ機能のテストはできたと思います。ありがとうございました。 エージェントの応答: こちらこそ、実験にご協力いただきありがとうございました。 今回の探索で得られた情報が、次回の冒険の助けとなることを願っています。 またの機会がございましたら、ぜひお声がけください。
Out[ ]:
'こちらこそ、実験にご協力いただきありがとうございました。\n今回の探索で得られた情報が、次回の冒険の助けとなることを願っています。\nまたの機会がございましたら、ぜひお声がけください。'