这样写完全可以，你的问题列表涵盖了大模型（LLM）Agent 框架从架构设计（Protocol、工厂模式）、异步并发（to_thread）、数据安全（脱敏）、状态管理（Memory、上下文控制）、工具链设计（安全、校验）到工程落地（CLI、Web API 化）的方方面面。这是一套非常高质量的、工业级的 Agent 框架深度思考题。
作为资深工程师，我将为你逐一提供详尽、深度且符合生产环境最佳实践的解答。
---
## 一、 基础架构与工程挑战
### 1. Python Protocol 是什么？和 ABC 抽象基类有什么区别？
* **Python Protocol（结构化子类型 / 鸭子类型）**：
自 Python 3.8 引入，属于**静态鸭子类型（Static Duck Typing）**。它不需要显式继承，只要一个类实现了 Protocol 中定义的属性和方法，类型检查器（如 mypy）就认为它“实现了该接口”。
* **ABC（抽象基类，名义子类型 / Nominal Subtyping）**：
显式继承。子类必须显式继承自 `abc.ABC`，并实现 `@abstractmethod`，否则在实例化时会报错。
* **核心区别与选型**：
* **耦合度**：ABC 是强耦合（必须显式继承），Protocol 是零耦合（只要长得像就行）。
* **生态适配**：如果你在写一个三方库，无法修改用户的类，但希望用户的类符合某种规范，用 Protocol；如果你在定义项目内部核心的、必须严格约束的生命周期组件，用 ABC 更好。
### 2. asyncio.to_thread() 为什么要用？同步阻塞代码和异步事件循环如何共存？
* **为什么要用**：
Python 的 `asyncio` 运行在一个单线程的事件循环（Event Loop）上。如果在这个循环里执行了同步阻塞代码（比如 `time.sleep()`、没有异步驱动的数据库查询、或者像 `requests.get()` 这样的网络 IO），**整个事件循环就会被冻结**，其他所有的并发任务都会被卡住。
* **如何共存**：
`asyncio.to_thread()` 会在幕后把这个同步阻塞函数扔进 Python 的线程池（ThreadPoolExecutor）里去运行，并返回一个 `awaitable` 对象。这样，同步代码在独立的线程里阻塞，而主线程的异步事件循环可以继续调度其他协程。
### 3. 工厂模式（Factory Pattern）在真实项目中的落地写法
在生产环境中，工厂模式通常结合**配置驱动**和**注册表机制**，避免一堆 `if-else`。
```python
from typing import Dict, Type
from athena.infra.llm import LLMClient, OpenAIClient, AnthropicClient # 假设的实现
class LLMClientFactory:
_registry: Dict[str, Type[LLMClient]] = {}
@classmethod
def register(cls, provider_name: str, client_cls: Type[LLMClient]):
cls._registry[provider_name.lower()] = client_cls
@classmethod
def create(cls, provider: str, api_key: str, **kwargs) -> LLMClient:
client_cls = cls._registry.get(provider.lower())
if not client_cls:
raise ValueError(f"Unsupported provider: {provider}")
# 参数校验通常会在这里触发或在 Client 内部用 Pydantic 校验
return client_cls(api_key=api_key, **kwargs)
# 注册组件（通常在应用启动时完成）
LLMClientFactory.register("openai", OpenAIClient)
LLMClientFactory.register("anthropic", AnthropicClient)
```
### 4. 日志脱敏——为什么要清洗错误信息？
* **防止凭据泄露**：大模型 API 的 SDK（如 LiteLLM、OpenAI）在报错时，经常会将整个请求上下文、HTTP Header 甚至 **API Key**（如 `Authorization: Bearer sk-...`）作为字符串塞进 Exception 的 message 里。
* **合规性与审计（PII/数据安全）**：生产环境的日志会被同步到 ELK、Datadog 或阿里云 SLS 等集中式日志系统。如果日志中包含未经脱敏的 API Key 或用户隐私数据（PII），任何有日志查看权限的人员都能看到，严重违反安全合规标准（如 GDPR、ISO27001）。
---
## 二、 LLMClient 进阶设计
### 1. asyncio.to_thread() vs 原生异步
* **如果 LiteLLM 提供了原生的 `async_completion()`，绝对不需要再用 `to_thread()**`。`to_thread()` 会带来线程切换的开销，而原生异步通过非阻塞 IO 效率最高。
* **改写方案**：
```python
# 改造后的 complete 变成真正的异步方法
async def complete(self, messages: list[LLMMessage], **kwargs) -> LLMResponse:
# 丢弃 _complete_sync 方法，直接调用原生的 async_completion
try:
response = await litellm.async_completion(
model=self.model,
messages=[m.to_dict() for m in messages],
**kwargs
)
return self._parse_response(response)
except Exception as e:
raise AthenaError(self._sanitize_error_message(str(e)))
```
### 2. 单一职责挑战：LLMClientFactory.create() 的 SRP 违背问题
* **同意该看法**。如果参数校验规则非常复杂（比如不仅判空，还要校验 Key 的格式、模型有效性），放在工厂里会导致工厂类频繁因为“校验规则变动”而修改。
* **拆分方案**：引入专门的 Validator 或利用 **Pydantic**。
```python
from pydantic import BaseModel, SecretStr, field_validator
class LLMConfig(BaseModel):
provider: str
api_key: SecretStr
model: str
@field_validator("api_key")
@classmethod
def validate_key(cls, v: SecretStr, info) -> SecretStr:
# 这里只做参数格式校验
if not v.get_secret_value().startswith("sk-"):
raise ValueError("Invalid API Key format")
return v
class LLMClientFactory:
@classmethod
def create(cls, config: LLMConfig) -> LLMClient:
# 工厂只负责：根据合法的配置创建对象（单一职责）
client_cls = cls._get_client_class(config.provider)
return client_cls(config)
```
### 3. 安全边界：通用脱敏版本设计
只处理 `"sk-"` 绝对不够用。Anthropic、Google、OpenAI 的前缀和长度各不相同，直接用复杂的正则表达式（Regex）进行泛化匹配和替换是更通用的做法。
```python
import re
def _sanitize_error_message(message: str) -> str:
# 匹配常见 AI 厂商的 API Key 模式（如 sk-..., sk-ant-..., AIzaSy... 等）
# 匹配规律：通常是 sk-[a-zA-Z0-9]{32,96} 或类似结构
patterns = [
r"sk-[a-zA-Z0-9\-]{24,96}", # OpenAI, Anthropic, DeepSeek
r"AIzaSy[a-zA-Z0-9_\-]{33}", # Google Gemini
]
sanitized = message
for pattern in patterns:
sanitized = re.sub(pattern, "[REDACTED_API_KEY]", sanitized)
return sanitized
```
### 4. 测试友好性（依赖注入）
如果把 `completion` 函数注入进来：
```python
class LiteLLMClient:
def __init__(self, completion_fn=litellm.completion):
self.completion_fn = completion_fn # 依赖注入
```
* **好处**：
* **单元测试极其方便**：测试时只需传入一个 Mock 函数 `lambda k: mock_res`，完全不需要 mock 掉整个 `litellm` 库，摆脱了 `monkeypatch`。
* **坏处**：
* **破坏了封装性**：类的调用方需要感知底层的实现细节，初始化参数变多了。如果不给默认值，会导致配置代码变冗长。
### 5. （选做）流式输出扩展
* **修改选择**：建议**新加一个 `stream()` 方法**，而不是修改现有的 `complete()`。因为两者的返回消费模式完全不同（一个是一次性返回，一个是迭代器），强行合并会导致上层调用必须写大量的 `if/else` 来判断是否是生成器，违反接口隔离原则。
```python
# LLMClient 协议扩展
class LLMClient(Protocol):
async def complete(self, messages: list[LLMMessage], **kwargs) -> LLMResponse: ...
async def stream(self, messages: list[LLMMessage], **kwargs) -> AsyncGenerator[str, None]: ...
```
---
## 三、 VectorStore 向量数据库进阶
### 1. InMemoryVectorStore 的性能边界与非 Milvus 提速方案
* **有多慢**：10 万条向量，如果维度是 1536（OpenAI 默认），每次 search 进行全量余弦相似度计算，由于 Python 的循环开销和 GIL，耗时大约在 **200ms ~ 500ms**。对于 Agent 的一步（Step）来说，这个延迟是不可接受的（因为还要叠加 LLM 延迟）。
* **不接入 Milvus 的提速方案**：
1. **引入 FAISS 或 Scikit-learn**：直接在内存里改用底层是 C++ 实现的 `faiss` 库进行 IVF（倒排索引）或 HNSW（分层导航可小世界）近似最近邻（ANN）搜索，延迟可降到 **几毫秒**。
2. **numpy 向量化计算**：如果不引入外部库，至少使用 `numpy` 的矩阵点乘代替 Python 循环，利用 CPU 的 SIMD 加速。
### 2. 连接池问题
* **问题**：高并发下（如 100 请求），频繁创建/销毁 TCP 连接会导致 **端口耗尽（TIME_WAIT 状态过多）**，大幅增加网络握手延迟（3-way handshake），甚至导致 Milvus 服务端拒绝连接。
* **改造方案（单例/连接池模式）**：
```python
class MilvusVectorStore:
_client_instance = None
def __init__(self, connection_args: dict):
self.connection_args = connection_args
@property
def client(self):
# 保证同一个进程/实例内部，全局复用同一个 Client 连接
if MilvusVectorStore._client_instance is None:
from pymilvus import MilvusClient
MilvusVectorStore._client_instance = MilvusClient(**self.connection_args)
return MilvusVectorStore._client_instance
```
### 3. embedding 字段的权衡
* **影响**：这是一种“偷懒”的做法。把返回的 `MemoryDocument.embedding` 填成“查询向量”，意味着你把**原始存储的信息给丢弃或覆盖了**。
* **潜在 Bug 场景**：
* **记忆更新/搬迁**：如果上层逻辑发现这条记忆需要更新（比如做权重衰减、或者重新写回另一个 collection），直接读取这条 `MemoryDocument` 并写入，就会把错误的 embedding（查询向量）存进去，导致知识库遭到污染。
### 4. 幂等性的局限：分布式竞争条件解决
* **分布式初始化解决方案**：
1. **外部分布式锁**：使用 Redis 或 ZooKeeper 加分布式锁。
2. **提前初始化（推荐）**：在系统 CD 部署（或主节点启动脚本）时，通过独立的 migration 脚本提前创建好 Collection，禁止在应用运行期（Runtime）并发做 DDL（数据定义语言）操作。
3. **捕获特定异常**：直接去创建，捕获 Milvus 返回的 `CollectionAlreadyExists` 异常，如果捕获到则证明其他实例创建成功了，直接忽略。
### 5. （选做）扩展新的向量库
* **步骤**：① 实现 `QdrantVectorStore(VectorStore)`；② 实现 `QdrantClientProtocol`。
* **上层代码改动**：上层代码**完全不需要改动**，因为上层只依赖 `VectorStore` 协议。
* **架构评估**：**完美达到了“对扩展开放，对修改关闭”（OCP）的目标**。这就是 Protocol 带来的面向接口编程的威力。
---
## 四、 Agent 抽象与多模态设计
### 1. 接口扩展问题：多模态支持
* **推荐做法：修改 `run()` 的参数结构（引入 Context/Message 对象），而不是新建 Protocol。**
* **理由与取舍**：
* 如果直接加 `image_url: str` 参数，未来支持音频、视频时接口会爆炸。
* **最佳方案**：将 `query: str` 改为 `query: list[UserContent]`（类似于大模型的 Multi-modal Message 结构）。
* **取舍**：直接加参数破坏向前兼容性；建新 Protocol 会导致框架割裂，增加上层路由的复杂度。
### 2. AgentResponse 的 steps 字段改造
* **不够用**：`list[str]` 纯文本无法进行结构化分析，很难通过代码精确提取出调用了什么工具、耗时多久、耗费了多少 Token。
* **数据结构改造**：
```python
class StepDetail(BaseModel):
step_number: int
thought: str
tool_name: Optional[str] = None
tool_input: Optional[dict] = None
tool_output: Optional[str] = None
duration_ms: float
tokens_used: int
class AgentResponse(BaseModel):
output: str
steps: list[StepDetail] # 结构化步骤
```
### 3. 单方法接口的局限（cancel, pause, get_status）
* **应用场景**：在 Web 界面、长文本生成、复杂长跑任务（Long-running Task）中，用户需要中途“取消（Cancel）”以节省 Token，或者查看 Agent 当前进展（“正在搜索资料...”）。
* **对现有代码的影响**：影响巨大。现有的 `run()` 是一切到底的同步/异步阻塞函数。为了支持这些，必须改写为**事件驱动架构**，或者将 `run()` 变成基于**任务 ID** 的异步任务流，状态存储在外部 Redis 中。
### 4. 流式输出的挑战（打字机效果）
* **签名修改**：
```python
async def run(self, query: str) -> AsyncGenerator[AgentEvent, None]: ...
```
* **事件设计**：不能只返回 `str`。因为中间既有 Agent 的 Thought，又有工具的 Observation，最后才是 Final Answer。应当定义一个 `AgentEvent` 类，前端根据事件类型（如 `type="thought"`, `type="answer"`）来决定是局部打印还是流式打字。
### 5. （选做）多 Agent 协作
* 现有的 `AgentResponse`（只有最终 output）只适合简单的单体任务。
* **需要添加的字段**：
```python
class AgentResponse(BaseModel):
output: str
context_carrier: dict # 包含中间状态、未解决的约束、提取的结构化实体
next_agent_hint: Optional[str] = None # 建议路由给下一个 Agent 的名字
```
---
## 五、 ReAct 核心逻辑缺陷与治理
### 1. scratchpad 的 Token 限制问题
* **滑动窗口（Sliding Window）**：只保留最近的 N 轮 Tool 迭代。
* *优缺点*：实现简单；但会导致 Agent 忘记前几步拿到的核心关键信息，引发循环调用。
* **压缩摘要（Summarization）（推荐）**：当长度超过阈值，调用一个轻量模型，将前 6 步的“思考+观察”压缩精简为一句总结（如：“前 6 步中我已尝试通过计算器得到 X=5”），释放上下文。
* *优缺点*：信息保留好；但增加了额外的 LLM 耗时和 Token 成本。
### 2. 并发安全性问题
* **重大问题**：如果两个用户同时请求同一个 `ReActAgent` 实例，由于它们**共享**了同一个 `WorkingMemory` 和 `scratchpad`，A 用户的请求和 B 用户的请求会**交叉混杂在同一个 scratchpad 里**，导致 AI 逻辑彻底错乱。
* **现有架构不支持多用户并发**。
* **改造方案**：`ReActAgent` 内部不应该持有任何状态（State）。`scratchpad` 和 `WorkingMemory` 应该作为**上下文变量（ContextVar）**，或者每次调用 `run(query, session_id)` 时，临时从外部加载或在方法内部实例化。
### 3. 工具调用失败后的策略
* **死循环风险**：如果工具连续报错（例如网络超时），AI 往往会非常“执着”地用相同的参数重复调用该工具，陷入 **死循环** 直到耗尽 `max_steps`。
* **熔断与降级机制**：
* 引入计数器：同一个工具连续失败 3 次，强制触发**熔断**。
* 在 Observation 里显式加入策略提示：“*（系统提示：该工具已连续失败，请尝试换一种方法或直接向用户报错）*”。
### 4. max_steps 耗尽的体验改进
* **改进兜底行为**：不直接报错，在 `step == max_steps` 时，强行修改最后的 Prompt，发送给 LLM 一条强制指令：
> “你已达到最大思考步数。请根据你目前在 scratchpad 中记录的现有线索和发现，做出一份尽力而为的总结，并直接回答用户，诚实说明哪些部分未完成。”
### 5. （选做）流式 ReAct
* `run()` 返回类型改为 `AsyncGenerator`。
* `yield` 关键字的作用：它把一个普通的函数变成了一个生成器。在 Agent 内部，每当 LLM 吐出一个 `Thought` 或者工具返回一个 `Observation` 时，立即 `yield` 丢给前端展示，然后再继续下一轮循环，完美实现“步骤级别的进度推送”。
---
## 六、 提示词工程与契约精神
### 1. 模板引擎的选择（str.format vs Jinja2）
* **Jinja2 能实现什么**：支持 `{% if scratchpad %}`、`{% for tool in tools %}` 等条件判断和循环。
* **何时值得引入**：当你的 Agent 支持**动态工具注入**（根据用户权限展示不同工具说明），或者面临**复杂的 Prompt 剪枝**（如果没有历史记忆，就彻底不渲染 `## History` 这一行，避免污染 LLM 认知）时，必须引入 Jinja2。
### 2. 提示词长度控制
* 当前代码通常没有任何控制，极其危险。
* **加在哪里**：应当在 `ContextAssembler.assemble()` 组装完成、发给 LLM 之前的**临界点**进行检查。可以使用 `tiktoken` 计算总 Token 数，如果超限，根据业务权重截断 `scratchpad` 或触发 `WorkingMemory` 紧急剪枝。
### 3. output_contract 的脆弱性
* **根本解决方案：开启大模型的强约束开关。**
* 在调用 LiteLLM/OpenAI 时，配置 `response_format={"type": "json_object"}`（JSON Mode），或者更高级地传入 Pydantic 结构体作为 `response_format`（即 Structured Outputs）。这样大模型的解码器（Decoder）会在 Token 级别做语法约束，保证返回的 **100% 是合法 JSON**，不再需要任何脆弱的正则提取。
### 4. 多语言支持
* **不建议改成中文系统提示词**：因为绝大多数基座模型（尤其是开源模型或 GPT/Claude 系列）对**英文 System Prompt 的遵从度明显高于中文**。
* **设计方案**：在 System Prompt 的结尾追加一句指令：`"Always respond to the user in the language they used (e.g., if the user asks in Chinese, your thought and final answer should be in Chinese)."`。
### 5. （选做）提示词版本管理
* **改造 ContextAssembler**：
```python
class ContextAssembler:
def __init__(self, template_dir: str = "./templates"):
self.template_dir = template_dir
def assemble(self, version: str = "v1", **kwargs) -> str:
# 根据版本号动态读取 react.v1.md 或 react.v2.md
template_path = f"{self.template_dir}/react.{version}.md"
with open(template_path) as f:
return f.read().format(**kwargs)
```
---
## 七、 短期记忆（WorkingMemory）深度优化
### 1. importance 分数的赋值策略
* **自动化设计**：在 `WorkingMemory.add_message(message: LLMMessage)` 内部做一层自动路由转换。
```python
def add_message(self, message: LLMMessage):
# 策略模式：根据角色或消息特征自动计算重要性
if message.role == "user":
importance = 2.0
elif message.role == "system":
importance = 5.0
elif "Final Answer" in message.content:
importance = 3.0
else:
importance = 1.0 # 默认中间过程
self._store_with_importance(message, importance)
```
### 2. 剪枝策略的局限
* **接下来会发生什么**：当那条 `importance=1.0` 的消息被删掉后，如果空间依然超限，代码会开始**在剩下的 4 条 `importance=2.0` 的重要消息中随机删除或者卡死在死循环里**（取决于循环退出条件）。
* **是否合理**：很不合理。应当引入时间权重衰减（FIFO 兜底）机制。当重要性一致时，优先删除“最早发生的”消息，保留最新的上下文。
### 3. render() 格式的影响
* **会有明显不同**：大模型对符号非常敏感。`Markdown` 格式（如 `### User:`）通常能让模型产生更好的“语义分块”认知，而传统的 `user: ` 在长文本下容易和正文混淆。
* **测试方法（A/B Testing Prompt）**：使用相同的一组多样化复杂问题（包含多轮对话），分别使用两套格式灌给模型，统计**任务完成率（Task Success Rate）**和**错幻觉率**。
### 4. 多轮对话的上下文完整性（长短期记忆协同）
* **根本局限**：WorkingMemory 是一定会忘掉老信息的。
* **协同解决思路**：
1. 当 WorkingMemory 触发剪枝时，**被删掉的消息不要直接丢弃**，而是异步转换成 Embedding 存入长期记忆 **`VectorStore`**。
2. 在每一轮用户提问时，先去 `VectorStore` 里召回相似的历史对话片段，作为 `## Context From History` 注入到当前请求中。这就是经典的 **RAG + 缓存** 记忆架构。
### 5. （选做）线程安全问题
* **实际会发生吗**：**在单线程的 `asyncio` 环境下，普通的 while 纯计算循环不会发生竞争条件**。因为 Python 协程切换只有在遇到 `await` 关键字、`async with` 或 `async for`（触发了异步 IO）时才会发生。如果 `_prune_if_needed()` 内部全是纯内存/数学计算，它会一口气执行完，中间没有任何协程切出点，因此是天然线程安全的。
---
## 八、 工具链与安全沙箱
### 1. 工具的安全性边界
* **沙箱执行（Sandbox）**：如果允许执行任意 Python 代码，必须放在轻量级隔离环境（如 **Docker 容器**、AWS Lambda 或专用的安全沙箱如 **Wasm/Wazero**）中运行。
* **允许列表（Allowlist）**：对危险模块（`os`, `shutil`, `subprocess`）进行静态代码扫描（AST 分析），发现直接拦截。
* **权限分级**：区分“只读工具”（如天气查询）和“写工具”（如删除数据库）。写工具触发时，必须在 CLI/Web 端引入 **Human-in-the-loop（人工审批拦截）**。
### 2. 参数类型校验
* **后果**：如果没有类型校验，`calculator` 内部如果执行 `expression.strip()` 就会直接抛出 `AttributeError: 'int' object has no attribute 'strip'`，导致 Agent 崩溃。
* **加在哪里**：应当**统一加在 `invoke()` 里**（或者利用 Pydantic/`inspect` 库在统一的基类层面拦截）。不要让每个工具函数内部去重复写 `isinstance`，保持工具开发的纯粹性。
### 3. 工具描述的质量影响与测试方法
* **AI 的表现**：描述模糊会导致 AI **“乱试工具”**（过度调用）或者 **“有工具却不用”**（幻觉）。
* **测试方法**：编写 **“工具召回测试集”**。给出一句特定的 query（例如：“帮我算一下 23*45”），不真正运行 Agent，只让 LLM 根据工具列表选择它认为对的工具 ID。统计 Top-1 准确率（Accuracy）。
### 4. 同名工具的覆盖保护机制
* **推荐策略：直接报错（Raise Error）**。
* **理由**：在分布式或大型团队协作开发中，工具同名往往意味着严重的命名冲突。隐式覆盖或只打警告，会导致极为隐蔽的 Bug（开发者以为调的是 A 团队的工具，实际运行的是 B 团队的同名工具）。只有在系统初始化阶段直接抛出 `ValueError` 熔断崩溃，才能让问题尽早暴露。
### 5. （选做）工具超时控制
```python
import asyncio
async def invoke(self, tool_name: str, args: dict, timeout: float = 5.0):
tool = self.tools.get(tool_name)
try:
# 使用 asyncio.wait_for 强制包裹工具的执行（假设工具已异步化）
result = await asyncio.wait_for(tool.run(**args), timeout=timeout)
return result
except asyncio.TimeoutError:
return f"Error: Tool {tool_name} timed out after {timeout} seconds."
```
---
## 九、 应用入口与架构演进
### 1. asyncio.run() 的反复创建问题
* `asyncio.run()` 每次调用都会全新创建和销毁一个 Loop，开销极高。
* **改造方案**：将整个 CLI 的生命周期包装在一个异步函数中，在最高层统一只调用一次 `asyncio.run()`。
```python
async def main_loop():
# 把整个 while 循环和交互逻辑变成 async
while True:
user_input = prompt("> ")
if user_input == "exit": break
await agent.run(user_input)
if __name__ == "__main__":
import asyncio
asyncio.run(main_loop()) # 全局仅此一次
```
### 2. start() 的记忆设计：重置对话
* **实现方式**：在 CLI 的 `start()` 交互循环里拦截特殊命令。
* **修改文件**：修改 `athena/cli/commands.py`（负责 CLI 命令响应的文件）。
```python
# 在 while 循环内部
if user_input.strip() == "/reset":
agent.working_memory.clear() # 假设 memory 提供了 clear 方法
print("对话已重置。")
continue
```
### 3. 错误处理策略差异
* **合理性评估**：
* `chat()`（单次执行命令）：遇到致命错误理应退出（Code=1），方便 Shell 脚本感知失败。
* `start()`（交互式会话）：打印而不退出是合理的，因为不能因为一次网络波动就把用户的整个聊天会话给直接杀掉。
* **何时应该终止 `start()**`：遇到 **API Key 无效（Unauthorized）**、磁盘空间彻底满（No space left on device）等“非一过性”、无论重试多少次都绝不可能成功的系统级环境致命错误时，`start()` 应该直接退出。
### 4. 新增命令的步骤（athena history）
1. **修改的地方**：CLI 注册文件（如 `cli.py`）增加 `@app.command(name="history")`。
2. **最大的挑战**：正如提示所说，`WorkingMemory` 目前属于**进程内内存存储**。一旦上一个命令执行完程序退出了，内存数据就彻底灰飞烟灭了。
3. **解决前提**：必须先将记忆层改造为**持久化存储**（例如每次对话自动序列化写入到本地的 `~/.athena/history.json` 或 SQLite 中），`athena history` 命令再去读取该文件。
### 5. （选做）Web API 化（FastAPI 改造）
* **支持度**：**现有架构对改造支持度极好**，因为核心能力都通过 Protocol 隔离了。
* **需要新建的文件**：新建 `athena/api/main.py` 用于写 FastAPI 的路由。
* **build_agent() 能复用吗**：**完全可以复用**，把它作为 FastAPI 的依赖项（Dependency Injection）或者在应用启动（`lifespan`）时初始化。
* **asyncio.run() 还需要吗**：**绝对不需要了**。FastAPI 本身底层基于 Uvicorn（ASGI 服务器），它已经托管了全局的事件循环，你的路由函数直接写 `async def` 并在内部 `await agent.run()` 即可。
