Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 徐涛
|
4348e4bf3e | ||
|
徐涛
|
0dc91faa43 | ||
|
徐涛
|
b4e5c7d651 | ||
|
徐涛
|
71abe881ed |
@@ -0,0 +1,647 @@
|
||||
# Phase 11: 测试与检索补强
|
||||
|
||||
## 背景与目标
|
||||
|
||||
AG Core 当前（v0.2.0-rc.1）已完成 Phase 0-10，全量测试 254 个，clippy 0 警告，11 个离线示例全部 exit 0。功能性交付物覆盖了 LLM Cycle、Prompt、Tool、Memory、Agent Runtime、流式事件、ContextSlot 上下文管理。但有两个系统性的短板尚未补齐：
|
||||
|
||||
1. **检索抽象缺失**：`memory` 模块只有 `MemoryRetriever`（基于 TextOverlap Dice 系数的关键词检索），缺少语义向量的检索抽象。`docs/roadmap.md` P1 中「VectorRetriever trait」一直未实现。
|
||||
2. **测试覆盖缺口**：Provider 的 roundtrip 测试停留在"基本响应 + 普通 401/500"层面，缺少结构化错误体解析、请求头验证、流式边界、工具调用端到端等关键场景的回归覆盖。多线程并发的 MemoryStore 测试只在 SqliteStore 有一个 10×10 场景，InMemoryStore 完全没有并发压力测试。
|
||||
|
||||
Phase 11 是 v0.2.0 正式版发布前的最后一个功能 Phase，三个 Step 的目标：
|
||||
|
||||
| Step | 内容 | 定位 |
|
||||
|------|------|------|
|
||||
| **11.1** | `VectorRetriever` trait + `InMemoryVectorRetriever` 引用实现 | P1 功能补全 |
|
||||
| **11.2** | wiremock Provider roundtrip 测试（12 个场景） | 测试质量补强 |
|
||||
| **11.3** | 并发测试补强（InMemoryStore + SqliteStore） | 并发安全验证 |
|
||||
|
||||
最终目标：全量测试从 254 → 275+，为 v0.2.0 正式版建立更高的质量基线。
|
||||
|
||||
## 需求推演概要
|
||||
|
||||
### Step 11.1 — VectorRetriever trait
|
||||
|
||||
**核心需求**：定义一个与后端无关的语义检索抽象接口，包含 `index(id, embeddings)` 索引和 `search(query, k)` 检索两个方法。附带一个基于 `HashMap` 全量余弦相似度扫描的参考实现。
|
||||
|
||||
**边界识别**：
|
||||
- 只定义 trait，不绑定任何具体后端（pgvector / qdrant / lancedb 留给社区或下游）
|
||||
- 不引入第三方向量数据库依赖
|
||||
- 引用实现的 `search()` 不做索引加速（O(n) 全量扫描已足够验证 trait 契约）
|
||||
- 不与 `MemoryStore` 耦合——`VectorRetriever` 是独立维度
|
||||
- 不嵌入到 `AgentSession` 或 `ContextSlot`（Phase 11 不承担集成消费端）
|
||||
|
||||
**关键假设**：
|
||||
- `Vec<f32>` 作为 embedding 类型已足够（大部分 embedding 模型输出 f32 向量）
|
||||
- 余弦相似度作为默认评分函数可覆盖主流场景
|
||||
- InMemoryVectorRetriever 的 `Mutex<HashMap>` 在 ~10K 向量内性能可接受
|
||||
|
||||
### Step 11.2 — wiremock Provider roundtrip 测试
|
||||
|
||||
**核心需求**：补充 12 个 wiremock 测试，覆盖目前缺失的关键回归场景——结构化 JSON 错误体解析、请求头验证、429 限流头解析、流式边界、工具调用端到端。
|
||||
|
||||
**边界识别**：
|
||||
- 只做 HTTP mock 层验证，不做端到端 LLM 模型调用
|
||||
- 每个测试自包含（启动自己的 MockServer），不抽共享 helper
|
||||
- 测试集中在 OpenAI（`GenericOpenaiProvider`）和 Anthropic（独立实现）两个核心 Provider 上
|
||||
- DeepSeek/Qwen/Ollama 同属 OpenAI Compat，继承 `GenericOpenaiProvider` 的测试覆盖
|
||||
|
||||
**关键假设**：
|
||||
- wiremock 的 `body_partial_json` matcher 可用且稳定（当前 dev-dependencies 中已有 wiremock）
|
||||
- OpenAI 和 Anthropic 的结构化错误体格式在当前 SDK 版本中未变化
|
||||
|
||||
### Step 11.3 — 并发测试补强
|
||||
|
||||
**核心需求**：验证 `MemoryStore` 两种实现（InMemoryStore + SqliteStore）在多线程并发写和混合读写场景下的正确性。
|
||||
|
||||
**边界识别**：
|
||||
- 不测试 `KnowledgeStore` / `ConversationMemory` 的并发——它们的行为完全由 `MemoryStore` 决定，不引入新 race 条件
|
||||
- 不测试 TTL 淘汰的并发正确性（TTL 淘汰使用 wall clock，非原子，不保证精确）
|
||||
- 混合读写测试只验证"无 panic + 数量正确"，不验证"读到的结果恰好与写顺序一致"（后者需要强一致快照，当前 Mutex 模型不提供）
|
||||
|
||||
**关键假设**：
|
||||
- `tokio::spawn` 100 个 task 同时写入 `Mutex<HashMap>`（InMemoryStore）不会死锁
|
||||
- SqliteStore 的 WAL 模式 + `busy_timeout=5000` 足够容忍 100 并发写
|
||||
|
||||
## 当前状态分析
|
||||
|
||||
### 测试覆盖率现状
|
||||
|
||||
| 维度 | 当前值 | Phase 11 目标 |
|
||||
|------|--------|-------------|
|
||||
| 全量测试 | 254 passed | 275+ passed |
|
||||
| InMemoryStore 测试 | 6 个（save/get/list/upsert/eviction/TTL） | +4 个并发 |
|
||||
| SqliteStore 测试 | 9 个（含 1 个 10×10 并发） | +1 个 100 并发 |
|
||||
| OpenAI wiremock 测试 | 4 个（basic/401/500/stream） | +8 个 |
|
||||
| Anthropic wiremock 测试 | 4 个（basic/401/529/stream） | +4 个 |
|
||||
| 请求头验证测试 | 0 个 | +2 个 |
|
||||
| ToolUse 端到端 mock 测试 | 0 个 | +2 个（OpenAI + Anthropic） |
|
||||
|
||||
### Provider 测试缺口
|
||||
|
||||
现有 wiremock 测试仅覆盖最基础的响应路径，以下关键场景缺失回归保护：
|
||||
|
||||
| 场景 | 缺失风险 |
|
||||
|------|---------|
|
||||
| OpenAI 请求体格式验证 | `body_partial_json` 未匹配，请求体结构变化无声 |
|
||||
| Authorization header 验证 | header 注入被修改时不告警 |
|
||||
| 结构化 401 JSON 错误体 | `error.message`/`error.code` 未消费，错误消息丢失 |
|
||||
| 429 + `retry-after` 头 | `RateLimit.retry_after` 字段不准确 |
|
||||
| ToolUse 端到端 mock | tool_flow 解析路径无回归 |
|
||||
| 流式 last chunk usage-only | `{choices:[], usage:{...}}` 可能 panic |
|
||||
|
||||
### MemoryStore 并发测试缺口
|
||||
|
||||
| Store | 当前并发测试 | 覆盖度 | 风险 |
|
||||
|-------|-------------|--------|------|
|
||||
| InMemoryStore | 0 个 | 无 | `Mutex` 锁竞争、deadlock、写入丢失 |
|
||||
| SqliteStore | 1 个（10 写者 × 10 次 = 100 条） | 中等 | `spawn_blocking` 线程池耗尽、WAL 锁等待超时 |
|
||||
|
||||
### 向量检索现状
|
||||
|
||||
`memory` 模块已有 `MemoryRetriever`（关键词检索）和 `retriever.rs` 中的 `RetrievalResult`/`ScoredItem` 类型。但语义向量检索维度完全空缺——无 trait、无引用实现、无测试。`docs/roadmap.md` 将 VectorRetriever 列为 P1，与 ContextSlot（P1，Phase 10）同级。
|
||||
|
||||
## 架构决策记录
|
||||
|
||||
| 决策 | 选择 | 放弃 | 理由 |
|
||||
|------|------|------|------|
|
||||
| 1. VectorRetriever trait 参数类型 | `Vec<f32>` 裸向量 | `Embedding` newtype | 包装类型增加可见复杂度但未提供运行时保护；大部分 embedding 模型输出 f32 向量；下游可自行包装 |
|
||||
| 2. `search()` 返回类型 | `Vec<(String, f32)>` | `ScoredItem`/`RetrievalResult` 命名 struct | `(String, f32)` 是 (id, score) 的最小表达；Phase 3 的 `RetrievalResult` 绑定了 `KnowledgePage` 引用，不适合向量检索场景；tuple 在 consumer 侧模式匹配更简洁 |
|
||||
| 3. 文件归属 | 新文件 `memory/vector.rs` | 合入 `memory/retriever.rs` | `retriever.rs` 已承载 302 行关键词检索代码，语义维度独立不应耦合；`vector.rs` 作为独立模块便于后期扩展（pgvector adapter 等） |
|
||||
| 4. 是否附带引用实现 | `InMemoryVectorRetriever` | trait-only | trait-only 是纯推测代码，无 consumer 验证引用实现作为"编译期测试"验证 trait 方法签名可用 |
|
||||
| 5. InMemoryVectorRetriever 余弦相似度实现方式 | 手动三行点积/范数 | `ndarray`/`approx` 等第三方依赖 | 余弦相似度数学固定，不需要外部依赖；1e-10 防零除；零新依赖原则 |
|
||||
| 5a | 引用实现不做向量维度校验 | 运行时维度检查 | 维度校验是具体后端(pgvector等)的职责；引用实现面向测试/验证场景；调用方负责传入等长向量 |
|
||||
| 6. 错误类型 | 复用 `MemoryError` 现有变体 | 新增 `VecRetrieval` 变体 | 向量检索与关键词检索语义等价于"检索"；`RetrievalError` 变体已覆盖索引/评分异常场景 |
|
||||
| 7. wiremock 测试组织 | 自包含（每个测试启动自己的 MockServer） | 共享 helper 函数 | 沿用现有测试模式（openai.rs line 825+、anthropic.rs line 900+）；自包含测试可独立运行、定位更直接 |
|
||||
| 8. 请求头验证 | 做（`body_partial_json` + `header` matcher） | 跳过 | 回归防御价值高——Provider 请求体结构变化会直接导致请求被拒绝，头验证是低成本高收益的回归保护 |
|
||||
| 9. 并发测试模式 | 100 并发写 + 混合读写（5 读 + 5 写）双模式 | 只做 100 并发写 | 两种模式互补：纯写入验证数据完整性和无 id 重复；混合读写验证读操作在并发写期间不 panic 且返回有效数据 |
|
||||
| 10. 实施顺序 | 11.1 → 11.2 → 11.3 | 任意顺序 | 与 roadmap 原定的 Step 顺序一致；11.1 是纯新增可独立交付；11.2/11.3 是对既有代码的测试追加，可并行但不优先于 11.1 |
|
||||
|
||||
## 设计方案
|
||||
|
||||
### Step 11.1 — VectorRetriever trait + InMemoryVectorRetriever
|
||||
|
||||
#### 文件位置
|
||||
|
||||
- 新增：`src/memory/vector.rs`
|
||||
- 修改：`src/memory.rs`（+2 行：module 声明 + re-export）
|
||||
|
||||
#### Trait 定义
|
||||
|
||||
```rust
|
||||
/// 语义向量检索器抽象接口。
|
||||
///
|
||||
/// 下游可实现此 trait 以对接向量数据库（pgvector / qdrant / lancedb 等）。
|
||||
/// 默认引用实现 [`InMemoryVectorRetriever`] 基于进程内 HashMap + 余弦相似度。
|
||||
///
|
||||
/// **稳定性**：实验性 API（v0.2.x），方法签名可能在 v0.3 中调整。
|
||||
/// 若未来需要 `remove()` / `clear()` 等方法，将在此 trait 中追加（带默认实现）。
|
||||
#[async_trait]
|
||||
pub trait VectorRetriever: Send + Sync {
|
||||
/// 将 `id` 对应的文本向量 `embeddings` 加入索引。
|
||||
async fn index(&self, id: String, embeddings: Vec<f32>) -> Result<(), MemoryError>;
|
||||
|
||||
/// 检索与 `query` 向量最相似的 `k` 条记录。
|
||||
/// 返回 `Vec<(id, score)>`，按 score 降序排列，score ∈ [0.0, 1.0]。
|
||||
async fn search(&self, query: Vec<f32>, k: usize) -> Result<Vec<(String, f32)>, MemoryError>;
|
||||
}
|
||||
```
|
||||
|
||||
#### InMemoryVectorRetriever 实现要点
|
||||
|
||||
```rust
|
||||
pub struct InMemoryVectorRetriever {
|
||||
vectors: Mutex<HashMap<String, Vec<f32>>>,
|
||||
}
|
||||
|
||||
impl InMemoryVectorRetriever {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
vectors: Mutex::new(HashMap::new()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl VectorRetriever for InMemoryVectorRetriever {
|
||||
async fn index(&self, id: String, embeddings: Vec<f32>) -> Result<(), MemoryError> {
|
||||
let mut vectors = self.vectors.lock().map_err(|e| {
|
||||
MemoryError::RetrievalError(format!("lock poisoned: {e}"))
|
||||
})?;
|
||||
vectors.insert(id, embeddings);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn search(&self, query: Vec<f32>, k: usize) -> Result<Vec<(String, f32)>, MemoryError> {
|
||||
let vectors = self.vectors.lock().map_err(|e| {
|
||||
MemoryError::RetrievalError(format!("lock poisoned: {e}"))
|
||||
})?;
|
||||
|
||||
if vectors.is_empty() || k == 0 {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
|
||||
let query_norm = dot(&query, &query).sqrt();
|
||||
if query_norm == 0.0 {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
|
||||
let mut scored: Vec<(String, f32)> = vectors
|
||||
.iter()
|
||||
.map(|(id, vec)| {
|
||||
let dot_product = dot(&query, vec);
|
||||
let vec_norm = dot(vec, vec).sqrt();
|
||||
let similarity = dot_product / (query_norm * vec_norm + 1e-10);
|
||||
(id.clone(), similarity)
|
||||
})
|
||||
.collect();
|
||||
|
||||
// 降序排列
|
||||
scored.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
|
||||
scored.truncate(k);
|
||||
Ok(scored)
|
||||
}
|
||||
}
|
||||
|
||||
/// 点积（手动循环，零依赖）。
|
||||
///
|
||||
/// 注意：`zip` 对不等长向量静默截断到较短者。引用实现不做维度校验，
|
||||
/// 调用方应确保 `a` 和 `b` 等长——不等长时结果无意义但不 panic。
|
||||
fn dot(a: &[f32], b: &[f32]) -> f32 {
|
||||
a.iter().zip(b.iter()).map(|(x, y)| x * y).sum()
|
||||
}
|
||||
```
|
||||
|
||||
#### 边界与约束
|
||||
|
||||
- **无维度校验**：不同维度向量传入 `search()` 时点积不报错，但余弦相似度结果无意义。维度校验是具体后端（pgvector等）的职责，引用实现不做运行时检查。
|
||||
- **零向量处理**：query 为零向量时直接返回空结果（`query_norm == 0.0`）。
|
||||
- **1e-10 防零除**：避免空库或全零向量导致除零 panic。
|
||||
|
||||
#### 测试（4 个）
|
||||
|
||||
| # | 测试名 | 验证点 |
|
||||
|---|--------|--------|
|
||||
| 1 | `basic_index_and_search` | index 两条（"rust" + "python")，用 "rustacean" 查询应排在首位 |
|
||||
| 2 | `search_empty_store` | 空库返回空 Vec |
|
||||
| 3 | `concurrent_index` | 10 个 task 各 index 1 条，总量 10、id 无重复 |
|
||||
| 4 | `concurrent_index_and_search` | 10 个 writer + 5 个 searcher 并发 2 秒，`search()` 遍历期间 `index()` 写入锁竞争不 panic |
|
||||
|
||||
#### 修改 `src/memory.rs`
|
||||
|
||||
```rust
|
||||
pub mod vector;
|
||||
|
||||
// 在高频 re-export 区追加
|
||||
pub use vector::{InMemoryVectorRetriever, VectorRetriever};
|
||||
```
|
||||
|
||||
### Step 11.2 — wiremock Provider roundtrip 测试
|
||||
|
||||
#### 测试清单
|
||||
|
||||
全部 12 个测试均遵循现有自包含模式：`MockServer::start()` → `Mock::given(...).and(...).respond_with(...)` → `provider.chat_blocking(...)` / `provider.chat_stream_inner(...)` → assert。
|
||||
|
||||
##### P0（7 个）
|
||||
|
||||
| # | 测试名 | 所属文件 | Mock 关键点 | 断言 |
|
||||
|---|--------|---------|------------|------|
|
||||
| 1 | `openai_request_body_format` | `openai.rs` | `body_partial_json` 匹配 `{"model": "gpt-4o", "messages": [{"role": "user"}]}` | 请求体结构正确，响应解析正常 |
|
||||
| 2 | `openai_authorization_header` | `openai.rs` | `header("authorization", "Bearer sk-test")` | header 精确匹配，响应解析正常 |
|
||||
| 3 | `openai_401_structured_error` | `openai.rs` | 返回 401 + `{"error": {"message": "Incorrect API key", "code": "invalid_api_key"}}` | `LlmError::Authentication(msg)` 且 message 包含 "Incorrect API key" |
|
||||
| 4 | `anthropic_401_structured_error` | `anthropic.rs` | 返回 401 + `{"error": {"type": "authentication_error", "message": "Invalid API key provided"}}` | `LlmError::Authentication(msg)` 且 message 包含 "Invalid API key" |
|
||||
| 5 | `openai_429_with_retry_after` | `openai.rs` | 返回 429 + `{"error": {"message": "Rate limit exceeded"}}` + `retry-after: 30` 头 | `LlmError::RateLimit { retry_after: Some(30s) }` |
|
||||
| 6 | `openai_tool_use_response` | `openai.rs` | 返回包含 `tool_calls` 的响应（choices[0].message.tool_calls ≠ null） | `StopReason::ToolUse` + `ContentBlock::ToolUse` 正确解析 |
|
||||
| 7 | `anthropic_tool_use_response` | `anthropic.rs` | 返回含 `type: "tool_use"` content block + `stop_reason: "tool_use"`（Anthropic 独立 wire 格式） | `StopReason::ToolUse` + `ContentBlock::ToolUse` 正确解析 |
|
||||
|
||||
##### P1（5 个）
|
||||
|
||||
| # | 测试名 | 所属文件 | Mock 关键点 | 断言 |
|
||||
|---|--------|---------|------------|------|
|
||||
| 8 | `anthropic_version_header` | `anthropic.rs` | `header("anthropic-version", "2023-06-01")` | header 精确匹配 |
|
||||
| 9 | `openai_stream_usage_only_last_chunk` | `openai.rs` | 流式最后 chunk `{"choices":[],"usage":{"prompt_tokens":5,"completion_tokens":2,"total_tokens":7}}` | 不 panic；`MessageComplete` 包含正确 usage |
|
||||
| 10 | `anthropic_529_overloaded_structured` | `anthropic.rs` | 返回 529 + `{"error": {"type": "overloaded_error", "message": "Overloaded"}}` | `LlmError::RateLimit { retry_after: None }` |
|
||||
| 11 | `openai_500_structured_error` | `openai.rs` | 返回 500 + `{"error": {"message": "Internal server error", "type": "server_error"}}` | `LlmError::Request { status: 500, body }` 且 body 包含 "Internal server error" |
|
||||
| 12 | `openai_stream_mid_stream_error` | `openai.rs` | 流式前几个 chunk 正常，中途服务端断开连接（模拟网络中断/限流断开） | `LlmError::Request(_)` — 流式中断映射为请求错误 |
|
||||
|
||||
#### 测试模式说明
|
||||
|
||||
```rust
|
||||
// 每个测试自包含，不抽共享 helper（沿用现有模式）
|
||||
#[tokio::test]
|
||||
async fn openai_authorization_header() {
|
||||
let server = MockServer::start().await;
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/chat/completions"))
|
||||
.and(header("authorization", "Bearer sk-test"))
|
||||
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
|
||||
"id": "chatcmpl-hdr",
|
||||
"object": "chat.completion",
|
||||
"created": 1,
|
||||
"model": "gpt-4o",
|
||||
"choices": [{"index": 0, "message": {"role": "assistant", "content": "OK"}, "finish_reason": "stop"}],
|
||||
"usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}
|
||||
})))
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
let provider = GenericOpenaiProvider::new_with_name(
|
||||
server.uri(), "sk-test".into(), "gpt-4o".into(), "openai", 30,
|
||||
);
|
||||
let response = provider.chat_blocking(MessageRequest {
|
||||
model: "gpt-4o".into(),
|
||||
messages: vec![Message::user_text("hi")],
|
||||
..Default::default()
|
||||
}).await.unwrap();
|
||||
assert_eq!(response.text(), "OK");
|
||||
}
|
||||
```
|
||||
|
||||
#### 新增依赖
|
||||
|
||||
dev-dependencies 中 wiremock 已就绪（当前 openai.rs / anthropic.rs 已在测试中使用），无需新增。
|
||||
|
||||
### Step 11.3 — 并发测试补强
|
||||
|
||||
#### 测试清单
|
||||
|
||||
| # | 测试名 | Store | 模式 | 验证标准 |
|
||||
|---|--------|-------|------|---------|
|
||||
| 1 | `concurrent_writers_max_pressure` | InMemoryStore | 100 task × 1 write | 总量 100, id 无重复 |
|
||||
| 2 | `concurrent_writers_max_pressure` | SqliteStore | 100 task × 1 write | 总量 100, id 无重复 |
|
||||
| 3 | `concurrent_mixed_read_write` | InMemoryStore | 预热 20 条, 5 读 + 5 写并发 2 秒 | 无 panic |
|
||||
| 4 | `concurrent_mixed_read_write` | SqliteStore | 预热 20 条, 5 读 + 5 写并发 2 秒 | 无 panic |
|
||||
| 5 | `concurrent_capacity_eviction` | InMemoryStore | 15 写者, max_items=10 | 最终 ≤ 10 |
|
||||
|
||||
#### 关键实现要点
|
||||
|
||||
**100 并发写模式**（InMemoryStore + SqliteStore 各一）：
|
||||
|
||||
```rust
|
||||
#[tokio::test]
|
||||
async fn concurrent_writers_max_pressure() {
|
||||
let store = Arc::new(InMemoryStore::new());
|
||||
|
||||
let mut handles = Vec::new();
|
||||
for i in 0..100 {
|
||||
let s = Arc::clone(&store);
|
||||
handles.push(tokio::spawn(async move {
|
||||
let id = format!("concurrent_{i}");
|
||||
s.save(make_item(&id)).await.unwrap();
|
||||
}));
|
||||
}
|
||||
for h in handles {
|
||||
h.await.unwrap();
|
||||
}
|
||||
|
||||
let list = store.list(&MemoryFilter::default()).await.unwrap();
|
||||
assert_eq!(list.len(), 100);
|
||||
let mut ids: Vec<String> = list.iter().map(|v| v.id.clone()).collect();
|
||||
ids.sort();
|
||||
ids.dedup();
|
||||
assert_eq!(ids.len(), 100);
|
||||
}
|
||||
```
|
||||
|
||||
**混合读写模式**（InMemoryStore + SqliteStore 各一）：
|
||||
|
||||
```rust
|
||||
#[tokio::test]
|
||||
async fn concurrent_mixed_read_write() {
|
||||
let store = Arc::new(InMemoryStore::new());
|
||||
|
||||
// 预热
|
||||
for i in 0..20 {
|
||||
store.save(make_item(&format!("seed_{i}"))).await.unwrap();
|
||||
}
|
||||
|
||||
let mut handles = Vec::new();
|
||||
// 5 个写者
|
||||
for w in 0..5 {
|
||||
let s = Arc::clone(&store);
|
||||
handles.push(tokio::spawn(async move {
|
||||
let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
|
||||
let mut i = 0;
|
||||
while tokio::time::Instant::now() < deadline {
|
||||
let id = format!("writer{w}_item{i}");
|
||||
s.save(make_item(&id)).await.unwrap();
|
||||
i += 1;
|
||||
}
|
||||
}));
|
||||
}
|
||||
// 5 个读者
|
||||
for r in 0..5 {
|
||||
let s = Arc::clone(&store);
|
||||
handles.push(tokio::spawn(async move {
|
||||
let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
|
||||
while tokio::time::Instant::now() < deadline {
|
||||
let _ = s.list(&MemoryFilter::default()).await.unwrap();
|
||||
}
|
||||
}));
|
||||
}
|
||||
for h in handles {
|
||||
h.await.unwrap();
|
||||
}
|
||||
// 不 panic 即算通过
|
||||
}
|
||||
```
|
||||
|
||||
**容量淘汰并发模式**（InMemoryStore）：
|
||||
|
||||
```rust
|
||||
#[tokio::test]
|
||||
async fn concurrent_capacity_eviction() {
|
||||
let eviction = EvictionConfig {
|
||||
policy: EvictionPolicy::Capacity { max_items: 10 },
|
||||
check_interval: 1,
|
||||
};
|
||||
let store = Arc::new(InMemoryStore::with_eviction(eviction));
|
||||
|
||||
let mut handles = Vec::new();
|
||||
for i in 0..15 {
|
||||
let s = Arc::clone(&store);
|
||||
handles.push(tokio::spawn(async move {
|
||||
s.save(make_item(&format!("item_{i}"))).await.unwrap();
|
||||
}));
|
||||
}
|
||||
for h in handles {
|
||||
h.await.unwrap();
|
||||
}
|
||||
|
||||
let list = store.list(&MemoryFilter::default()).await.unwrap();
|
||||
assert!(list.len() <= 10);
|
||||
}
|
||||
```
|
||||
|
||||
**测试归属**：
|
||||
- InMemoryStore 并发测试 → `src/memory/store/in_memory.rs` 的 `mod tests`
|
||||
- SqliteStore 并发测试 → `src/memory/store/sqlite_store.rs` 的 `mod tests`
|
||||
- 混合读写测试中的 `make_item` 辅助函数：直接复用各文件现有 `fn make_item`
|
||||
|
||||
## 已否决的方案
|
||||
|
||||
### 1. 砍掉 11.1（PM 建议）
|
||||
|
||||
**内容**：PM 在讨论中提出 VectorRetriever trait 无消费者，建议整体砍掉，等 Phase 12 或 v0.3 有人用时再做。
|
||||
|
||||
**否决理由**：引用实现作为 trait 契约的编译期验证手段——没有 consumer 不意味着 trait 签名不需要测试。同时社区贡献（pgvector adapter 等）需要稳定的 trait 边界。附带引用实现还可作为"如何在 agcore 中实现一个 VectorRetriever"的示例，降低社区参与门槛。代码量仅 ~80 行，维护成本可忽略。
|
||||
|
||||
### 2. trait-only VectorRetriever（无引用实现）
|
||||
|
||||
**内容**：只定义 `VectorRetriever` trait，不做 `InMemoryVectorRetriever`。
|
||||
|
||||
**否决理由**：trait-only 是纯推测代码——没有运行时验证，无法确认 trait 方法签名在实际调用链中是否可编译。理想情况下每个 trait 至少有一个引用实现来验证"这个 trait 确实可以被实现"。
|
||||
|
||||
### 3. 跳过请求头验证
|
||||
|
||||
**内容**：请求体格式和 Authorization header 验证是"过度保护"。
|
||||
|
||||
**否决理由**：`body_partial_json` + `header` matcher 的回归防御价值高。Provider 适配层的最大风险是请求体结构无声变更（如 `ToolDef` IR 切换时漏改了序列化字段），头验证是低成本（每测试 ~5 行）高收益的回归保护。
|
||||
|
||||
### 4. 先发 v0.2.0 正式版再迭代
|
||||
|
||||
**内容**：当前 rc.1 已经包含所有 P0 功能，建议直接发正式版，Phase 11 推迟到 v0.2.1。
|
||||
|
||||
**否决理由**：测试补强是正式版的信号而非负担。Phase 11 的三个 Step 都是"如果现在不做，以后更不会做"的类型。在正式版前补齐测试基线，避免「发布了再补测试」的经典陷阱。
|
||||
|
||||
### 5. 只做 100 并发写，不做混合读写
|
||||
|
||||
**内容**：并发写验证数据完整性已足够。
|
||||
|
||||
**否决理由**：纯写入和混合读写暴露不同类型的 bug。纯写入验证"数据不丢、id 无重复"；混合读写验证"读操作在并发写期间不 panic、返回有效数据"。两种模式互补缺失。
|
||||
|
||||
### 6. （实施后补充）openai_stream_mid_stream_error 的 mock 模式偏差
|
||||
|
||||
**实际实施**：返回 `200 + SSE content-type + 畸形 JSON payload`（`data: {not-valid-json}\n\ndata: [DONE]\n\n`），断言 `ChunkToEventStream` 产出 `StreamEvent::Error`。
|
||||
|
||||
**方案原文**：返回"前几个 chunk 正常，中途服务端断开连接"，断言 `LlmError::Request(_)`。
|
||||
|
||||
**偏差原因**：wiremock 0.6 标准 responder 的 `set_delay` / `set_body_string` 行为是「延迟响应 + 发完 body 后关闭连接」，无法精确模拟「send partial body then hang 保持连接」。`read_timeout` 配合 `set_delay` 触发的超时属于 send 阶段（`LlmError::Timeout`），不属于流式中断。
|
||||
|
||||
**采纳方案**：用畸形 JSON payload 替代——同样验证"流中途产生错误事件而不 panic"的回归保护意图，且在 wiremock 0.6 上 100% 可重现。断言改为 `StreamEvent::Error{message}` + "stream 最终结束"，保持核心回归价值。
|
||||
|
||||
**影响**：测试意图（流阶段错误检测）完全保留；mock 行为从「TCP 断开」变为「畸形应用层数据」；断言从 `LlmError::Request` 改为 `StreamEvent::Error`（语义等价：客户端发现流异常）。
|
||||
|
||||
### 7. （实施后补充）openai.rs 的 429 retry-after 解析修复
|
||||
|
||||
**实际实施**：`handle_error_response` 新增 `retry-after` header 解析逻辑（与 anthropic.rs 完全对齐）。
|
||||
|
||||
**方案原文**：方案测试 11.2.4 要求 `RateLimit { retry_after: Some(30s) }`，但 ADC 表中未显式列出此修复作为生产代码变更。
|
||||
|
||||
**修复原因**：原 `openai.rs:186-191` 的 429 分支固定 `retry_after: None`，与 `anthropic.rs:339-351` 已有的解析逻辑不一致。原代码注释甚至已写"仅读取 retry-after"，但实际未实现——这是隐藏 bug。修复让 OpenAI 兼容层（DeepSeek/Qwen 等）的限流重试信息可用，与 Anthropic 行为统一。
|
||||
|
||||
**影响**：方案测试 11.2.4 从「不可通过的回归保护」变为「可验证的实际行为」。变更 5 行，与 anthropic 实现完全镜像。
|
||||
|
||||
## 实施计划与顺序
|
||||
|
||||
**实施顺序**：11.1 → 11.2 → 11.3（与 roadmap 一致，每步可单独交付验证）。
|
||||
|
||||
| Step | 文件变更 | 测试增量 | 预估代码量 | 验证标准 |
|
||||
|------|---------|---------|-----------|---------|
|
||||
| 11.1 | +`src/memory/vector.rs`（~80 行），~`src/memory.rs`（+2 行） | +4 | ~85 行实现 + 70 行测试 | `cargo build --all-targets` 编译通过，4 个测试通过 |
|
||||
| 11.2 | ~`src/llm/provider/openai.rs`（+6 个测试），~`src/llm/provider/anthropic.rs`（+2 个测试） | +12（7 P0 + 5 P1） | ~290 行（含 test mod 和 mock 数据） | `cargo test --all-targets` 全绿，wiremock 12 场景均绿 |
|
||||
| 11.3 | ~`src/memory/store/in_memory.rs`（+3 个测试），~`src/memory/store/sqlite_store.rs`（+2 个测试） | +5 | ~120 行 | `cargo test --all-targets` 全绿 |
|
||||
| **总计** | 6 个文件（1 新增 + 5 修改） | +21 | ~500 行 | 全量 254 → 275+，`cargo clippy --all-targets -- -D warnings` 0 警告 |
|
||||
|
||||
### 验证通过标准
|
||||
|
||||
1. `cargo build --all-targets` —— 编译通过，无 warning
|
||||
2. `cargo test --all-targets` —— 全部通过（254 + 21 = 275+）
|
||||
3. `cargo clippy --all-targets -- -D warnings` —— 0 警告
|
||||
4. `cargo test --all-targets 2>&1 | grep -E "test result:"` —— 确认新增测试全部出现在执行列表中
|
||||
5. 新增 wiremock 测试单独验证网络隔离（无需 API key，纯本地 mock）
|
||||
|
||||
## 参考来源
|
||||
|
||||
- **讨论收口结论**：Phase 11 讨论，含 PM/SA 双视角输入（2026-07-07）
|
||||
- **现有代码模式**：
|
||||
- `src/llm/provider/openai.rs` line 824-1034——wiremock 测试模式（`MockServer::start` → `Mock::given(...).and(...).respond_with(...)` → `provider.chat_blocking` → assert）
|
||||
- `src/llm/provider/anthropic.rs` line 899-1079——Anthropic provider wiremock 测试
|
||||
- `src/memory/store/sqlite_store.rs` line 458-483——`concurrent_writers_no_data_loss` 10×10 并发模式
|
||||
- `src/memory/store.rs`——`MemoryStore` trait 定义（`#[async_trait]` 风格）
|
||||
- `src/memory/error.rs`——`MemoryError` 枚举（`#[non_exhaustive]` + `RetrievalError` 变体）
|
||||
- `src/memory/retriever.rs`——现有检索模块（`RetrievalResult` / `ScoredItem`）
|
||||
- `src/memory.rs`——模块根与 re-export 模式
|
||||
- **方案文档**：`docs/roadmap.md` Phase 11 章节（line 516-528）
|
||||
- **编译器 pragma**：`#[non_exhaustive]` —— 新增枚举变体需要此标记，公开结构体字段未来变化预留兼容空间
|
||||
|
||||
## 关键假设与风险
|
||||
|
||||
### 关键假设清单
|
||||
|
||||
| # | 假设 | 影响 | 推翻后的应对 |
|
||||
|---|------|------|------------|
|
||||
| 1 | wiremock `body_partial_json` matcher 在 wiremock 0.6+ 中可用 | Step 11.2 测试 #1 的实现方式 | 改用 `body_json`（精确匹配）或 `body_string`（部分串匹配） |
|
||||
| 2 | `tokio::spawn` 100 task 并发写入 `Mutex<HashMap>` 无死锁 | Step 11.3 #1 InMemoryStore 并发 | 降低并发数（50）继续验证，或换 `tokio::sync::Mutex` |
|
||||
| 3 | SqliteStore 的 `busy_timeout=5000` 能容忍 100 并发写 | Step 11.3 #2 SqliteStore 并发 | 增加 `busy_timeout`（10s），或限制最大并发数 |
|
||||
| 4 | 新增测试不使用 wiremock 以外的未列在 dev-dependencies 中的依赖 | Phase 11 零新增外部依赖 | 若需要额外 matcher，评估后加入 dev-dependencies |
|
||||
| 5 | InMemoryVectorRetriever 的 O(n) 全量扫描在测试规模下 (<1000 向量) 性能可接受 | Step 11.1 测试通过 | 如果有竞态问题，改为 read/write lock（`RwLock<HashMap>`） |
|
||||
| 6 | `MemoryError::RetrievalError` 变体足够覆盖向量检索的索引/评分失败场景 | Step 11.1 错误映射 | 如果不够，可增加新的 `MemoryError` 变体 |
|
||||
| 7 | OpenAI 和 Anthropic 的结构化错误体格式在当前 SDK 版本中未变化 | Step 11.2 测试 #3/#4/#7/#10/#11（结构化错误解析断言） | 若 SDK 变更错误体格式，更新 mock body 和断言匹配新格式 |
|
||||
| 8 | 调用方传入 `search()` 的向量与已索引向量维度一致（引用实现不做维度校验，`dot()` 对不等长向量静默截断） | Step 11.1 InMemoryVectorRetriever 正确性 | 若须维度校验，在 `index()` 时记录维度并在 `search()` 时断言；引用实现维持零校验 |
|
||||
|
||||
### 已识别的风险
|
||||
|
||||
| 风险 | 等级 | 缓解措施 |
|
||||
|------|------|---------|
|
||||
| wiremock `body_partial_json` matcher 行为在版本升级后变化 | 低 | 限定 wiremock 版本范围（当前已在 Cargo.lock 中锁定）；P0 测试不依赖该 matcher |
|
||||
| 100 并发写暴露 SqliteStore 的 `spawn_blocking` 线程池瓶颈 | 中 | 观察 CI 执行时间；如果超时，降低并发到 50 或增加 `max_blocking_threads` |
|
||||
| InMemoryVectorRetriever 的 `Mutex` 锁争用导致测试 flaky | 低 | Mutex 不会死锁（单线程持有不 await），测试不依赖精确时序 |
|
||||
| 新增 wiremock 测试与现有测试冲突（端口占用） | 低 | `MockServer::start()` 自动选择随机端口，不冲突 |
|
||||
| `cargo test --all-targets` 执行时间增加 >30% | 低 | 预估 +21 个测试，增量约 8%（254→275），其中 wiremock 测试有网络 IO 但延迟 <10ms/个 |
|
||||
|
||||
### 非阻塞已知项
|
||||
|
||||
- **Ollama Provider** 是 OpenAI Compat，wiremock 测试继承 `GenericOpenaiProvider`，不单独新增
|
||||
- **DeepSeek / Qwen Provider** 同样通过 `GenericOpenaiProvider` 实现，继承测试
|
||||
- **Step 11.1 不消费到 AgentSession / ContextSlot**，留待 Phase 12 或 v0.3 做消费端集成
|
||||
- **Phase 11 完成后**，全量测试预计 275+，`cargo test --all-targets` 执行时间预计 < 60s
|
||||
|
||||
---
|
||||
|
||||
## 实施计划（附录）
|
||||
|
||||
**实施顺序**：11.1 → 11.2 ‖ 11.3（11.2 与 11.3 无文件冲突，可并行交付；11.2 优先因回归保护价值更高）。每步完成后运行 `cargo test --all-targets` + `cargo clippy --all-targets -- -D warnings` 验证无回归。
|
||||
|
||||
---
|
||||
|
||||
### Step 11.1 — VectorRetriever trait + InMemoryVectorRetriever（~160 行，4 测试）
|
||||
|
||||
**前置**：无。与 Step 11.2/11.3 可并行开发但优先交付。
|
||||
|
||||
| 任务 | 描述 | 文件 | 前置 | 工作量 | 风险 | 验收条件 |
|
||||
|------|------|------|------|--------|------|---------|
|
||||
| **11.1.1** | 创建 `memory/vector.rs`：定义 `VectorRetriever` trait + `InMemoryVectorRetriever` struct + `dot()` 辅助函数 | `src/memory/vector.rs` | 无 | S | 低 | `cargo build --all-targets` 编译通过 |
|
||||
| **11.1.2** | 实现 `InMemoryVectorRetriever`：`index()` — Mutex insert；`search()` — 全量余弦扫描 + 降序排列 + k 截断 | `src/memory/vector.rs` | 11.1.1 | S | 低 | trait 实现编译通过；`Mutex::lock()` 使用 `map_err` 处理 poison，不 panic |
|
||||
| **11.1.3** | 添加 4 个内联测试：`basic_index_and_search`、`search_empty_store`、`concurrent_index`、`concurrent_index_and_search` | `src/memory/vector.rs` (mod tests) | 11.1.2 | S | 低 | 4 测试全部通过 |
|
||||
| **11.1.4** | 修改 `src/memory.rs`：加 `pub mod vector;` + `pub use vector::{VectorRetriever, InMemoryVectorRetriever};` | `src/memory.rs` | 11.1.1 | S | 低 | `cargo build --all-targets` 无 warning |
|
||||
|
||||
**Step 验证**：
|
||||
```
|
||||
cargo test --all-targets # 254 + 4 = 258+ passed
|
||||
cargo clippy --all-targets -- -D warnings # 0 warning
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Step 11.2 — wiremock Provider roundtrip 测试（12 测试，P0=7 + P1=5）
|
||||
|
||||
**前置**：无。与 Step 11.1 无文件冲突，可并行。
|
||||
|
||||
**`openai.rs` 新增测试（8 个：P0=5 + P1=3）**：
|
||||
|
||||
| 任务 | 测试名 | 优先级 | 前置 | 工作量 | 风险 | Mock 模式 | 验收条件 |
|
||||
|------|--------|--------|------|--------|------|----------|---------|
|
||||
| **11.2.1** | `openai_request_body_format` | P0 | 无 | S | 低 | `body_partial_json` 匹配 model/messages | 请求体结构正确，响应解析正常 |
|
||||
| **11.2.2** | `openai_authorization_header` | P0 | 无 | S | 低 | `header("authorization", "Bearer sk-test")` | header 精确匹配 |
|
||||
| **11.2.3** | `openai_401_structured_error` | P0 | 无 | S | 低 | 401 + `{"error":{"message":"...","code":"invalid_api_key"}}` | `LlmError::Authentication` 含 "Incorrect API key" |
|
||||
| **11.2.4** | `openai_429_with_retry_after` | P0 | 无 | S | 低 | 429 + `retry-after: 30` | `RateLimit { retry_after: Some(30s) }` |
|
||||
| **11.2.5** | `openai_tool_use_response` | P0 | 无 | S | 低 | 响应含 `tool_calls` | `StopReason::ToolUse` + `ContentBlock::ToolUse` 正确解析 |
|
||||
| **11.2.6** | `openai_stream_usage_only_last_chunk` | P1 | 无 | S | 低 | 流式最后 chunk `{choices:[], usage:{...}}` | 不 panic，`MessageComplete` 含正确 usage |
|
||||
| **11.2.7** | `openai_500_structured_error` | P1 | 无 | S | 低 | 500 + `{"error":{"message":"server error"}}` | `Request { status: 500 }` 含 body |
|
||||
| **11.2.8** | `openai_stream_mid_stream_error` | P1 | 无 | M | 中 | 前几个 chunk 正常后连接断开 | `LlmError::Request(_)` 流中断映射 |
|
||||
|
||||
**`anthropic.rs` 新增测试（4 个：P0=2 + P1=2）**：
|
||||
|
||||
| 任务 | 测试名 | 优先级 | 前置 | 工作量 | 风险 | Mock 模式 | 验收条件 |
|
||||
|------|--------|--------|------|--------|------|----------|---------|
|
||||
| **11.2.9** | `anthropic_401_structured_error` | P0 | 无 | S | 低 | 401 + `{"error":{"type":"authentication_error","message":"..."}}` | `LlmError::Authentication` 消息透传 |
|
||||
| **11.2.10** | `anthropic_tool_use_response` | P0 | 无 | S | 低 | 响应含 `type:"tool_use"` content block + `stop_reason:"tool_use"` | `StopReason::ToolUse` + `ContentBlock::ToolUse` 正确解析 |
|
||||
| **11.2.11** | `anthropic_version_header` | P1 | 无 | S | 低 | `header("anthropic-version", "2023-06-01")` | header 精确匹配 |
|
||||
| **11.2.12** | `anthropic_529_overloaded_structured` | P1 | 无 | S | 低 | 529 + `{"error":{"type":"overloaded_error","message":"Overloaded"}}` | `RateLimit { retry_after: None }` |
|
||||
|
||||
**Step 验证**：
|
||||
```
|
||||
cargo test --all-targets # 258 + 12 = 270+ passed
|
||||
cargo clippy --all-targets -- -D warnings # 0 warning
|
||||
```
|
||||
每个测试自包含（`MockServer::start()` → `Mock::given(...)` → `provider.chat_blocking()/chat_stream_inner()` → assert），无需共享 helper。
|
||||
|
||||
---
|
||||
|
||||
### Step 11.3 — 并发测试补强（5 测试）
|
||||
|
||||
**前置**：无。与 Step 11.1/11.2 无文件冲突。
|
||||
|
||||
| 任务 | 测试名 | 文件 | 前置 | 工作量 | 风险 | 模式 | 验收条件 |
|
||||
|------|--------|------|------|--------|------|------|---------|
|
||||
| **11.3.1** | `concurrent_writers_max_pressure` | `in_memory.rs` | 无 | S | 中 | 100 task × 1 write | 总量 100，id 无重复 |
|
||||
| **11.3.2** | `concurrent_writers_max_pressure` | `sqlite_store.rs` | 无 | S | 中 | 100 task × 1 write | 总量 100，id 无重复 |
|
||||
| **11.3.3** | `concurrent_mixed_read_write` | `in_memory.rs` | 无 | S | 中 | 预热 20 条，5 写 + 5 读并发 2 秒 | 无 panic |
|
||||
| **11.3.4** | `concurrent_mixed_read_write` | `sqlite_store.rs` | 无 | S | 中 | 预热 20 条，5 写 + 5 读并发 2 秒 | 无 panic |
|
||||
| **11.3.5** | `concurrent_capacity_eviction` | `in_memory.rs` | 无 | S | 中 | 15 写者，max_items=10 | 最终 ≤ 10（竞争激烈时可能过渡态 >10，主断言 ≤ 10，宽松备选 ≤ 15） |
|
||||
|
||||
**实现要点**：
|
||||
- 沿用现有 `Arc<Store> + tokio::spawn + h.await.unwrap()` 模式（参考 `sqlite_store.rs:458-483`）
|
||||
- `make_item` 辅助函数直接复用各文件现有实现
|
||||
- SqliteStore 测试使用 `:memory:` 数据库（与现有并发测试一致）
|
||||
- 混合读写测试使用 `tokio::time::Instant::now() + Duration` 做时限
|
||||
- 100 并发写是一次性 spawn 100 task（非分批），暴露最大锁竞争压力
|
||||
|
||||
**Step 验证**：
|
||||
```
|
||||
cargo test --all-targets # 270 + 5 = 275+ passed
|
||||
cargo clippy --all-targets -- -D warnings # 0 warning
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### 整体发布核查清单
|
||||
|
||||
| # | 检查项 | 验证命令 | 预期结果 |
|
||||
|---|--------|---------|---------|
|
||||
| 1 | 编译 | `cargo build --all-targets` | 通过，0 warning |
|
||||
| 2 | 全量测试 | `cargo test --all-targets` | 275+ passed，0 failed |
|
||||
| 3 | Lint | `cargo clippy --all-targets -- -D warnings` | 0 warning |
|
||||
| 4 | 文档 | `cargo doc --no-deps` | 0 warning（VectorRetriever trait 公共 API doc 完整） |
|
||||
| 5 | 确认新增测试 | `cargo test --all-targets 2>&1 \| grep -E "test result:"` | 所有新增测试名出现在执行列表中 |
|
||||
| 6 | wiremock 隔离 | 新增 wiremock 测试不依赖网络 | 纯本地 mock，无需 API key |
|
||||
| 7 | 并行安全 | 并发测试独立运行时无 flaky | 连续 3 次 `cargo test` 结果一致 |
|
||||
| 8 | 存量零回归 | 已有 254 个测试全部通过 | 与 Phase 10 基线对比无 fail |
|
||||
| 9 | 公共 API doc comment | `grep -r "pub trait VectorRetriever" src/ && rg "^///" -c src/memory/vector.rs` | trait 和方法都有 `///` 注释 |
|
||||
|
||||
**若核查项失败的回退策略**：
|
||||
- **测试失败（P0）**：阻断发布。定位到具体测试名 → 检查 Mock JSON 格式与 Provider 解析逻辑是否匹配（结构化错误体格式变化 → 更新 mock body；流式状态机变化 → 更新 `chat_stream_inner` 路径测试）
|
||||
- **测试失败（P1）**：不阻断发布。标记 `#[ignore]` + file issue，确认无 P0 失败后即可发布
|
||||
- **clippy warning**：修复 lint 后重跑；若为 `#[allow(...)]` 可抑制，在 code review 中申明理由
|
||||
- **flaky 并发测试**：检查 `tokio::spawn` 是否跨 `.await` 持锁；若 SqliteStore 超时，增加 `busy_timeout` 或降低并发数
|
||||
- **11.1 模块发布阻塞**：若 `InMemoryVectorRetriever` 无法按时交付，可临时注释 `src/memory.rs` 中的 `pub mod vector;` 行，跳过整个模块（零消费者，不影响发布）。回退后再补交
|
||||
@@ -0,0 +1,183 @@
|
||||
# LangChain & LangGraph 功能调研笔记
|
||||
|
||||
> 调研时间：2026-07-06
|
||||
> 两者关系：同一公司（LangChain Inc.）维护的堆栈上下两层，不是竞品
|
||||
|
||||
---
|
||||
|
||||
## 两者关系
|
||||
|
||||
```
|
||||
┌──────────────────────────────────────────┐
|
||||
│ LangChain (v1.0 GA) │ ← 高层框架：模型抽象、工具、提示词、600+集成
|
||||
│ create_agent / LCEL / 组件库 │
|
||||
├──────────────────────────────────────────┤
|
||||
│ LangGraph (v1.0 GA) │ ← 底层运行时：有向图执行引擎
|
||||
│ StateGraph / Checkpointing / HITL │
|
||||
├──────────────────────────────────────────┤
|
||||
│ LangSmith (可观测性) │
|
||||
└──────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
2025年10月22日同时达到 v1.0 GA，官方分工：
|
||||
|
||||
> **LangChain** = agent framework：abstractions and integrations for models, tools, and agent loops.
|
||||
> **LangGraph** = orchestration runtime：durable execution, streaming, human-in-the-loop, and persistence.
|
||||
|
||||
LangChain v1.0 的 `create_agent` 内部已运行在 LangGraph 引擎上。
|
||||
|
||||
---
|
||||
|
||||
## LangChain v1.0
|
||||
|
||||
### 定位
|
||||
高层应用框架，提供 agent 所需的**组件抽象**和**集成生态**。
|
||||
|
||||
### 精简后的核心模块
|
||||
|
||||
| 模块 | 功能 |
|
||||
|------|------|
|
||||
| `langchain.agents` | `create_agent`, `AgentState`（取代旧 AgentExecutor） |
|
||||
| `langchain.chat_models` | `init_chat_model`, `BaseChatModel`（统一模型初始化） |
|
||||
| `langchain.tools` | `@tool`, `BaseTool` |
|
||||
| `langchain.messages` | 消息类型、内容块、`trim_messages` |
|
||||
| `langchain.embeddings` | `init_embeddings`, `Embeddings` |
|
||||
|
||||
旧组件（`LLMChain`、`ConversationChain` 等）移入 `langchain-classic`。
|
||||
|
||||
### 七大组件类别
|
||||
|
||||
| 类别 | 关键组件 |
|
||||
|------|----------|
|
||||
| **Models** | Chat models, LLMs, Embeddings — 统一接口跨 provider 切换 |
|
||||
| **Tools** | 600+ provider 集成：API、数据库、搜索引擎等 |
|
||||
| **Agents** | `create_agent`, ReAct agents, Tool-calling agents |
|
||||
| **Memory** | 消息历史、自定义状态 |
|
||||
| **Retrievers** | 向量检索器、网络检索器 |
|
||||
| **Document** | 加载器、分割器、转换器 |
|
||||
| **Vector Stores** | Chroma, Pinecone, FAISS 等集成 |
|
||||
|
||||
### v1.0 关键新特性
|
||||
|
||||
**1. Middleware 中间件系统** — `create_agent` 的钩子系统：
|
||||
- `before_model` — 模型调用前注入/修改
|
||||
- `after_model` — 模型调用后验证/后处理
|
||||
- `wrap_tool_call` — 拦截工具调用错误
|
||||
|
||||
**2. Standard Message Content** — 跨 provider 标准化消息内容格式：
|
||||
- 推理/思维链、引用、多模态（图片/音视频/文档）
|
||||
- 工具调用、provider 特有工具（web search, code execution）
|
||||
- 通过 `.content_blocks` 属性访问，向后兼容
|
||||
|
||||
**3. `create_agent`** — 取代旧 AgentExecutor，内部运行在 LangGraph 运行时上
|
||||
|
||||
### 成熟度
|
||||
|
||||
| 维度 | 状态 |
|
||||
|------|------|
|
||||
| 版本 | v1.0 GA（2025-10） |
|
||||
| 稳定性 | 稳定，agent 层经重构后已稳定 |
|
||||
| 生产证明 | Replit, Clay, Rippling, Cloudflare, Workday |
|
||||
| 支持 | LTS-style support track |
|
||||
| 适用场景 | RAG、信息提取、单 agent 助手、快速原型 |
|
||||
|
||||
---
|
||||
|
||||
## LangGraph v1.0
|
||||
|
||||
### 定位
|
||||
底层编排运行时，专为**有状态、长时间运行、多步骤**工作流设计。
|
||||
|
||||
### 核心抽象链
|
||||
|
||||
```
|
||||
StateGraph → Nodes (纯 Python 函数) → Edges (路由逻辑)
|
||||
↓
|
||||
Shared State (TypedDict / Pydantic)
|
||||
↓
|
||||
Checkpointer (每个 super-step 快照)
|
||||
```
|
||||
|
||||
- **StateGraph**: 有状态图，参数化 State 类型
|
||||
- **Nodes**: 纯函数，`(State) → updates`
|
||||
- **Edges**: `add_conditional_edges`，支持循环/分支/合并
|
||||
- **State**: `TypedDict` 或 Pydantic，带 reducer 处理并发更新
|
||||
- **Reducers**: `add_messages` 等，自动处理追加 vs 覆盖
|
||||
|
||||
### 完整功能矩阵
|
||||
|
||||
| 功能 | 状态 | 细节 |
|
||||
|------|------|------|
|
||||
| **StateGraph** | ✅ 稳定 | 循环图（非 DAG），条件边缘，并行 fan-out |
|
||||
| **Checkpointing** | ✅ v4.1.1 | SQLite / PostgreSQL / Redis 后端 |
|
||||
| **Durable Execution** | ✅ 稳定 | 跨失败自动恢复，从精确断点继续 |
|
||||
| **Human-in-the-loop** | ✅ 一等公民 | `interrupt()` + `Command(resume=...)` |
|
||||
| **Time-travel 调试** | ✅ 稳定 | 回滚任意 checkpoint，fork 重放 |
|
||||
| **流式输出** | ✅ 稳定 | Token 级 + State 级 + Event 级 |
|
||||
| **多 Agent 编排** | ✅ 稳定 | Supervisor / Swarm / 层级 / Subgraph |
|
||||
| **Comprehensive Memory** | ✅ 稳定 | 短时工作记忆 + 长时持久记忆 |
|
||||
| **增量状态存储** | 🧪 DeltaChannel beta (v4.1.0+) | 长消息列表只存 delta |
|
||||
| **跨进程状态同步** | 🧪 RemoteCheckpointer (v4.1.0+) | 分布式多 agent 架构 |
|
||||
| **自动 checkpoint 清理** | ✅ keep_latest TTL (v4.0.2) | 避免无限制积累历史 |
|
||||
| **LangGraph Platform** | ✅ 稳定 | Agent Server：持久化、任务队列、版本管理 |
|
||||
| **LangGraph Studio** | ✅ 稳定 | 可视化 agent 工作流 |
|
||||
|
||||
### 成熟度
|
||||
|
||||
| 维度 | 状态 |
|
||||
|------|------|
|
||||
| 版本 | v1.0 GA（2025-10），checkpointer v4.1.1 (2026-05) |
|
||||
| 稳定性 | 高，持久化为架构一等公民 |
|
||||
| 生产证明 | Klarna, Replit, Elastic |
|
||||
| 支持 | LTS-style support track |
|
||||
| 适用场景 | 多步骤 agent、多 agent 系统、人工审批、长时间运行任务 |
|
||||
|
||||
---
|
||||
|
||||
## 功能边界对比
|
||||
|
||||
| 维度 | LangChain | LangGraph |
|
||||
|------|-----------|-----------|
|
||||
| **层次** | 高层应用框架 | 底层编排运行时 |
|
||||
| **核心抽象** | `create_agent`, LCEL, 组件库 | `StateGraph`, Nodes, Edges, State |
|
||||
| **思维模型** | 线性或 DAG 管道 | 节点 + 边缘的循环有向图 |
|
||||
| **循环/分支** | 受限 | **一等公民**：任意循环、分支、合并 |
|
||||
| **状态持久化** | 无原生支持 | **一等公民**：Checkpointer |
|
||||
| **Human-in-loop** | 需手动编排 | **一等公民**：`interrupt()` + `Command` |
|
||||
| **Time-travel 调试** | 无 | **一等公民**：回滚 fork 重放 |
|
||||
| **Durable Execution** | 无 | **一等公民**：跨故障自动恢复 |
|
||||
| **流式** | Token 级 | Token + State + Event 每节点流式 |
|
||||
| **多 Agent 编排** | 需手动组合 | **原生**：Supervisor/Swarm/Subgraph |
|
||||
| **模型抽象** | **核心优势** | 复用 LangChain |
|
||||
| **600+ 集成** | **核心优势** | 可复用 LangChain 集成 |
|
||||
| **LCEL 线性链** | **有** | 无 |
|
||||
| **Middleware** | **v1.0 特有** | 无 |
|
||||
| **学习曲线** | 中等 | 较陡（需图思维） |
|
||||
| **部署平台** | 无独立平台 | LangGraph Platform + Studio |
|
||||
|
||||
---
|
||||
|
||||
## 决策路线
|
||||
|
||||
```
|
||||
你的 workflow 需要什么？
|
||||
│
|
||||
├─ 线性、始终相同步骤 → LangChain (LCEL / create_agent)
|
||||
├─ 需要循环/分支/重试 → LangGraph (StateGraph)
|
||||
├─ 需要持久化/故障恢复 → LangGraph (Checkpointer)
|
||||
├─ 需要人工审批 → LangGraph (interrupt())
|
||||
├─ 需要 time-travel 调试 → LangGraph (checkpoint + fork)
|
||||
├─ 需要多 agent 协作 → LangGraph (Supervisor/Swarm/Subgraph)
|
||||
└─ 不确定 → 先用 create_agent，遇到瓶颈下钻到 StateGraph
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 参考来源
|
||||
|
||||
- [LangChain Blog: v1.0 Milestone](https://www.langchain.com/blog/langchain-langgraph-1dot0)
|
||||
- [LangChain Documentation](https://docs.langchain.com/oss/python/langchain/overview)
|
||||
- [LangGraph Documentation](https://docs.langchain.com/oss/python/langgraph/overview)
|
||||
- [LangGraph GitHub](https://github.com/langchain-ai/langgraph)
|
||||
- [Atlan: LangChain vs LangGraph 2026](https://atlan.com/know/ai-agent/ai-agent-memory/langchain-vs-langgraph/)
|
||||
- [truefoundry: LangChain vs LangGraph](https://www.truefoundry.com/blog/langchain-vs-langgraph)
|
||||
+29
-13
@@ -1,13 +1,13 @@
|
||||
# AG Core Roadmap
|
||||
|
||||
> 定稿日期：2026-05-11
|
||||
> 最后更新：2026-07-07
|
||||
> 最后更新：2026-07-06
|
||||
|
||||
## 愿景
|
||||
|
||||
AG Core 定位为构建 AI 智能体的底层工具箱，通过模块化、可插拔的架构，提供大模型调用、提示词工程、工具系统、记忆检索四大核心能力，支持快速组合出符合业务需求的智能体应用。
|
||||
|
||||
**当前状态**：v0.1.0 已发布（2026-07-04）。Phase 0-10 全部完成，v0.2.0-rc.1 已打标签。Provider IR 重构 + LlmCycle 简化 + 11 个离线示例（含 `quick_start` 30 行最小示例、`end_to_end` 完整集成示例、`context_slot_demo` 分支对话示例）+ SqliteStore 持久化 + 14 个公开枚举 `#[non_exhaustive]` 护栏 + `StepStatus` IR 迁移 + `submit_turn_stream` 流式体验 + ContextSlot 多上下文分区管理已交付。下一步进入 Phase 11（测试与检索补强）。
|
||||
**当前状态**：v0.1.0 已发布（2026-07-04）。Phase 0-11 全部完成，v0.2.0-rc.1 已打标签。Provider IR 重构 + LlmCycle 简化 + 11 个离线示例（含 `quick_start` 30 行最小示例、`end_to_end` 完整集成示例、`context_slot_demo` 分支对话示例）+ SqliteStore 持久化 + 14 个公开枚举 `#[non_exhaustive]` 护栏 + `StepStatus` IR 迁移 + `submit_turn_stream` 流式体验 + ContextSlot 多上下文分区管理 + `VectorRetriever` 语义检索 trait + 12 个 wiremock Provider roundtrip 测试 + 5 个并发测试已交付。下一步进入 v0.2.0 正式版打 tag 流程。
|
||||
|
||||
---
|
||||
|
||||
@@ -519,12 +519,27 @@ pub struct ContextBudget { system, history, tools, tool_results, reserve }
|
||||
|
||||
| Step | 内容 | 验证标准 |
|
||||
|------|------|---------|
|
||||
| **11.1** | `VectorRetriever` trait：`index(id, embeddings)` + `search(query, k)` | 编译 + mock 测试 |
|
||||
| **11.2** | wiremock Provider roundtrip 测试：模拟 OpenAI/Anthropic HTTP 端点 | `cargo test` 新增 10+ roundtrip 测试 |
|
||||
| **11.3** | 并发测试补强：InMemoryStore + SqliteStore 多线程写入验证 | 跑 100 轮无 race |
|
||||
| **11.1** ✅ | `VectorRetriever` trait：`index(id, embeddings)` + `search(query, k)` | 编译 + mock 测试 |
|
||||
| **11.2** ✅ | wiremock Provider roundtrip 测试：模拟 OpenAI/Anthropic HTTP 端点 | `cargo test` 新增 10+ roundtrip 测试 |
|
||||
| **11.3** ✅ | 并发测试补强：InMemoryStore + SqliteStore 多线程写入验证 | 跑 100 轮无 race |
|
||||
|
||||
**依赖**：无（可随时做）
|
||||
**优先级**：P1
|
||||
**实际新增**（2026-07-06 commit `71abe88` / `b4e5c7d`，详见 `docs/18-phase11-testing-and-retrieval.md`）：
|
||||
- 方案文档：`docs/18-phase11-testing-and-retrieval.md`（647 行，含 11.1/11.2/11.3 设计 + 10 项架构决策 + 实施后补充 2 条偏差记录 #6 mid-stream mock 模式 + #7 429 retry-after 修复）
|
||||
- 新增文件 1 个：`src/memory/vector.rs`（237 行 — `VectorRetriever` trait + `InMemoryVectorRetriever` 引用实现 + `dot()` 零依赖 + 6 个内联测试）
|
||||
- 修改文件 5 个：
|
||||
- `src/memory.rs`（+2 行：module 声明 + re-export）
|
||||
- `src/llm/provider/openai.rs`（+8 wiremock 测试 + `handle_error_response` 429 retry-after 解析修复 5 行）
|
||||
- `src/llm/provider/anthropic.rs`（+4 wiremock 测试）
|
||||
- `src/memory/store/in_memory.rs`（+3 并发测试：100 并发写、5 写+5 读混合、15 写者容量淘汰）
|
||||
- `src/memory/store/sqlite_store.rs`（+2 并发测试：100 并发写、5 写+5 读混合）
|
||||
- 关键设计：
|
||||
- **零依赖 dot()**：手写点积/范数，零新增 crate 依赖
|
||||
- **Wiremock 测试自包含**：每个测试独立 `MockServer::start()`，沿用现有模式
|
||||
- **429 retry-after 修复**：`openai.rs` 与 `anthropic.rs` 行为对齐（5 行代码）
|
||||
- **偏差记录**：方案文档「已否决的方案 #6/#7」记录两处实施偏差，便于后续审计追溯
|
||||
- 验证：254 → 277 测试（+23 个新测试），clippy 0 警告，doc 0 warning；并发测试连续 3 次运行稳定无 flaky
|
||||
- **依赖**：无（与方案一致）
|
||||
- **状态**：✅ Phase 11 全部交付物已完成
|
||||
|
||||
---
|
||||
|
||||
@@ -552,7 +567,7 @@ graph BT
|
||||
P8["<b>Phase 8: MVP 出口</b><br/>rc.1 标签<br/>14 枚举 #[non_exhaustive]<br/>StepStatus IR 迁移<br/>quick_start + end_to_end"]:::done
|
||||
P9["<b>Phase 9: 流式体验增强</b><br/>submit_turn_stream<br/>submit_with_tools_stream<br/>9 单元测试 + 2 集成测试"]:::done
|
||||
P10["<b>Phase 10: ContextSlot</b><br/>ContextSlot 类型<br/>JSON blob 持久化<br/>AgentSession 集成<br/>43 个新测试"]:::done
|
||||
P11["Phase 11<br/>测试与检索"]:::p1
|
||||
P11["<b>Phase 11: 测试与检索补强</b><br/>VectorRetriever trait<br/>12 wiremock tests<br/>5 并发测试"]:::done
|
||||
P12["Phase 12<br/>P2 锦上添花"]:::p2
|
||||
|
||||
P8 --> P5
|
||||
@@ -586,7 +601,7 @@ graph BT
|
||||
| **M4** | **Phase 8 (rc.1)** | P0 五项全部交付，`cargo run --example quick_start` 跑通 | ✅ 2026-07-05 |
|
||||
| **M5** | Phase 9 | `submit_turn_stream` 流式事件序列验证通过 | ✅ 2026-07-06 |
|
||||
| **M6** | Phase 10 | ContextSlot 创建/切换/派生集成测试通过 | ✅ 2026-07-07 |
|
||||
| **M7** | Phase 11 | wiremock + 并发测试补强，测试总量 200+ | ⏳ |
|
||||
| **M7** | Phase 11 | wiremock + 并发测试补强，测试总量 200+ | ✅ 2026-07-06 |
|
||||
| **M8** | Phase 12（可选） | P2 功能按需交付 | ⏳ |
|
||||
|
||||
---
|
||||
@@ -630,10 +645,10 @@ graph BT
|
||||
|
||||
## 下一步行动
|
||||
|
||||
1. **Phase 11 启动**：测试与检索补强（`VectorRetriever` trait + wiremock Provider roundtrip + 并发写入验证），P1 功能
|
||||
2. **示例先行**：每完成一个 Phase 立即更新对应示例，验证通过后再合入
|
||||
3. **里程碑追踪**：以 Phase 10（ContextSlot，已完成）为最新节点，逐 Phase 验收
|
||||
4. **v0.2.0 正式版**：Phase 8-11 全部完成后，去掉 rc 后缀打 `v0.2.0` 正式版
|
||||
1. **v0.2.0 正式版打 tag**：Phase 8-11 全部完成，去掉 rc 后缀打 `v0.2.0` 正式版标签；CHANGELOG 整理 + Cargo.toml version 0.2.0-rc.1 → 0.2.0
|
||||
2. **Phase 12 评估**（可选）：P2 锦上添花三项（文件系统 MemoryStore / MCP StreamableHttp / Gemini Provider）按需选做
|
||||
3. **示例先行**：v0.2 范围内每完成一个 Phase 立即更新对应示例，验证通过后再合入
|
||||
4. **里程碑追踪**：以 Phase 11（已完成，2026-07-06）为最新节点，逐 Phase 验收
|
||||
|
||||
**已完成 / 进行中阶段**：
|
||||
- ✅ Phase 0 Foundation — 全部交付物已完成
|
||||
@@ -649,6 +664,7 @@ graph BT
|
||||
- ✅ **Phase 8 MVP 集成出口** — 14 个公开枚举追加 `#[non_exhaustive]`（P0 核心 IR + P0 Error + P1 其他） + `StepStatus::Completed(ChatResponse)` → `Completed(MessageResponse)` 迁移 + CHANGELOG v0.2.0-rc.1 + 2 个新示例（`quick_start` 60 行 + `end_to_end` 246 行），10 个离线示例全部 exit 0；**v0.2.0-rc.1 标签已打**；实施后三方审查发现 6 项问题（1 🔴 + 2 🟡 + 3 💭）已全部修复
|
||||
- ✅ **Phase 9 流式体验增强** — `AgentSession::submit_turn_stream` 流式事件序列 + `LlmCycle::submit_with_tools_stream` spawn + mpsc 状态机 + `StreamEvent::ToolExecutionStarted`/`Completed` 新变体 + 9 单元测试 + 2 集成测试（含 `submit_turn_stream_end_to_end` 端到端 mock 验证 + `submit_turn_stream_triggers_turn_hooks` Hook 触发验证），全量 200 → 211；`CycleConfig` 加 `Clone` derive；方案文档 `docs/16-phase9-streaming-experience.md`（821 行）
|
||||
- ✅ **Phase 10 ContextSlot 上下文管理** — `src/agent/context.rs` 新增 `ContextSlot` 核心类型（Full / Focused / Readonly 三种模式，New / Derived / Static 三种来源）+ JSON blob 批次持久化（每 slot 3-4 条 MemoryItem，`slot_config` key 自恢复支持旧版本兼容）；`AgentSession` 扩展 slots 字段 + 5 个管理方法（`create_slot` / `switch_slot` / `list_slots` / `derive_slot` / `delete_slot`，自动创建 `"default"` slot，`delete_slot` 双重保护禁止删 default/最后一个）；`submit_turn`/`finalize_turn` 改造为基于当前 slot 的增量追加写回（`cycle.messages()[input_len..]` 提取本轮新增消息，确保 Focused 模式"读时过滤"语义不丢失数据）；`finalize_turn` 签名变更（新增 `new_messages_from_cycle: Vec<Message>` 参数，返回 `Result<(), AgentError>`）；`agent/error.rs` 新增 3 个 Slot 错误变体（`SlotReadonly` / `SlotNotFound` / `SlotAlreadyExists`）；`examples/context_slot_demo.rs` 新增分支对话示例（法律咨询入口 → 两个派生方向 → 切换 → 隔离验证 → 删除保护）；方案文档 `docs/17-phase10-contextslot.md`（1227 行，含 §5 推荐方案、§6 实施建议、§9 实施计划，经过 4 轮方案/计划/实施审查 + 1 轮非阻塞建议修复）；全量 211 → 254（+43 新测试），clippy 0 警告，doc 0 warning，11 个离线示例全部 exit 0
|
||||
- ✅ **Phase 11 测试与检索补强** — `src/memory/vector.rs` 新增 `VectorRetriever` trait（index + search 抽象）+ `InMemoryVectorRetriever` 引用实现（HashMap + 全量余弦相似度扫描 + 零依赖 `dot()`），6 个内联测试覆盖 basic/empty/zero-vector/k=0/2 个并发；wiremock Provider roundtrip 测试 12 个（OpenAI 8 + Anthropic 4）覆盖请求体/header/401/429/500/529/流式 usage-only/流式错误/ToolUse/结构化错误体；`MemoryStore` 并发测试 5 个（InMemoryStore 3 + SqliteStore 2）覆盖 100 并发写、5 写+5 读混合 2 秒、15 写者容量淘汰；`openai.rs` `handle_error_response` 修复 429 retry-after 解析（5 行，与 anthropic 对齐）；方案文档 `docs/18-phase11-testing-and-retrieval.md`（647 行，含 10 项架构决策 + 2 条实施偏差记录 #6 mid-stream mock 模式 + #7 retry-after 修复）；全量 254 → 277（+23 新测试），clippy 0 警告，doc 0 warning，并发测试 3 次稳定无 flaky
|
||||
- ✅ Provider IR 重构 — 统一类型系统 + OpenAI/Anthropic/DeepSeek/Qwen/Ollama 适配
|
||||
- ✅ LlmCycle 简化 — IR 消息类型切换 + Phase 0 桥接层移除
|
||||
- ✅ v0.1 Release — 技术债扫清、MockProvider 公开化、8 个离线示例（含 `simple_visit`）、README + 错误消息友好化、CHANGELOG 初始化
|
||||
|
||||
@@ -901,7 +901,7 @@ mod tests {
|
||||
use super::*;
|
||||
use crate::llm::types::request_v2::MessageRequest;
|
||||
use serde_json::json;
|
||||
use wiremock::matchers::{method, path};
|
||||
use wiremock::matchers::{header, method, path};
|
||||
use wiremock::{Mock, MockServer, ResponseTemplate};
|
||||
|
||||
fn make_provider(base_url: String) -> AnthropicProvider {
|
||||
@@ -1077,4 +1077,134 @@ event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n";
|
||||
assert_eq!(body.max_tokens, DEFAULT_MAX_TOKENS);
|
||||
assert_eq!(body.model, "claude-sonnet-4-20250514");
|
||||
}
|
||||
|
||||
// ===== Phase 11 Step 11.2 wiremock roundtrip 测试 =====
|
||||
|
||||
#[tokio::test]
|
||||
async fn anthropic_401_structured_error() {
|
||||
let server = MockServer::start().await;
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/v1/messages"))
|
||||
.respond_with(ResponseTemplate::new(401).set_body_json(json!({
|
||||
"type": "error",
|
||||
"error": {
|
||||
"type": "authentication_error",
|
||||
"message": "Invalid API key provided: sk-ant-test"
|
||||
}
|
||||
})))
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
let provider = make_provider(server.uri());
|
||||
let err = provider
|
||||
.chat_blocking(MessageRequest {
|
||||
model: "claude-sonnet-4-20250514".into(),
|
||||
messages: vec![Message::user_text("Hi")],
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap_err();
|
||||
match err {
|
||||
LlmError::Authentication(msg) => assert!(msg.contains("Invalid API key")),
|
||||
other => panic!("expected Authentication, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn anthropic_tool_use_response() {
|
||||
let server = MockServer::start().await;
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/v1/messages"))
|
||||
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
|
||||
"id": "msg_tool",
|
||||
"type": "message",
|
||||
"model": "claude-sonnet-4-20250514",
|
||||
"content": [
|
||||
{"type": "text", "text": "Let me check."},
|
||||
{"type": "tool_use", "id": "toolu_abc", "name": "lookup", "input": {"q": "rust"}}
|
||||
],
|
||||
"stop_reason": "tool_use",
|
||||
"usage": {"input_tokens": 8, "output_tokens": 12}
|
||||
})))
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
let provider = make_provider(server.uri());
|
||||
let response = provider
|
||||
.chat_blocking(MessageRequest {
|
||||
model: "claude-sonnet-4-20250514".into(),
|
||||
messages: vec![Message::user_text("Look up rust")],
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(response.stop_reason, StopReason::ToolUse);
|
||||
let tool_use = match &response.message {
|
||||
Message::Assistant { content } => content.iter().find_map(|b| match b {
|
||||
ContentBlock::ToolUse { id, name, .. } => Some((id.clone(), name.clone())),
|
||||
_ => None,
|
||||
}),
|
||||
_ => None,
|
||||
};
|
||||
assert_eq!(tool_use, Some(("toolu_abc".into(), "lookup".into())));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn anthropic_version_header() {
|
||||
let server = MockServer::start().await;
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/v1/messages"))
|
||||
.and(header("anthropic-version", "2023-06-01"))
|
||||
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
|
||||
"id": "msg_v",
|
||||
"type": "message",
|
||||
"model": "claude-sonnet-4-20250514",
|
||||
"content": [{"type": "text", "text": "OK"}],
|
||||
"stop_reason": "end_turn",
|
||||
"usage": {"input_tokens": 1, "output_tokens": 1}
|
||||
})))
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
let provider = make_provider(server.uri());
|
||||
let response = provider
|
||||
.chat_blocking(MessageRequest {
|
||||
model: "claude-sonnet-4-20250514".into(),
|
||||
messages: vec![Message::user_text("Hi")],
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(response.text(), "OK");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn anthropic_529_overloaded_structured() {
|
||||
let server = MockServer::start().await;
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/v1/messages"))
|
||||
.respond_with(ResponseTemplate::new(529).set_body_json(json!({
|
||||
"type": "error",
|
||||
"error": {
|
||||
"type": "overloaded_error",
|
||||
"message": "Overloaded: Anthropic API is temporarily overloaded"
|
||||
}
|
||||
})))
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
let provider = make_provider(server.uri());
|
||||
let err = provider
|
||||
.chat_blocking(MessageRequest {
|
||||
model: "claude-sonnet-4-20250514".into(),
|
||||
messages: vec![Message::user_text("Hi")],
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap_err();
|
||||
match err {
|
||||
LlmError::RateLimit { retry_after } => assert!(retry_after.is_none()),
|
||||
other => panic!("expected RateLimit, got {other:?}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+363
-6
@@ -180,15 +180,17 @@ impl GenericOpenaiProvider {
|
||||
/// （无法解析为 JSON），此处直接用 status code + 原始 body 兜底。
|
||||
async fn handle_error_response(response: reqwest::Response) -> LlmError {
|
||||
let status = response.status().as_u16();
|
||||
let retry_after = response
|
||||
.headers()
|
||||
.get("retry-after")
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.and_then(|v| v.parse::<u64>().ok())
|
||||
.map(std::time::Duration::from_secs);
|
||||
let body = response.text().await.unwrap_or_default();
|
||||
|
||||
match status {
|
||||
401 => LlmError::Authentication(body),
|
||||
429 => {
|
||||
// ponytail: 仅读取 retry-after，不在 OpenAI-compatible 上假设格式
|
||||
// 与 OpenAI 完全一致；DeepSeek/Qwen 通常遵循。
|
||||
LlmError::RateLimit { retry_after: None }
|
||||
}
|
||||
429 => LlmError::RateLimit { retry_after },
|
||||
_ if status >= 500 => LlmError::Request { status, body },
|
||||
_ if status == 400 && body.contains("context_length_exceeded") => {
|
||||
LlmError::ContextLength {
|
||||
@@ -818,7 +820,8 @@ mod tests {
|
||||
use crate::llm::convert::content_to_blocks;
|
||||
use crate::llm::types::usage::Usage;
|
||||
use serde_json::json;
|
||||
use wiremock::matchers::{method, path};
|
||||
use std::time::Duration;
|
||||
use wiremock::matchers::{body_partial_json, header, method, path};
|
||||
use wiremock::{Mock, MockServer, ResponseTemplate};
|
||||
|
||||
#[tokio::test]
|
||||
@@ -1031,4 +1034,358 @@ data: [DONE]\n\n";
|
||||
assert_eq!(blocks.len(), 1);
|
||||
assert!(matches!(blocks[0], ContentBlock::Text { ref text } if text == "plain text"));
|
||||
}
|
||||
|
||||
// ===== Phase 11 Step 11.2 wiremock roundtrip 测试 =====
|
||||
|
||||
#[tokio::test]
|
||||
async fn openai_request_body_format() {
|
||||
let server = MockServer::start().await;
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/chat/completions"))
|
||||
.and(body_partial_json(json!({
|
||||
"model": "gpt-4o",
|
||||
"messages": [{"role": "user", "content": "hi"}]
|
||||
})))
|
||||
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
|
||||
"id": "chatcmpl-body",
|
||||
"object": "chat.completion",
|
||||
"created": 1,
|
||||
"model": "gpt-4o",
|
||||
"choices": [{"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}],
|
||||
"usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}
|
||||
})))
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
let provider = GenericOpenaiProvider::new_with_name(
|
||||
server.uri(),
|
||||
"sk-test".into(),
|
||||
"gpt-4o".into(),
|
||||
"openai",
|
||||
30,
|
||||
);
|
||||
let response = provider
|
||||
.chat_blocking(MessageRequest {
|
||||
model: "gpt-4o".into(),
|
||||
messages: vec![Message::user_text("hi")],
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(response.text(), "ok");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn openai_authorization_header() {
|
||||
let server = MockServer::start().await;
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/chat/completions"))
|
||||
.and(header("authorization", "Bearer sk-test"))
|
||||
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
|
||||
"id": "chatcmpl-hdr",
|
||||
"object": "chat.completion",
|
||||
"created": 1,
|
||||
"model": "gpt-4o",
|
||||
"choices": [{"index": 0, "message": {"role": "assistant", "content": "OK"}, "finish_reason": "stop"}],
|
||||
"usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}
|
||||
})))
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
let provider = GenericOpenaiProvider::new_with_name(
|
||||
server.uri(),
|
||||
"sk-test".into(),
|
||||
"gpt-4o".into(),
|
||||
"openai",
|
||||
30,
|
||||
);
|
||||
let response = provider
|
||||
.chat_blocking(MessageRequest {
|
||||
model: "gpt-4o".into(),
|
||||
messages: vec![Message::user_text("hi")],
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(response.text(), "OK");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn openai_401_structured_error() {
|
||||
let server = MockServer::start().await;
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/chat/completions"))
|
||||
.respond_with(ResponseTemplate::new(401).set_body_json(json!({
|
||||
"error": {
|
||||
"message": "Incorrect API key provided: sk-test. You can find your API key at https://example.com",
|
||||
"type": "invalid_request_error",
|
||||
"code": "invalid_api_key"
|
||||
}
|
||||
})))
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
let provider = GenericOpenaiProvider::new_with_name(
|
||||
server.uri(),
|
||||
"sk-test".into(),
|
||||
"gpt-4o".into(),
|
||||
"openai",
|
||||
30,
|
||||
);
|
||||
let err = provider
|
||||
.chat_blocking(MessageRequest {
|
||||
model: "gpt-4o".into(),
|
||||
messages: vec![Message::user_text("hi")],
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap_err();
|
||||
match err {
|
||||
LlmError::Authentication(msg) => assert!(msg.contains("Incorrect API key")),
|
||||
other => panic!("expected Authentication, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn openai_429_with_retry_after() {
|
||||
let server = MockServer::start().await;
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/chat/completions"))
|
||||
.respond_with(
|
||||
ResponseTemplate::new(429)
|
||||
.insert_header("retry-after", "30")
|
||||
.set_body_json(json!({
|
||||
"error": {"message": "Rate limit reached", "type": "rate_limit_error"}
|
||||
})),
|
||||
)
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
let provider = GenericOpenaiProvider::new_with_name(
|
||||
server.uri(),
|
||||
"sk-test".into(),
|
||||
"gpt-4o".into(),
|
||||
"openai",
|
||||
30,
|
||||
);
|
||||
let err = provider
|
||||
.chat_blocking(MessageRequest {
|
||||
model: "gpt-4o".into(),
|
||||
messages: vec![Message::user_text("hi")],
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap_err();
|
||||
match err {
|
||||
LlmError::RateLimit { retry_after } => {
|
||||
assert_eq!(retry_after, Some(Duration::from_secs(30)));
|
||||
}
|
||||
other => panic!("expected RateLimit, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn openai_tool_use_response() {
|
||||
let server = MockServer::start().await;
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/chat/completions"))
|
||||
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
|
||||
"id": "chatcmpl-tool",
|
||||
"object": "chat.completion",
|
||||
"created": 1,
|
||||
"model": "gpt-4o",
|
||||
"choices": [{
|
||||
"index": 0,
|
||||
"message": {
|
||||
"role": "assistant",
|
||||
"content": "",
|
||||
"tool_calls": [{
|
||||
"id": "call_abc",
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "lookup",
|
||||
"arguments": "{\"q\":\"rust\"}"
|
||||
}
|
||||
}]
|
||||
},
|
||||
"finish_reason": "tool_calls"
|
||||
}],
|
||||
"usage": {"prompt_tokens": 5, "completion_tokens": 3, "total_tokens": 8}
|
||||
})))
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
let provider = GenericOpenaiProvider::new_with_name(
|
||||
server.uri(),
|
||||
"sk-test".into(),
|
||||
"gpt-4o".into(),
|
||||
"openai",
|
||||
30,
|
||||
);
|
||||
let response = provider
|
||||
.chat_blocking(MessageRequest {
|
||||
model: "gpt-4o".into(),
|
||||
messages: vec![Message::user_text("hi")],
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(response.stop_reason, StopReason::ToolUse);
|
||||
let tool_use = match &response.message {
|
||||
Message::Assistant { content } => content.iter().find_map(|b| match b {
|
||||
ContentBlock::ToolUse { id, name, .. } => Some((id.clone(), name.clone())),
|
||||
_ => None,
|
||||
}),
|
||||
_ => None,
|
||||
};
|
||||
assert_eq!(tool_use, Some(("call_abc".into(), "lookup".into())));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn openai_500_structured_error() {
|
||||
let server = MockServer::start().await;
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/chat/completions"))
|
||||
.respond_with(ResponseTemplate::new(500).set_body_json(json!({
|
||||
"error": {"message": "Internal server error", "type": "server_error", "code": null}
|
||||
})))
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
let provider = GenericOpenaiProvider::new_with_name(
|
||||
server.uri(),
|
||||
"sk-test".into(),
|
||||
"gpt-4o".into(),
|
||||
"openai",
|
||||
30,
|
||||
);
|
||||
let err = provider
|
||||
.chat_blocking(MessageRequest {
|
||||
model: "gpt-4o".into(),
|
||||
messages: vec![Message::user_text("hi")],
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap_err();
|
||||
match err {
|
||||
LlmError::Request { status, body } => {
|
||||
assert_eq!(status, 500);
|
||||
assert!(body.contains("Internal server error"));
|
||||
}
|
||||
other => panic!("expected Request(500), got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn openai_stream_usage_only_last_chunk() {
|
||||
let server = MockServer::start().await;
|
||||
|
||||
let sse = "data: {\"id\":\"c1\",\"object\":\"chat.completion.chunk\",\"created\":1,\"model\":\"gpt-4o\",\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"content\":\"Hi\"},\"finish_reason\":null}],\"usage\":null}\n\n\
|
||||
data: {\"id\":\"c1\",\"object\":\"chat.completion.chunk\",\"created\":1,\"model\":\"gpt-4o\",\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}],\"usage\":null}\n\n\
|
||||
data: {\"id\":\"c1\",\"object\":\"chat.completion.chunk\",\"created\":1,\"model\":\"gpt-4o\",\"choices\":[],\"usage\":{\"prompt_tokens\":5,\"completion_tokens\":2,\"total_tokens\":7}}\n\n\
|
||||
data: [DONE]\n\n";
|
||||
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/chat/completions"))
|
||||
.respond_with(
|
||||
ResponseTemplate::new(200)
|
||||
.insert_header("content-type", "text/event-stream")
|
||||
.set_body_string(sse),
|
||||
)
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
let provider = GenericOpenaiProvider::new_with_name(
|
||||
server.uri(),
|
||||
"sk-test".into(),
|
||||
"gpt-4o".into(),
|
||||
"openai",
|
||||
30,
|
||||
);
|
||||
let mut stream = provider
|
||||
.chat_stream_inner(MessageRequest {
|
||||
model: "gpt-4o".into(),
|
||||
messages: vec![Message::user_text("hi")],
|
||||
stream: true,
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
use futures_util::StreamExt;
|
||||
let mut collected: Vec<StreamEvent> = Vec::new();
|
||||
while let Some(ev) = stream.next().await {
|
||||
collected.push(ev.unwrap());
|
||||
}
|
||||
|
||||
let complete = collected
|
||||
.iter()
|
||||
.find_map(|e| match e {
|
||||
StreamEvent::MessageComplete { full_response } => Some(full_response.clone()),
|
||||
_ => None,
|
||||
})
|
||||
.expect("expected MessageComplete");
|
||||
assert_eq!(complete.text(), "Hi");
|
||||
assert_eq!(complete.usage.prompt_tokens, 5);
|
||||
assert_eq!(complete.usage.completion_tokens, 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn openai_stream_mid_stream_error() {
|
||||
let server = MockServer::start().await;
|
||||
|
||||
// 服务端返回 200 + SSE content-type 但 body 是畸形 JSON —— 模拟流中途发送错误载荷。
|
||||
// ChunkToEventStream 在 handle_chunk_json 时应产生 Error 事件而非 panic。
|
||||
let malformed_sse = "data: {not-valid-json}\n\ndata: [DONE]\n\n";
|
||||
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/chat/completions"))
|
||||
.respond_with(
|
||||
ResponseTemplate::new(200)
|
||||
.insert_header("content-type", "text/event-stream")
|
||||
.set_body_string(malformed_sse),
|
||||
)
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
let provider = GenericOpenaiProvider::new_with_name(
|
||||
server.uri(),
|
||||
"sk-test".into(),
|
||||
"gpt-4o".into(),
|
||||
"openai",
|
||||
30,
|
||||
);
|
||||
|
||||
let mut stream = provider
|
||||
.chat_stream_inner(MessageRequest {
|
||||
model: "gpt-4o".into(),
|
||||
messages: vec![Message::user_text("hi")],
|
||||
stream: true,
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
use futures_util::StreamExt;
|
||||
let mut saw_error_event = false;
|
||||
let mut completed_normally = false;
|
||||
while let Some(ev) = stream.next().await {
|
||||
match ev {
|
||||
Ok(StreamEvent::Error { .. }) => saw_error_event = true,
|
||||
Ok(StreamEvent::MessageComplete { .. }) => completed_normally = true,
|
||||
Err(_) => saw_error_event = true,
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
// 畸形 payload 必须被检测 —— 要么产出 Error 事件，要么最终消息完整事件标记异常。
|
||||
// 不允许流静默完成（既无 Error 也无 MessageComplete），那是 bug。
|
||||
assert!(
|
||||
saw_error_event || completed_normally,
|
||||
"malformed SSE payload neither errored nor completed normally"
|
||||
);
|
||||
assert!(
|
||||
saw_error_event,
|
||||
"expected an Error event for malformed SSE chunk"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -6,6 +6,7 @@ pub mod knowledge;
|
||||
pub mod retriever;
|
||||
pub mod store;
|
||||
pub mod types;
|
||||
pub mod vector;
|
||||
|
||||
// 高频类型（大多数下游需要）
|
||||
pub use conversation::{ConversationMemory, ConversationMemoryConfig};
|
||||
@@ -13,6 +14,7 @@ pub use error::MemoryError;
|
||||
pub use knowledge::KnowledgeStore;
|
||||
pub use retriever::MemoryRetriever;
|
||||
pub use store::{InMemoryStore, MemoryStore, SqliteStore};
|
||||
pub use vector::{InMemoryVectorRetriever, VectorRetriever};
|
||||
|
||||
// 低频类型（配置/高级使用）
|
||||
pub use conversation::MemoryStrategy;
|
||||
|
||||
@@ -263,4 +263,104 @@ mod tests {
|
||||
let list = store.list(&MemoryFilter::default()).await.unwrap();
|
||||
assert_eq!(list.len(), 100);
|
||||
}
|
||||
|
||||
// ===== Phase 11 Step 11.3 并发测试 =====
|
||||
|
||||
#[tokio::test]
|
||||
async fn concurrent_writers_max_pressure() {
|
||||
use std::sync::Arc;
|
||||
let store = Arc::new(InMemoryStore::new());
|
||||
|
||||
let mut handles = Vec::new();
|
||||
for i in 0..100 {
|
||||
let s = Arc::clone(&store);
|
||||
handles.push(tokio::spawn(async move {
|
||||
let id = format!("concurrent_{i}");
|
||||
s.save(make_item(&id)).await.unwrap();
|
||||
}));
|
||||
}
|
||||
for h in handles {
|
||||
h.await.unwrap();
|
||||
}
|
||||
|
||||
let list = store.list(&MemoryFilter::default()).await.unwrap();
|
||||
assert_eq!(list.len(), 100);
|
||||
let mut ids: Vec<String> = list.iter().map(|v| v.id.clone()).collect();
|
||||
ids.sort();
|
||||
ids.dedup();
|
||||
assert_eq!(ids.len(), 100);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn concurrent_mixed_read_write() {
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
let store = Arc::new(InMemoryStore::new());
|
||||
|
||||
// 预热 20 条
|
||||
for i in 0..20 {
|
||||
store.save(make_item(&format!("seed_{i}"))).await.unwrap();
|
||||
}
|
||||
|
||||
let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
|
||||
let mut handles = Vec::new();
|
||||
|
||||
// 5 个写者
|
||||
for w in 0..5 {
|
||||
let s = Arc::clone(&store);
|
||||
handles.push(tokio::spawn(async move {
|
||||
let mut i = 0;
|
||||
while tokio::time::Instant::now() < deadline {
|
||||
let id = format!("writer{w}_item{i}");
|
||||
s.save(make_item(&id)).await.unwrap();
|
||||
i += 1;
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
// 5 个读者
|
||||
for _ in 0..5 {
|
||||
let s = Arc::clone(&store);
|
||||
handles.push(tokio::spawn(async move {
|
||||
while tokio::time::Instant::now() < deadline {
|
||||
let _ = s.list(&MemoryFilter::default()).await.unwrap();
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
for h in handles {
|
||||
h.await.unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn concurrent_capacity_eviction() {
|
||||
use std::sync::Arc;
|
||||
|
||||
let eviction = EvictionConfig {
|
||||
policy: EvictionPolicy::Capacity { max_items: 10 },
|
||||
check_interval: 1,
|
||||
};
|
||||
let store = Arc::new(InMemoryStore::with_eviction(eviction));
|
||||
|
||||
let mut handles = Vec::new();
|
||||
for i in 0..15 {
|
||||
let s = Arc::clone(&store);
|
||||
handles.push(tokio::spawn(async move {
|
||||
s.save(make_item(&format!("item_{i}"))).await.unwrap();
|
||||
}));
|
||||
}
|
||||
for h in handles {
|
||||
h.await.unwrap();
|
||||
}
|
||||
|
||||
let list = store.list(&MemoryFilter::default()).await.unwrap();
|
||||
// 写者全部完成后必 ≤ max_items（部分路径上可能短暂 >10 但全部完成时应 ≤10）
|
||||
assert!(
|
||||
list.len() <= 10,
|
||||
"expected <= 10 items after all writers done, got {}",
|
||||
list.len()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -542,4 +542,82 @@ mod tests {
|
||||
assert!(store.get("x").await.unwrap().is_none());
|
||||
}
|
||||
}
|
||||
|
||||
// ===== Phase 11 Step 11.3 并发测试 =====
|
||||
|
||||
#[tokio::test]
|
||||
async fn concurrent_writers_max_pressure() {
|
||||
use std::time::Duration;
|
||||
let store = Arc::new(SqliteStore::open(":memory:").unwrap());
|
||||
|
||||
let mut handles = Vec::new();
|
||||
for i in 0..100 {
|
||||
let s = Arc::clone(&store);
|
||||
handles.push(tokio::spawn(async move {
|
||||
let id = format!("concurrent_{i}");
|
||||
// 设置每次 save 的 per-call timeout —— busy_timeout=5000ms 应足够
|
||||
match tokio::time::timeout(
|
||||
Duration::from_secs(10),
|
||||
s.save(make_item(&id)),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(res) => res.unwrap(),
|
||||
Err(_) => panic!("save({id}) timed out under 100-way concurrency"),
|
||||
}
|
||||
}));
|
||||
}
|
||||
for h in handles {
|
||||
h.await.unwrap();
|
||||
}
|
||||
|
||||
let list = store.list(&MemoryFilter::default()).await.unwrap();
|
||||
assert_eq!(list.len(), 100);
|
||||
let mut ids: Vec<String> = list.iter().map(|v| v.id.clone()).collect();
|
||||
ids.sort();
|
||||
ids.dedup();
|
||||
assert_eq!(ids.len(), 100);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn concurrent_mixed_read_write() {
|
||||
use std::time::Duration;
|
||||
|
||||
let store = Arc::new(SqliteStore::open(":memory:").unwrap());
|
||||
|
||||
// 预热 20 条
|
||||
for i in 0..20 {
|
||||
store.save(make_item(&format!("seed_{i}"))).await.unwrap();
|
||||
}
|
||||
|
||||
let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
|
||||
let mut handles = Vec::new();
|
||||
|
||||
// 5 个写者
|
||||
for w in 0..5 {
|
||||
let s = Arc::clone(&store);
|
||||
handles.push(tokio::spawn(async move {
|
||||
let mut i = 0;
|
||||
while tokio::time::Instant::now() < deadline {
|
||||
let id = format!("writer{w}_item{i}");
|
||||
s.save(make_item(&id)).await.unwrap();
|
||||
i += 1;
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
// 5 个读者
|
||||
for _ in 0..5 {
|
||||
let s = Arc::clone(&store);
|
||||
handles.push(tokio::spawn(async move {
|
||||
while tokio::time::Instant::now() < deadline {
|
||||
let _ = s.list(&MemoryFilter::default()).await.unwrap();
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
for h in handles {
|
||||
h.await.unwrap();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,238 @@
|
||||
//! 语义向量检索抽象。
|
||||
//!
|
||||
//! 提供 [`VectorRetriever`] trait 定义与进程内引用实现 [`InMemoryVectorRetriever`]。
|
||||
//! 下游可实现此 trait 以对接向量数据库（pgvector / qdrant / lancedb 等）。
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Mutex;
|
||||
|
||||
use async_trait::async_trait;
|
||||
|
||||
use crate::memory::error::MemoryError;
|
||||
|
||||
/// 语义向量检索器抽象接口。
|
||||
///
|
||||
/// 下游可实现此 trait 以对接向量数据库（pgvector / qdrant / lancedb 等）。
|
||||
/// 默认引用实现 [`InMemoryVectorRetriever`] 基于进程内 HashMap + 余弦相似度。
|
||||
///
|
||||
/// **稳定性**：实验性 API（v0.2.x），方法签名可能在 v0.3 中调整。
|
||||
/// 若未来需要 `remove()` / `clear()` 等方法，将在此 trait 中追加（带默认实现）。
|
||||
#[async_trait]
|
||||
pub trait VectorRetriever: Send + Sync {
|
||||
/// 将 `id` 对应的文本向量 `embeddings` 加入索引。
|
||||
///
|
||||
/// 重复调用同一 `id` 会覆盖已有向量。调用方负责保证 `embeddings` 维度
|
||||
/// 与已索引向量一致——本 trait 不做维度校验。
|
||||
async fn index(&self, id: String, embeddings: Vec<f32>) -> Result<(), MemoryError>;
|
||||
|
||||
/// 检索与 `query` 向量最相似的 `k` 条记录。
|
||||
///
|
||||
/// 返回 `Vec<(id, score)>`，按 score 降序排列，score ∈ [0.0, 1.0]
|
||||
/// （余弦相似度）。当 `k == 0`、索引为空或 query 为零向量时返回空 Vec。
|
||||
async fn search(
|
||||
&self,
|
||||
query: Vec<f32>,
|
||||
k: usize,
|
||||
) -> Result<Vec<(String, f32)>, MemoryError>;
|
||||
}
|
||||
|
||||
/// 进程内向量检索器 —— 基于 HashMap + 全量余弦相似度扫描。
|
||||
///
|
||||
/// 适用场景：单元测试、小规模验证（<10K 向量）。生产环境请对接真正的向量数据库。
|
||||
///
|
||||
/// **不保证**：
|
||||
/// - 不做向量维度校验（不同维度向量查询结果无意义但不 panic）
|
||||
/// - `search()` 是 O(n) 全量扫描，未做索引加速
|
||||
/// - 不保证高并发下查询时序与写入顺序一致
|
||||
pub struct InMemoryVectorRetriever {
|
||||
vectors: Mutex<HashMap<String, Vec<f32>>>,
|
||||
}
|
||||
|
||||
impl InMemoryVectorRetriever {
|
||||
/// 创建空检索器。
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
vectors: Mutex::new(HashMap::new()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for InMemoryVectorRetriever {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl VectorRetriever for InMemoryVectorRetriever {
|
||||
async fn index(&self, id: String, embeddings: Vec<f32>) -> Result<(), MemoryError> {
|
||||
let mut vectors = self
|
||||
.vectors
|
||||
.lock()
|
||||
.map_err(|e| MemoryError::RetrievalError(format!("lock poisoned: {e}")))?;
|
||||
vectors.insert(id, embeddings);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn search(
|
||||
&self,
|
||||
query: Vec<f32>,
|
||||
k: usize,
|
||||
) -> Result<Vec<(String, f32)>, MemoryError> {
|
||||
let vectors = self
|
||||
.vectors
|
||||
.lock()
|
||||
.map_err(|e| MemoryError::RetrievalError(format!("lock poisoned: {e}")))?;
|
||||
|
||||
if vectors.is_empty() || k == 0 {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
|
||||
let query_norm = dot(&query, &query).sqrt();
|
||||
if query_norm == 0.0 {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
|
||||
let mut scored: Vec<(String, f32)> = vectors
|
||||
.iter()
|
||||
.map(|(id, vec)| {
|
||||
let dot_product = dot(&query, vec);
|
||||
let vec_norm = dot(vec, vec).sqrt();
|
||||
let similarity = dot_product / (query_norm * vec_norm + 1e-10);
|
||||
(id.clone(), similarity)
|
||||
})
|
||||
.collect();
|
||||
|
||||
scored.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
|
||||
scored.truncate(k);
|
||||
Ok(scored)
|
||||
}
|
||||
}
|
||||
|
||||
/// 点积（手动循环，零依赖）。
|
||||
///
|
||||
/// 注意：`zip` 对不等长向量静默截断到较短者。引用实现不做维度校验，
|
||||
/// 调用方应确保 `a` 和 `b` 等长——不等长时结果无意义但不 panic。
|
||||
fn dot(a: &[f32], b: &[f32]) -> f32 {
|
||||
a.iter().zip(b.iter()).map(|(x, y)| x * y).sum()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
#[tokio::test]
|
||||
async fn basic_index_and_search() {
|
||||
let retriever = InMemoryVectorRetriever::new();
|
||||
retriever
|
||||
.index("rust".into(), vec![1.0, 0.0, 0.0])
|
||||
.await
|
||||
.unwrap();
|
||||
retriever
|
||||
.index("python".into(), vec![0.0, 1.0, 0.0])
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let results = retriever.search(vec![0.9, 0.1, 0.0], 2).await.unwrap();
|
||||
assert_eq!(results.len(), 2);
|
||||
assert_eq!(results[0].0, "rust");
|
||||
assert!(results[0].1 > results[1].1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_empty_store() {
|
||||
let retriever = InMemoryVectorRetriever::new();
|
||||
let results = retriever.search(vec![1.0, 0.0, 0.0], 5).await.unwrap();
|
||||
assert!(results.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_zero_vector_returns_empty() {
|
||||
let retriever = InMemoryVectorRetriever::new();
|
||||
retriever
|
||||
.index("a".into(), vec![1.0, 0.0, 0.0])
|
||||
.await
|
||||
.unwrap();
|
||||
let results = retriever.search(vec![0.0, 0.0, 0.0], 5).await.unwrap();
|
||||
assert!(results.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_with_k_zero_returns_empty() {
|
||||
let retriever = InMemoryVectorRetriever::new();
|
||||
retriever
|
||||
.index("a".into(), vec![1.0, 0.0, 0.0])
|
||||
.await
|
||||
.unwrap();
|
||||
let results = retriever.search(vec![1.0, 0.0, 0.0], 0).await.unwrap();
|
||||
assert!(results.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn concurrent_index() {
|
||||
let retriever = Arc::new(InMemoryVectorRetriever::new());
|
||||
let mut handles = Vec::new();
|
||||
for i in 0..10 {
|
||||
let r = Arc::clone(&retriever);
|
||||
handles.push(tokio::spawn(async move {
|
||||
r.index(format!("item_{i}"), vec![i as f32, 0.0, 0.0])
|
||||
.await
|
||||
.unwrap();
|
||||
}));
|
||||
}
|
||||
for h in handles {
|
||||
h.await.unwrap();
|
||||
}
|
||||
|
||||
let results = retriever.search(vec![1.0, 0.0, 0.0], 20).await.unwrap();
|
||||
assert_eq!(results.len(), 10);
|
||||
let mut ids: Vec<String> = results.iter().map(|(id, _)| id.clone()).collect();
|
||||
ids.sort();
|
||||
ids.dedup();
|
||||
assert_eq!(ids.len(), 10);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn concurrent_index_and_search() {
|
||||
let retriever = Arc::new(InMemoryVectorRetriever::new());
|
||||
|
||||
for i in 0..5 {
|
||||
retriever
|
||||
.index(format!("seed_{i}"), vec![i as f32, 0.0, 0.0])
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
|
||||
|
||||
let mut handles = Vec::new();
|
||||
|
||||
for w in 0..5 {
|
||||
let r = Arc::clone(&retriever);
|
||||
handles.push(tokio::spawn(async move {
|
||||
let mut i = 0;
|
||||
while tokio::time::Instant::now() < deadline {
|
||||
r.index(format!("writer{w}_{i}"), vec![i as f32, 0.0, 0.0])
|
||||
.await
|
||||
.unwrap();
|
||||
i += 1;
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
for _ in 0..5 {
|
||||
let r = Arc::clone(&retriever);
|
||||
handles.push(tokio::spawn(async move {
|
||||
while tokio::time::Instant::now() < deadline {
|
||||
let _ = r.search(vec![1.0, 0.0, 0.0], 3).await.unwrap();
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
for h in handles {
|
||||
h.await.unwrap();
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user