第3章:無限ループ

問題があります。

第2章のスクリプトを2回実行してみましょう。「私の名前はアリスです」と言うと、Claudeは挨拶を返します。もう一度実行して「私の名前は何ですか?」と尋ねると、Claudeは「わかりません」と答えます。

これは、LLMがステートレスだからです。完全な記憶喪失状態なのです。すべてのリクエストが、まるで初めて会ったかのようになります。

エージェントを構築するには、人工的な記憶を作成することでこの問題を解決する必要があります。

記憶の幻想

LLMにおける「記憶」は、ハードドライブではありません。ログファイルなのです。

ChatGPTとチャットする時、5分前に何を言ったかを「記憶」しているわけではありません。舞台裏では、新しいメッセージが送られるたびに、コードが会話履歴全体をモデルに送り返しているのです。

コンテキストの蓄積:ターン1では「User: Hi」だけがAPIに送信されます。ターン2では履歴全体—「User: Hi」、「Assistant: Hello」、「User: How are you?」—がAPIに送信されます。
図 2. コンテキストの蓄積:ターン1では「User: Hi」だけがAPIに送信されます。ターン2では履歴全体—「User: Hi」、「Assistant: Hello」、「User: How are you?」—がAPIに送信されます。

モデルは毎回、完全な会話記録を見ているのです。それがトリックの正体です。

このようなコンテキストループを手動で実装していきます。しかし、その前にコードをテスト可能にする必要があります。

テストの課題

ここで厳しい事実を述べましょう:LLMを使用したアプリケーションを、実際にLLMを呼び出してテストすることはできません。

API呼び出しは遅く(2-10秒かかる)、高価で(1回ごとに実際のお金がかかる)、非決定的(毎回異なる応答が得られる可能性がある)です。$5かかり20分もかかるテストスイートを想像してみてください。絶対に実行したくないでしょう。

解決策は依存性の注入です。エージェント内にAPI呼び出しをハードコーディングする代わりに、「brain」オブジェクトを渡します。本番環境では、このbrainはClaudeです。テストでは、予測可能な応答を返す偽物のbrainを使用します。

これから本番コードを書く前に、このパターンを確立していきます。

レスポンスの型

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

コンテキスト: Claudeは1つの応答でテキスト、ツール呼び出し、またはその両方を返すことができます。これらの可能性を表現するためのデータクラスが必要です。

コード:

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はAI実行環境がツールの実行を要求する際に使用されます。idは追跡用の一意の識別子です(Claudeが結果を報告する際に必要とします)。nameは実行するツールを指定し、argsはパラメータを含む辞書です。

現時点ではAI実行環境がツールを呼び出すことができないため、ToolCallはまだ使用しませんが、これがThoughtレスポンスタイプの一部となるため、ここで定義しています。ツールを追加すると、Claudeがファイルの読み込みやコマンドの実行を要求する際にこれを返すようになります。

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

Thoughtは、思考処理後にブレインが返すものです。テキスト、ツールコール、その両方、またはどちらも含まない場合があります。この抽象化により、他のコードを変更することなく、後でClaudeをDeepSeekに置き換えることが可能になります。

フェイクブレインパターン

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

コンテキスト: 予測可能なレスポンスを返し、呼び出された回数を追跡し、受け取った会話を記録するブレインが必要です。

コード:

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に解析する必要があります。

コード:

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

コードの解説:

  • 39-41行目: APIキーをロードし、存在しない場合は即時終了します。
  • 43-44行目: 設定を保存します。後でモデルを設定可能にします。
  • 46行目: think()メソッドは、FakeBrainと同様のブレインのインターフェースです。
  • 52-55行目: ペイロードには"messages": conversationが含まれています—現在のメッセージだけでなく、完全な会話履歴です。これがコンテキストループです。
  • 61行目: Claudeの複雑なレスポンス形式を、シンプルなThoughtに解析します。

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

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

ClaudeのAPIは「コンテンツブロック」のリストを返します。各ブロックにはtypeがあり、"text""tool_use"のいずれかです。すべてのテキストブロックを1つの文字列にまとめ、tool_useブロックをToolCallオブジェクトに変換します。

Agentクラス(更新版)

ここでは、第1章のAgentを更新し、ブレインを受け入れ、会話履歴を保持できるようにします。

コード:

 86 class Agent:
 87     """A coding agent with conversation memory."""
 88 
 89     def __init__(self, brain):
 90         self.brain = brain
 91         self.conversation = []
 92 
 93     def handle_input(self, user_input):
 94         """Handle user input. Returns output string, raises AgentStop to quit."""
 95         if user_input.strip() == "/q":
 96             raise AgentStop()
 97 
 98         if not user_input.strip():
 99             return ""
100 
101         self.conversation.append({"role": "user", "content": user_input})
102 
103         try:
104             thought = self.brain.think(self.conversation)
105             text = thought.text or ""
106             self.conversation.append({"role": "assistant", "content": text})
107             return text
108         except Exception as e:
109             self.conversation.pop()  # Remove failed user message
110             return f"Error: {e}"

ウォークスルー:

  • 89-91行目: 依存性注入によってbrainを受け取ります。空の会話リストを初期化します。
  • 101行目: brainを呼び出す前にユーザーのメッセージを履歴に追加します。
  • 103-107行目: brainを呼び出し、テキストを抽出し、応答を履歴に追加します。
  • 108-110行目: APIコールが失敗した場合、追加したばかりのユーザーメッセージを削除します。これにより会話が有効な状態に保たれます。

101行目に注目してください:brainを呼び出す前にユーザーメッセージを追加します。brainは現在のメッセージを含む会話全体を見る必要があるからです。

メインループ(更新版)

メインループは現在、単なる薄いI/Oラッパーとなっています:

115 def main():
116     brain = Claude()
117     agent = Agent(brain)
118     print("⚡ Nanocode v0.2 (Conversation Memory)")
119     print("Type '/q' to quit.\n")
120 
121     while True:
122         try:
123             user_input = input("❯ ")
124             output = agent.handle_input(user_input)
125             if output:
126                 print(f"\n{output}\n")
127 
128         except (AgentStop, KeyboardInterrupt):
129             print("\nExiting...")
130             break
131 
132 
133 if __name__ == "__main__":
134     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

すべてグリーンです。テストは1回のAPIコールも行うことなく、私たちの実装を検証します。

メモリのテスト

では、実際の本体でテストを行いましょう:

1 python nanocode.py

この会話をお試しください:

1 ❯ I am building a Python agent.
2 (Claude is thinking...)
3 
4 That sounds exciting! Building a Python agent is a great project...
5 
6 ❯ What language am I using?
7 (Claude is thinking...)
8 
9 You are using Python.

会話リストは期待通りに機能しています。

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

「これを永遠に実行し続けることはできるの?」と考えているかもしれません。

いいえ。

ループが1回実行されるたびに、messagesリストは大きくなっていきます:

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

最終的に、コンテキスト制限に到達します—Claude Sonnetでは200kトークン、DeepSeekでは128kトークン、一部のローカルモデルでは4kトークンと低くなることもあります。この制限を超えると、APIは400 Bad Requestを返します。108行目のエラー処理がこれを捕捉してエラーを報告するため、エージェントが静かにクラッシュすることはありません。しかし、会話は実質的に行き詰まります—履歴が長すぎる状態が続くため、その後のメッセージもすべて失敗します。

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

まとめ

Claudeは今や記憶を持っています—より正確には、記憶があるように見せかけることができました。会話リストは各ターンで成長し、FakeBrainを使えば1セントも使わずにすべてをテストできます。

これらのパターンは本書全体を通して継続します。私たちが構築するすべてのブレイン(Claude、DeepSeek、Ollama)は同じthink()インターフェースを実装し、FakeBrainがそれらすべてをテストします。

未解決の問題が1つあります:現在のコードはAnthropicのAPIに強く結びついています。DeepSeekやローカルモデルを追加しようとすると、多くのコードを複製しなければならないでしょう。


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