8 changed files with 327 additions and 294 deletions
@ -1,197 +0,0 @@ |
|||
# 消息流系统 (Message Stream System) |
|||
|
|||
## 概述 |
|||
|
|||
DeepSearcher 的消息流系统是一个新的消息传输机制，用于替代原来的日志传输方式。该系统支持三种类型的消息：`search`、`think` 和 `answer`，并提供了灵活的消息管理和传输功能。 |
|||
|
|||
## 消息类型 |
|||
|
|||
### 1. Search 消息 |
|||
- **类型**: `search` |
|||
- **用途**: 表示搜索相关的操作和状态 |
|||
- **示例**: |
|||
- "在向量数据库中搜索相关信息..." |
|||
- "找到5个相关文档片段" |
|||
- "搜索人工智能的定义..." |
|||
|
|||
### 2. Think 消息 |
|||
- **类型**: `think` |
|||
- **用途**: 表示思考和推理过程 |
|||
- **示例**: |
|||
- "开始处理查询: 什么是人工智能?" |
|||
- "分析搜索结果..." |
|||
- "生成子查询: 人工智能的定义、历史、应用" |
|||
|
|||
### 3. Answer 消息 |
|||
- **类型**: `answer` |
|||
- **用途**: 表示最终答案和结果 |
|||
- **示例**: |
|||
- "==== FINAL ANSWER====" |
|||
- "人工智能是计算机科学的一个分支..." |
|||
|
|||
## 核心组件 |
|||
|
|||
### MessageStream 类 |
|||
|
|||
主要的消息流管理器，提供以下功能： |
|||
|
|||
```python |
|||
from deepsearcher.utils.message_stream import MessageStream |
|||
|
|||
# 创建消息流实例 |
|||
message_stream = MessageStream() |
|||
|
|||
# 发送消息 |
|||
message_stream.send_search("搜索内容...") |
|||
message_stream.send_think("思考内容...") |
|||
message_stream.send_answer("答案内容...") |
|||
|
|||
# 获取消息 |
|||
messages = message_stream.get_messages() |
|||
messages_dict = message_stream.get_messages_as_dicts() |
|||
messages_json = message_stream.get_messages_as_json() |
|||
``` |
|||
|
|||
### 全局函数 |
|||
|
|||
为了方便使用，提供了全局函数： |
|||
|
|||
```python |
|||
from deepsearcher.utils.message_stream import send_search, send_think, send_answer |
|||
|
|||
# 直接发送消息 |
|||
send_search("搜索内容...") |
|||
send_think("思考内容...") |
|||
send_answer("答案内容...") |
|||
``` |
|||
|
|||
## API 接口 |
|||
|
|||
### 1. 获取消息 |
|||
``` |
|||
GET /messages/ |
|||
``` |
|||
|
|||
返回所有消息的列表： |
|||
```json |
|||
{ |
|||
"messages": [ |
|||
{ |
|||
"type": "search", |
|||
"content": "搜索内容...", |
|||
"timestamp": 1755043653.9606102, |
|||
"metadata": {} |
|||
} |
|||
] |
|||
} |
|||
``` |
|||
|
|||
### 2. 清空消息 |
|||
``` |
|||
POST /clear-messages/ |
|||
``` |
|||
|
|||
清空所有消息并返回成功状态： |
|||
```json |
|||
{ |
|||
"message": "Messages cleared successfully" |
|||
} |
|||
``` |
|||
|
|||
### 3. 查询接口（已更新） |
|||
``` |
|||
GET /query/?original_query=<query>&max_iter=<iterations> |
|||
``` |
|||
|
|||
现在返回结果包含消息流： |
|||
```json |
|||
{ |
|||
"result": "最终答案...", |
|||
"messages": [ |
|||
{ |
|||
"type": "search", |
|||
"content": "搜索内容...", |
|||
"timestamp": 1755043653.9606102, |
|||
"metadata": {} |
|||
} |
|||
] |
|||
} |
|||
``` |
|||
|
|||
## 前端集成 |
|||
|
|||
前端界面现在包含一个消息流显示区域，实时显示处理过程中的各种消息： |
|||
|
|||
### CSS 样式 |
|||
- `.message-search`: 搜索消息样式（蓝色边框） |
|||
- `.message-think`: 思考消息样式（黄色边框） |
|||
- `.message-answer`: 答案消息样式（绿色边框） |
|||
|
|||
### JavaScript 功能 |
|||
- `displayMessages(messages)`: 显示消息流 |
|||
- 自动滚动到最新消息 |
|||
- 时间戳显示 |
|||
|
|||
## 使用示例 |
|||
|
|||
### 后端使用 |
|||
|
|||
```python |
|||
from deepsearcher.utils.message_stream import send_search, send_think, send_answer |
|||
|
|||
# 在搜索过程中发送消息 |
|||
send_think("开始处理查询...") |
|||
send_search("在数据库中搜索...") |
|||
send_search("找到相关文档...") |
|||
send_think("分析结果...") |
|||
send_answer("最终答案...") |
|||
``` |
|||
|
|||
### 前端使用 |
|||
|
|||
```javascript |
|||
// 获取消息 |
|||
const response = await fetch('/query/?original_query=test&max_iter=3'); |
|||
const data = await response.json(); |
|||
|
|||
// 显示消息流 |
|||
if (data.messages && data.messages.length > 0) { |
|||
displayMessages(data.messages); |
|||
} |
|||
``` |
|||
|
|||
## 优势 |
|||
|
|||
1. **结构化数据**: 消息包含类型、内容、时间戳和元数据 |
|||
2. **类型安全**: 使用枚举确保消息类型的一致性 |
|||
3. **灵活传输**: 支持多种输出格式（字典、JSON） |
|||
4. **实时显示**: 前端可以实时显示处理过程 |
|||
5. **易于扩展**: 可以轻松添加新的消息类型和功能 |
|||
|
|||
## 迁移说明 |
|||
|
|||
从原来的日志系统迁移到新的消息流系统： |
|||
|
|||
### 原来的代码 |
|||
```python |
|||
log.color_print(f"<search> Search [{query}] in [{collection}]... </search>\n") |
|||
log.color_print(f"<think> Summarize answer... </think>\n") |
|||
log.color_print(f"<answer> Final answer... </answer>\n") |
|||
``` |
|||
|
|||
### 新的代码 |
|||
```python |
|||
send_search(f"Search [{query}] in [{collection}]...") |
|||
send_think("Summarize answer...") |
|||
send_answer("Final answer...") |
|||
``` |
|||
|
|||
## 测试 |
|||
|
|||
运行测试脚本验证系统功能： |
|||
|
|||
```bash |
|||
python test_message_stream.py |
|||
``` |
|||
|
|||
这将测试消息流的基本功能，包括消息发送、获取和格式化。 |
|||
@ -0,0 +1,180 @@ |
|||
# 简化消息格式 |
|||
|
|||
## 概述 |
|||
|
|||
为了简化前后端的消息通信，我们将原来的多种消息类型统一为简单的 `type` 和 `content` 格式。 |
|||
|
|||
## 消息格式 |
|||
|
|||
所有消息都遵循以下统一格式： |
|||
|
|||
```json |
|||
{ |
|||
"type": "消息类型", |
|||
"content": "消息内容", |
|||
"timestamp": 时间戳, |
|||
"metadata": {} |
|||
} |
|||
``` |
|||
|
|||
## 消息类型 |
|||
|
|||
### 1. start - 开始消息 |
|||
- **用途**: 表示查询或操作的开始 |
|||
- **示例**: |
|||
```json |
|||
{ |
|||
"type": "start", |
|||
"content": "开始处理查询: 请写一篇关于Milvus向量数据库的报告" |
|||
} |
|||
``` |
|||
|
|||
### 2. info - 信息消息 |
|||
- **用途**: 表示处理过程中的信息、状态更新、迭代信息等 |
|||
- **示例**: |
|||
```json |
|||
{ |
|||
"type": "info", |
|||
"content": "iteration 1: 正在搜索相关文档..." |
|||
} |
|||
``` |
|||
|
|||
### 3. answer - 答案消息 |
|||
- **用途**: 表示最终答案或重要结果 |
|||
- **示例**: |
|||
```json |
|||
{ |
|||
"type": "answer", |
|||
"content": "Milvus的详细报告: ..." |
|||
} |
|||
``` |
|||
|
|||
### 4. complete - 完成消息 |
|||
- **用途**: 表示操作完成 |
|||
- **示例**: |
|||
```json |
|||
{ |
|||
"type": "complete", |
|||
"content": "查询完成" |
|||
} |
|||
``` |
|||
|
|||
### 5. error - 错误消息 |
|||
- **用途**: 表示错误信息 |
|||
- **示例**: |
|||
```json |
|||
{ |
|||
"type": "error", |
|||
"content": "查询失败: 无法连接到数据库" |
|||
} |
|||
``` |
|||
|
|||
## 使用示例 |
|||
|
|||
### 后端使用 |
|||
|
|||
```python |
|||
from deepsearcher.utils.message_stream import ( |
|||
send_start, send_info, send_answer, send_complete, send_error |
|||
) |
|||
|
|||
# 发送开始消息 |
|||
send_start("开始处理查询: 请写一篇关于Milvus的报告") |
|||
|
|||
# 发送信息消息 |
|||
send_info("iteration 1: 正在搜索相关文档...") |
|||
send_info("找到5个相关文档片段") |
|||
|
|||
# 发送答案消息 |
|||
send_answer("Milvus的详细报告: ...") |
|||
|
|||
# 发送完成消息 |
|||
send_complete("查询完成") |
|||
|
|||
# 发送错误消息 |
|||
send_error("查询失败: 无法连接到数据库") |
|||
``` |
|||
|
|||
### 前端处理 |
|||
|
|||
```javascript |
|||
// 处理消息流 |
|||
function handleStreamMessage(data) { |
|||
const message = JSON.parse(data); |
|||
|
|||
switch (message.type) { |
|||
case 'start': |
|||
console.log('开始:', message.content); |
|||
break; |
|||
case 'info': |
|||
console.log('信息:', message.content); |
|||
break; |
|||
case 'answer': |
|||
console.log('答案:', message.content); |
|||
break; |
|||
case 'complete': |
|||
console.log('完成:', message.content); |
|||
break; |
|||
case 'error': |
|||
console.error('错误:', message.content); |
|||
break; |
|||
} |
|||
} |
|||
``` |
|||
|
|||
## 优势 |
|||
|
|||
1. **简化格式**: 统一的消息格式，易于理解和处理 |
|||
2. **类型清晰**: 5种基本类型覆盖所有使用场景 |
|||
3. **易于扩展**: 可以轻松添加新的消息类型 |
|||
4. **前端友好**: 前端处理逻辑更加简单 |
|||
5. **调试方便**: 消息格式清晰，便于调试和日志记录 |
|||
|
|||
## 迁移说明 |
|||
|
|||
### 从旧格式迁移 |
|||
|
|||
| 旧类型 | 新类型 | 说明 | |
|||
|--------|--------|------| |
|||
| `search` | `info` | 搜索相关信息 | |
|||
| `think` | `info` | 思考过程信息 | |
|||
| `answer` | `answer` | 保持不变 | |
|||
| `complete` | `complete` | 保持不变 | |
|||
| `query_start` | `start` | 查询开始 | |
|||
| `query_error` | `error` | 查询错误 | |
|||
| `stream_error` | `error` | 流错误 | |
|||
|
|||
### 代码更新 |
|||
|
|||
所有使用旧消息类型的代码都需要更新： |
|||
|
|||
```python |
|||
# 旧代码 |
|||
send_search("搜索内容...") |
|||
send_think("思考内容...") |
|||
|
|||
# 新代码 |
|||
send_info("搜索内容...") |
|||
send_info("思考内容...") |
|||
``` |
|||
|
|||
## 测试 |
|||
|
|||
可以使用以下方式测试消息格式： |
|||
|
|||
```python |
|||
from deepsearcher.utils.message_stream import get_message_stream |
|||
|
|||
# 获取消息流实例 |
|||
message_stream = get_message_stream() |
|||
|
|||
# 获取所有消息 |
|||
messages = message_stream.get_messages_as_dicts() |
|||
|
|||
# 验证格式 |
|||
for msg in messages: |
|||
assert 'type' in msg |
|||
assert 'content' in msg |
|||
assert 'timestamp' in msg |
|||
print(f"类型: {msg['type']}, 内容: {msg['content']}") |
|||
``` |
|||
@ -0,0 +1,38 @@ |
|||
#!/usr/bin/env python3 |
|||
""" |
|||
测试修复后的查询功能 |
|||
""" |
|||
|
|||
import sys |
|||
import os |
|||
sys.path.append(os.path.dirname(os.path.abspath(__file__))) |
|||
|
|||
from deepsearcher.configuration import Configuration, init_config |
|||
from deepsearcher.online_query import query |
|||
|
|||
def test_query_fix(): |
|||
"""测试修复后的查询功能""" |
|||
print("=== 测试修复后的查询功能 ===") |
|||
|
|||
# 初始化配置 |
|||
config = Configuration() |
|||
init_config(config) |
|||
|
|||
try: |
|||
print("开始查询...") |
|||
result_text, retrieval_results = query("什么是Milvus?", max_iter=1) |
|||
|
|||
print(f"查询完成!") |
|||
print(f"结果长度: {len(result_text) if result_text else 0}") |
|||
print(f"检索结果数量: {len(retrieval_results) if retrieval_results else 0}") |
|||
|
|||
if result_text: |
|||
print(f"结果预览: {result_text[:200]}...") |
|||
|
|||
except Exception as e: |
|||
import traceback |
|||
print(f"查询失败: {e}") |
|||
print(f"错误详情: {traceback.format_exc()}") |
|||
|
|||
if __name__ == "__main__": |
|||
test_query_fix() |
|||
Loading…
Reference in new issue