第3章:無限ループ

問題があります。

第2章のスクリプトを2回実行する場面を想像してください。「私の名前はAliceです」と言うと、Claudeは挨拶を返します。もう一度実行して「私の名前は何ですか?」と聞くと、Claudeは「わかりません」と答えます。

これはLLMがステートレスだからです。LLMには記憶がまったくありません。すべてのリクエストにおいて、LLMにとってあなたは初対面です。

エージェントを構築するには、人工的なメモリを作成してこの問題を解決する必要があります。

メモリという幻想

LLMにおける「メモリ」はハードドライブではありません。ログファイルです。

ChatGPTとチャットするとき、5分前に言ったことを「覚えている」わけではありません。舞台裏では、新しいメッセージのたびに会話履歴全体がモデルに送信されています。

コンテキストの蓄積:ターン1ではAPIに「User: Hi」だけを送信する。ターン2ではAPIに「User: Hi」「Assistant: Hello」「User: How are you?」という全履歴を送信する。
図 2. コンテキストの蓄積:ターン1ではAPIに「User: Hi」だけを送信する。ターン2ではAPIに「User: Hi」「Assistant: Hello」「User: How are you?」という全履歴を送信する。

モデルは毎回、会話の全文を見ています。それがこの仕掛けです。

このコンテキストループを手動で実装してみましょう。ただし、その前にコードをテスト可能な状態にする必要があります。

テストの問題

厳しい現実をお伝えします:LLMを実際に呼び出してLLM搭載アプリケーションをテストすることはできません。

API呼び出しは遅く(1回あたり2〜10秒)、コストがかかり(呼び出しのたびに課金されます)、非決定的です(毎回異なるレスポンスが返ってくる可能性があります)。5ドルのコストで20分かかるテストスイートの実行を想像してみてください。とても実行する気にはなれないでしょう。

解決策は依存性の注入です。API呼び出しをエージェントの内部にハードコーディングする代わりに、「brain」オブジェクトを外部から渡します。本番環境では、brainはClaudeです。テストでは、brainは予測可能なレスポンスを返すフェイクです。

このパターンを今のうちに確立しておきましょう。本番コードをこれ以上書く前に。

レスポンスの型

brainを構築する前に、それが返す値を定義する必要があります。ClaudeのAPIは、複数のコンテンツブロックを含む複雑なJSONを返します。扱いやすいシンプルなPythonオブジェクトが必要です。

コンテキスト: Claudeは1回のレスポンスで、テキスト、ツール呼び出し、またはその両方を返すことができます。これらの可能性を表現するためのシンプルなデータオブジェクトが必要です。(@dataclassは意図的に使用しません――これらのクラスはシンプルなので、デコレーターを使えば数行は省略できますが、__init__が実際に何をしているかが見えにくくなります。)

コード:

17 class ToolCall:
18     """A tool invocation request from the brain."""
19 
20     def __init__(self, id, name, args):
21         self.id = id
22         self.name = name
23         self.args = args  # dict

ToolCall は、ブレインがツールの実行を私たちに依頼していることを表します。id は追跡用の一意の識別子です(結果を返す際に Claude が必要とします)。name は実行するツールの名前で、args はパラメータの辞書です。

ToolCall はまだ使用しません——ブレインはまだツールを呼び出せないからです——しかし、Thought レスポンス型の一部であるため、ここで定義しておきます。ツールを追加すると、Claude がファイルの読み込みやコマンドの実行を行いたいときに、これらを返すようになります。

26 class Thought:
27     """Standardized response from any Brain."""
28 
29     def __init__(self, text=None, tool_calls=None, thinking=None):
30         self.text = text  # str or None
31         self.tool_calls = tool_calls or []  # list of ToolCall
32         self.thinking = thinking  # str or None

Thoughtは、ブレインが思考した後に返すものです。テキスト、ツール呼び出し、その両方、またはどちらも含まない場合があります。thinkingフィールドはモデルの推論の要約を記録します。それがどこから来るのかは、以下でClaudeクラスを構築する際に確認します。この抽象化により、他のコードを一切変更せずに、後でClaudeをDeepSeekに差し替えることができます。

FakeBrainパターン

これで、テスト用のフェイクブレインを構築できます。

背景: 予測可能な応答を返し、呼び出された回数を追跡し、受け取った会話を記録するブレインが必要です。

コード:

class FakeBrain:
    """Fake brain for testing - returns predictable responses."""

    def __init__(self, responses=None):
        self.responses = responses or [Thought(text="Fake response")]
        self.call_count = 0
        self.last_conversation = None

    def think(self, conversation):
        self.last_conversation = list(conversation)  # Store a copy
        if self.call_count < len(self.responses):
            response = self.responses[self.call_count]
            self.call_count += 1
            return response
        return Thought(text="No more responses")

これはtest_nanocode.pyに記述するもので、プロダクションコードには書きません。FakeBrainは、実際のブレインが持つことになるのと同じインターフェースを持っていることに注目してください。つまり、会話を受け取ってThoughtを返すthink()メソッドです。

An icon of a info-circle1

補足: 実際の依存オブジェクトを、テスト用の予測可能なフェイクに置き換えるこのパターンは、テストダブルと呼ばれます。Martin Fowlerの記事「Mocks Aren’t Stubs」1では、その種類(フェイク、スタブ、モック、スパイ)について詳しく説明されています。LLMのテストでは、固定レスポンスを持つシンプルなフェイクがあれば、たいていの場合は十分です。

成功の定義

プロダクションコードを書く前に、成功とはどのような状態かを定義しておきましょう。これらのテストが実装の指針となります。

テスト1:ブレインがレスポンスを返す

1 def test_handle_input_returns_brain_response():
2     """Verify handle_input returns the brain's response text."""
3     brain = FakeBrain(responses=[Thought(text="Hello from brain!")])
4     agent = Agent(brain=brain)
5     result = agent.handle_input("hi")
6     assert result == "Hello from brain!"

brain=brain を Agent に渡していることに注目してください。これが依存性の注入の実例です。

テスト2:会話が蓄積される

 1 def test_conversation_accumulates():
 2     """Verify conversation list grows with each interaction."""
 3     brain = FakeBrain(responses=[
 4         Thought(text="Response 1"),
 5         Thought(text="Response 2")
 6     ])
 7     agent = Agent(brain=brain)
 8 
 9     agent.handle_input("First message")
10     assert len(agent.conversation) == 2  # user + assistant
11 
12     agent.handle_input("Second message")
13     assert len(agent.conversation) == 4  # 2 users + 2 assistants

各やり取りの後、会話にはユーザーメッセージとアシスタントの応答の両方が含まれている必要があります。

テスト3:正しいメッセージ構造

 1 def test_conversation_contains_correct_roles():
 2     """Verify conversation has correct role alternation."""
 3     brain = FakeBrain(responses=[Thought(text="AI response")])
 4     agent = Agent(brain=brain)
 5 
 6     agent.handle_input("User message")
 7 
 8     assert agent.conversation[0]["role"] == "user"
 9     assert agent.conversation[0]["content"] == "User message"
10     assert agent.conversation[1]["role"] == "assistant"
11     assert agent.conversation[1]["content"] == "AI response"

メッセージは、Claude が期待する正確なフォーマットである必要があります:{"role": "user", "content": "..."}

テスト4:Brain が会話を受け取る

 1 def test_brain_receives_conversation():
 2     """Verify brain.think is called with the conversation list."""
 3     brain = FakeBrain()
 4     agent = Agent(brain=brain)
 5 
 6     agent.handle_input("Test message")
 7 
 8     assert brain.last_conversation is not None
 9     assert len(brain.last_conversation) == 1
10     assert brain.last_conversation[0]["content"] == "Test message"

ブレインは現在のメッセージだけでなく、会話全体を受け取る必要があります。

今すぐこれらのテストを実行してください——すべて失敗するはずです:

1 pytest test_nanocode.py -v
1 FAILED test_nanocode.py::test_handle_input_returns_brain_response
2 FAILED test_nanocode.py::test_conversation_accumulates
3 ...

よし。では、テストをパスさせましょう。

Claude クラス

さて、いよいよ本物の頭脳部分です。

背景: Claude API をラップするクラスが必要です。このクラスは認証を処理し、会話履歴を送信し、レスポンスを Thought にパースする役割を担います。また、拡張思考も有効にします。これは、モデルが回答する前に内部的なメモ書きを行う機能です。モデルが話す前にメモ帳で自分自身に語りかけるイメージです。追加のトークンが必要になりますが、品質の向上は顕著です。特に第5章でツールを追加した後、モデルがどのツールをなぜ呼び出すべきかを推論する必要が生じる場面では、その効果が際立ちます。

コード:

37 class Claude:
38     """Claude API - the brain of our agent."""
39 
40     def __init__(self):
41         self.api_key = os.getenv("ANTHROPIC_API_KEY")
42         if not self.api_key:
43             raise ValueError("ANTHROPIC_API_KEY not found in .env")
44         self.model = "claude-sonnet-4-6"
45         self.url = "https://api.anthropic.com/v1/messages"
46 
47     def think(self, conversation):
48         headers = {
49             "x-api-key": self.api_key,
50             "anthropic-version": "2023-06-01",
51             "content-type": "application/json"
52         }
53         payload = {
54             "model": self.model,
55             "max_tokens": 16000,
56             "thinking": {
57                 "type": "enabled",
58                 "budget_tokens": 10000
59             },
60             "messages": conversation
61         }
62 
63         response = requests.post(self.url, headers=headers, json=payload, timeout=120)
64         response.raise_for_status()
65         return self._parse_response(response.json()["content"])

ウォークスルー:

  • 41〜43行目: APIキーを読み込み、見つからない場合は即座に失敗します。
  • 44〜45行目: 設定を格納します。モデルは後で設定可能にします。
  • 47行目: think() メソッドはブレインのインターフェースで、FakeBrain と同じです。
  • 55〜59行目: extended thinking(拡張思考)を有効にします。これにより、モデルは応答する前に推論の要約を生成し、複雑なタスクの品質が向上する一方、より多くのトークンを消費します。budget_tokens は、モデルが推論に使用できるトークン数の上限を設定します(ここでは10,000)。これらのトークンは出力トークンと同様に料金の対象となります。今回の設定では、max_tokens は思考と応答を含む合計出力をカバーするため、Anthropic は max_tokensbudget_tokens を超えるよう要求しています。思考トークンが10,000、最大トークンが16,000の場合、応答自体には最大6,000トークンを使用できます。
  • 60行目: ペイロードには "messages": conversation が含まれます。これは現在のメッセージだけでなく、会話の全履歴です。これがコンテキストループです。
  • 65行目: Claude の複雑なレスポンス形式をパースして、シンプルな Thought に変換します。

次はレスポンスパーサーです:

67     def _parse_response(self, content):
68         """Convert Claude's response format to Thought."""
69         text_parts = []
70         tool_calls = []
71         thinking = None
72 
73         for block in content:
74             if block["type"] == "thinking":
75                 thinking = block["thinking"]
76             elif block["type"] == "text":
77                 text_parts.append(block["text"])
78             elif block["type"] == "tool_use":
79                 tool_calls.append(ToolCall(
80                     id=block["id"],
81                     name=block["name"],
82                     args=block["input"]
83                 ))
84 
85         return Thought(
86             text="\n".join(text_parts) if text_parts else None,
87             tool_calls=tool_calls,
88             thinking=thinking
89         )

Claude の API は「コンテンツブロック」のリストを返します。各ブロックには type があり、"thinking""text""tool_use" のいずれかです。thinking ブロックが最初に届き、モデルの推論の要約が含まれています。これは呼び出し元が表示できるよう Thought に保存します。text ブロックはレスポンスになり、tool_use ブロックは ToolCall オブジェクトになります。パーサーは何も出力せず、生の JSON をきれいな Thought に変換するだけです。

Agent クラス(更新版)

第 1 章の Agent を更新して、ブレインを引数として受け取り、会話履歴を保持できるようにします。

コード:

 94 class Agent:
 95     """A coding agent with conversation memory."""
 96 
 97     def __init__(self, brain):
 98         self.brain = brain
 99         self.conversation = []
100 
101     def handle_input(self, user_input):
102         """Handle user input. Returns output string, raises AgentStop to quit."""
103         if user_input.strip() == "/q":
104             raise AgentStop()
105 
106         if not user_input.strip():
107             return ""
108 
109         self.conversation.append({"role": "user", "content": user_input})
110 
111         try:
112             thought = self.brain.think(self.conversation)
113             if thought.thinking:
114                 lines = thought.thinking.strip().split("\n")[:5]
115                 for i, line in enumerate(lines):
116                     prefix = "  💭 " if i == 0 else "     "
117                     print(f"\033[2m{prefix}{line}\033[0m")
118             text = thought.text or ""
119             self.conversation.append({"role": "assistant", "content": text})
120             return text
121         except Exception as e:
122             self.conversation.pop()  # Remove failed user message
123             return f"Error: {e}"

ウォークスルー:

  • 97〜99行目: 依存性の注入によってブレインを受け取ります。空の会話リストを初期化します。
  • 109行目: ブレインを呼び出す前に、ユーザーのメッセージを履歴に追加します。
  • 112〜120行目: ブレインを呼び出し、最大5行の思考内容をディム(薄暗く)表示します(\033[2mはdim表示のANSIエスケープコード、\033[0mでリセット)。レスポンスを抽出し、履歴に追加します。
  • 121〜123行目: API呼び出しが失敗した場合、直前に追加したユーザーメッセージを削除します。これにより、会話が有効な状態に保たれます。

109行目に注目してください。ブレインを呼び出す前にユーザーのメッセージを追加しています。ブレインは、現在のメッセージを含む会話全体を参照する必要があるためです。

メインループ(更新版)

メインループは、現在は薄いI/Oラッパーにすぎません:

128 def main():
129     brain = Claude()
130     agent = Agent(brain)
131     print("⚡ Nanocode v0.2 (Conversation Memory)")
132     print("Type '/q' to quit.\n")
133 
134     while True:
135         try:
136             user_input = input("❯ ")
137             output = agent.handle_input(user_input)
138             if output:
139                 print(f"\n{output}\n")
140 
141         except (AgentStop, KeyboardInterrupt):
142             print("\nExiting...")
143             break
144 
145 
146 if __name__ == "__main__":
147     main()

ロジックはすべて Agent クラスに含まれています。ループは入力を読み取り、handle_input() を呼び出して、結果を出力するだけです。この分離により、エージェントがテストしやすくなります——input()print() をモックする必要なく、Agent.handle_input() を直接テストできます。

テストが通ることを確認する

テストを再度実行してみましょう:

1 pytest test_nanocode.py -v
1 test_nanocode.py::test_handle_input_returns_brain_response PASSED
2 test_nanocode.py::test_conversation_accumulates PASSED
3 test_nanocode.py::test_conversation_contains_correct_roles PASSED
4 test_nanocode.py::test_brain_receives_conversation PASSED

全てグリーン。テストはAPIコールを一切行わずに、実装を検証します。

メモリをテストする

次は本物のブレインでテストしてみましょう:

1 python nanocode.py

次の会話を試してみましょう:

 1 ❯ I am building a Python agent.
 2   💭 The user is telling me about their project. They want to build
 3      a Python agent. I should respond helpfully and ask what kind
 4      of agent they're building.
 5 
 6 That sounds exciting! What kind of agent are you building?
 7 
 8 ❯ What language am I using?
 9   💭 The user previously said they are building a Python agent.
10      The answer is Python.
11 
12 You are using Python.

会話リストはその役割を果たしています。

コンテキストウィンドウの問題

「これをずっと動かし続けられるの?」と思っているかもしれません。

いいえ。

ループが反復するたびに、messages リストは大きくなっていきます:

ターン おおよそのトークン数
1 50
10 5,000
100 50,000

やがて、コンテキスト上限に達します——Claude Sonnet では 200k トークン、DeepSeek では 128k、ローカルモデルによっては 4k という低さのものもあります。上限を超えると、API は 400 Bad Request を返します。121 行目のエラーハンドリングがこれを捕捉してエラーを報告するので、エージェントが気づかないままクラッシュすることはありません。しかし、会話は実質的に行き詰まります——履歴がまだ長すぎるため、その後のメッセージもすべて失敗し続けるからです。

今のところ、エージェントを再起動すれば履歴がクリアされ、作業を再開できます。適切なコンテキスト圧縮——API レスポンスからトークン使用量を追跡し、オーバーフローする前に古いメッセージを自動的に要約する機能——は、第 9 章でフィードバックループを構築する際に追加します。会話が実際に膨れ上がるのはそこであり、その修正が真価を発揮するのもそこです。

まとめ

Claude はこれで記憶を持つようになりました——正確には、記憶を持っているかのように錯覚させることに成功しました。会話リストはターンごとに大きくなり、FakeBrain を使えば一切コストをかけずに全体をテストできます。

どちらのパターンも本書の残りの部分でも引き続き使用します。構築するすべてのブレイン(Claude、DeepSeek、Ollama)は同じ think() インターフェースを実装し、FakeBrain がそれらすべてをテストします。

一つ未解決の問題があります:私たちのコードは Anthropic の API にハードコードされています。DeepSeek やローカルモデルを追加しようとすると、多くのコードを複製しなければなりません。


  1. https://martinfowler.com/articles/mocksArentStubs.html↩︎