第三章：无限循环

我们遇到了一个问题。

想象一下，把第二章的脚本运行两次。你说“我叫 Alice。“Claude 说了声你好。再次运行，问“我叫什么名字？“Claude 说“我不知道。”

这是因为 LLM 是无状态的。它们完全没有记忆。每一次请求，对它来说都是第一次见面。

要构建一个智能体，我们需要通过创建人工记忆来解决这个问题。

记忆的幻觉

LLM 中的“记忆“不是硬盘，而是一份日志文件。

当你和 ChatGPT 聊天时，它并不真正“记得“你五分钟前说过什么。在幕后，代码会在每条新消息发出时，把完整的对话历史一并发送给模型。

图 2. 上下文累积：第一轮只向 API 发送“User: Hi“。第二轮则将完整历史——“User: Hi”、“Assistant: Hello”、“User: How are you?”——全部发送给 API。

模型每次都能看到完整的对话记录。这就是其中的诀窍。

让我们手动实现这个上下文循环。但在此之前，我们需要让代码具备可测试性。

测试的难题

有一个残酷的现实：你不能通过真正调用 LLM 来测试一个由 LLM 驱动的应用程序。

API 调用很慢（每次需要 2 到 10 秒），很贵（每次调用都需要花费真实的金钱），而且具有不确定性（每次可能得到不同的响应）。想象一下，运行一套测试套件要花 5 美元、耗时 20 分钟——你根本不会想去运行它。

解决方案是依赖注入。我们不把 API 调用硬编码在智能体内部，而是传入一个“大脑“对象。在生产环境中，大脑是 Claude；在测试中，大脑是一个能返回可预测响应的伪对象。

我们现在就建立这个模式，在编写更多生产代码之前。

响应类型

在构建大脑之前，我们需要定义它返回的内容。Claude 的 API 会返回包含多个内容块的复杂 JSON。我们需要简单的 Python 对象来处理这些数据。

背景： Claude 可以在单个响应中返回文本、工具调用，或两者兼而有之。我们需要普通的数据对象来表示这些可能性。（我们有意跳过了 @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 汇报结果时，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 与我们真实的 brain 具有相同的接口——一个接受对话并返回 Thought 的 think() 方法。

旁注： 这种模式——用可预测的伪对象替换真实依赖以便测试——称为测试替身。Martin Fowler 的文章 “Mocks Aren’t Stubs”¹ 解释了各种变体（fakes、stubs、mocks、spies）。对于 LLM 测试来说，一个带有预置响应的简单伪对象通常就足够了。

定义成功

在编写生产代码之前，让我们先定义何为成功。这些测试将指导我们的实现。

测试 1：brain 返回响应

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。我们还启用了扩展思考——这是一项让模型在回答之前先进行内部草稿式思考的功能。可以把它想象成模型在开口说话之前，先在一张草稿纸上自言自语。这会消耗额外的 tokens，但质量提升十分显著，尤其是在第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 行： 我们启用了扩展思考——模型在响应之前会生成推理摘要，从而提升处理复杂任务时的质量，但代价是消耗更多 token。budget_tokens 限制了模型可用于推理的 token 数量（此处为 10,000）——这些 token 和输出 token 一样计入费用。在我们的设置中，max_tokens 涵盖了包括思考和响应在内的全部输出，因此 Anthropic 要求它必须超过 budget_tokens。当思考 token 上限为 10,000、max_tokens 为 16,000 时，响应本身最多可使用 6,000 个 token。
第 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 行： 通过依赖注入接收一个 brain。初始化一个空的对话列表。
第 109 行： 在调用 brain 之前，将用户消息追加到历史记录中。
第 112-120 行： 调用 brain，以暗淡文本显示最多五行思考内容（\033[2m 是用于暗淡显示的 ANSI 转义码，\033[0m 用于重置），提取响应内容，并将其追加到历史记录中。
第 121-123 行： 如果 API 调用失败，则移除刚刚添加的用户消息。这样可以确保对话保持在有效状态。

请注意第 109 行：我们在调用 brain 之前就添加了用户消息。brain 需要看到包含当前消息在内的完整对话内容。

主循环（已更新）

主循环现在只是一个精简的 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()，并打印结果。这种分离使得 agent 具有可测试性——我们可以直接测试 Agent.handle_input()，而无需模拟 input() 或 print()。

验证测试通过

再次运行测试：

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 列表都会增长：

轮次	近似 Token 数
1	50
10	5,000
100	50,000

最终，你会触及上下文限制——Claude Sonnet 为 200k token，DeepSeek 为 128k，某些本地模型甚至低至 4k。一旦超出，API 就会返回 400 Bad Request。我们在第 121 行的错误处理会捕获这个错误并报告，所以智能体不会悄无声息地崩溃。但对话实际上已经卡死了——由于历史记录依然过长，后续每条消息都会失败。

目前，重启智能体可以清空历史记录，让你重新回到正轨。我们将在第 9 章构建反馈循环时，加入真正的上下文压缩机制——即追踪 API 响应中的 token 用量，并在消息溢出之前自动对旧消息进行摘要。那才是对话真正容易爆掉的地方，也是这个修复方案真正派上用场的时候。

本章小结

Claude 现在“记得“了——或者说，我们让它误以为自己记得。对话列表随着每一轮交互不断增长，而 FakeBrain 让我们无需花费一分钱就能测试整个流程。

这两种模式将贯穿本书其余部分。我们构建的每一个大脑（Claude、DeepSeek、Ollama）都将实现相同的 think() 接口，而 FakeBrain 将对它们全部进行测试。

还有一个遗留问题：我们的代码是硬编码到 Anthropic API 的。如果想添加 DeepSeek 或本地模型，就不得不重复大量代码。

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

第二章：原始请求

Up next

第四章：通用适配器