feat: protocol base lib + rule-based parser MVP + 31 tests

This commit is contained in:
root
2026-06-01 11:12:45 +08:00
parent 6edda94a89
commit ac663fca16
22 changed files with 1066 additions and 373 deletions

104
README.md
View File

@ -80,53 +80,93 @@ graph TB
所有子网间不传递自然语言,而是通过**统一张量协议**通信。该协议基于**适配器模式(Adapter Pattern)**和**门面模式(Facade Pattern)**设计,旨在解决模块异构性与通信效率问题。
### 核心设计模式
### 统一通信张量协议 (Unified Tensor Protocol)
#### 适配器模式(编解码层)
所有子网间通过**纯数值张量**通信,不传递自然语言、JSON 或任何非张量结构。协议基于**适配器模式**设计。
每个子网通过一个**编解码层(Codec Layer)**与外部通信,负责将子网内部的私有数据格式(如 PyTorch Tensor、自定义结构等)转换为统一的标准张量格式:
#### 协议定义
统一通信张量由两部分组成:
**1. 元数据(独立于张量,由底层总线传递)**
| 字段 | 说明 |
|------|------|
| `source` | 发出模型 ID |
| `target` | 目标模型 ID |
| `reply_req` | 是否需要回复 (true/false) |
**2. 张量数据(纯数值张量,ndarray)**
```
┌─────────────┐ ┌───────────────────┐ ┌─────────────┐
│ 子网内部 │ ──→ │ 编码器 (Encoder) │ ──→ │ 标准张量 │
│ 私有格式 │ │ 适配器: 内部→标准 │ │ (统一格式) │
└─────────────┘ └───────────────────┘ └─────────────┘
│
▼
┌─────────────┐ ┌───────────────────┐ ┌─────────────┐
│ 子网内部 │ ←── │ 解码器 (Decoder) │ ←── │ 标准张量 │
│ 私有格式 │ │ 适配器: 标准→内部 │ │ (统一格式) │
└─────────────┘ └───────────────────┘ └─────────────┘
┌──────────────────────────────────────────────────┐
│ Unified Communication Tensor │
├──────────────────────────────────────────────────┤
│ [fmt_id] [parser_id] [padding...] [payload...] │
│ └───────── 自描述前缀 ──────────┘ └──真实数据──┘│
└──────────────────────────────────────────────────┘
```
**作用**:屏蔽不同子网/框架的实现差异。新的子网只要实现编解码适配器,就能接入系统,无需修改调度 Core 或其他已有模块。
#### 门面模式(统一张量接口)
协议对外暴露统一的、简单的接口,隐藏底层复杂的通信过程:
其中 `fmt_id` 和 `parser_id` 为**协议注册表**中预定义的标量标识:
- `fmt_id`:标识数据格式(如 0=ref_tensor, 1=memory_result, 2=reasoning_path)
- `parser_id`:标识应使用哪个解析网络(如 0=ref_parser, 1=memory_parser)
```python
# 统一接口门面 (Facade)
interface TensorBus:
def send(tensor: StandardTensor, target: str) -> None
def recv(source: str) -> StandardTensor
# 协议注册表(系统设计时预定义)
PROTOCOL_REGISTRY = {
"fmt": {0: "ref_tensor", 1: "memory_result", 2: "reasoning_path", ...},
"parser": {0: "ref_parser", 1: "memory_parser", 2: "reasoning_parser", ...},
}
```
调度 Core 和各子网无需关心数据是通过共享内存、NPU 显存直传还是 TCP/IP 传输,只需与统一的张量门面交互。
#### 编解码层(适配器模式)
### 标准张量格式
每个子网实现编解码适配器,在子系统内部格式与统一通信张量之间转换:
```
┌─────────────────────────────────────────────┐
│ 标准张量 (Standard Tensor) │
├─────────────────────────────────────────────┤
│ Header: 源子网 · 目标子网 · 类型 · 时间戳 │
│ Metadata: 维度 · 精度 · 压缩率 · 置信度 │
│ Payload: 实际数据(向量/结构化数据) │
│ Signature: SHA-256 校验 │
└─────────────────────────────────────────────┘
┌─────────────┐ ┌─────────────────────┐ ┌────────────────┐
│ 子网内部 │ ──→ │ 编码适配器 │ ──→ │ 统一通信张量 │
│ 内部张量 │ │ 拼接 fmt_id+parser_id│ │ [fmt][prs][数据]│
└─────────────┘ └─────────────────────┘ └────────────────┘
│
▼
┌─────────────┐ ┌─────────────────────┐ ┌────────────────┐
│ 子网内部 │ ←── │ 解码适配器 │ ←── │ 统一通信张量 │
│ 内部张量 │ │ 解析 parser_id → 解码 │ │ [fmt][prs][数据]│
└─────────────┘ └─────────────────────┘ └────────────────┘
```
**编解码器接口**:
```python
class CodecAdapter:
"""每个子网实现此接口"""
def encode(self, internal_tensor: Tensor) -> Tensor:
# 1. 查协议注册表获取 fmt_id
# 2. 拼接 [fmt_id, parser_id, ...padding..., payload]
pass
def decode(self, comm_tensor: Tensor) -> Tensor:
# 1. 读取前缀中的 parser_id
# 2. 路由到对应解码器
# 3. 剥离前缀,返回内部格式
pass
```
#### 门面模式(通信总线)
底层通信机制(共享内存/NPU 显存直传/TCP/IP)对子网透明:
```python
class TensorBus:
def send(tensor: Tensor, source: str, target: str, reply_req: bool) -> None
def recv(source: str) -> tuple[Tensor, str, str, bool] # tensor, source, target, reply_req
def register(subnet_id: str, codec: CodecAdapter) -> None
```
#### 核心特性
- **纯粹张量**:无 JSON,无嵌套结构,全部为数值数据
- **自描述前缀**:fmt_id + parser_id 编码在张量头部
- **协议预定义**:所有 fmt_id/parser_id 系统设计时查表确定
- **硬件亲和**:张量是 NPU/GPU 原生单元,可实现零拷贝
### 核心优势
#### 动态并网与即插即用

View File

@ -55,109 +55,138 @@ NextAgent 采用**认知解耦架构(Cognitive Decoupled Architecture)**,
## 统一张量协议(Unified Tensor Protocol)
这是 NextAgent 的核心创新——子网间通过标准化的张量格式通信,不传递自然语言。该协议基于**适配器模式(Adapter Pattern)**和**门面模式(Facade Pattern)**设计,旨在解决多子网架构中的模块异构性与通信效率问题。
NextAgent 的核心——子网间通过**纯数值张量**通信,不传递自然语言、JSON 或任何非张量结构。
### 核心设计模式
### 协议定义
#### 适配器模式(编解码层)
统一通信张量由**元数据**(独立于张量,由底层总线携带)+ **张量数据**(纯数值)两部分组成。
每个子网通过一个编解码层(Codec Layer)与外部通信。该层充当**适配器**,负责将子网内部的私有数据格式(如 PyTorch Tensor、MindSpore Tensor 或自定义结构)转换为统一的标准张量格式。
#### 元数据(总线层传递,非张量的一部分)
**作用**:屏蔽不同子网/框架的实现差异。新的子网只要实现对应的编解码器(Adapter),就能接入系统,无需修改调度 Core 或其他已有模块的代码。
| 字段 | 说明 |
|------|------|
| `source` | 发出模型 ID (如 `routing`, `memory`, `logic_trace`, `llm`, `core`) |
| `target` | 目标模型 ID |
| `reply_req` | 是否需要回复 (true/false) |
#### 门面模式(统一张量接口)
#### 张量数据(纯数值 ndarray)
协议对外暴露统一的、简单的接口,隐藏底层复杂的通信过程:
```python
class TensorBus:
"""统一张量总线 - 门面模式"""
def send(tensor: StandardTensor, target: str) -> None:
# 隐藏了底层实现(共享内存/NPU显存直传/TCP/IP)
pass
def recv(source: str) -> StandardTensor:
pass
def register(subnet_id: str, codec: CodecAdapter) -> None:
"""动态注册子网 - 运行时热插拔"""
pass
```
┌────────────────────────────────────────────────────┐
│ Unified Communication Tensor │
├────────────────────────────────────────────────────┤
│ [fmt_id] [parser_id] [padding...] [payload...] │
│ └───────── 自描述前缀 (前缀张量) ──────┘ └──数据──┘│
└────────────────────────────────────────────────────┘
```
调度 Core 和各子网只需与统一的张量门面交互,无需关心底层通信细节。
张量本身就是一个纯粹的数值阵列(如 PyTorch Tensor / NumPy ndarray)。其前缀维度编码了协议信息:
- `dim[0]` = **fmt_id**:数据格式标识(查协议注册表)
- `dim[1]` = **parser_id**:应使用的解析网络标识
- `dim[2..k]`:保留 / padding
- `dim[k+1..]`:实际载荷数据
### 标准张量格式
#### 协议注册表(系统设计时预定义)
```json
{
"header": {
"source": "routing_compression | memory | logic_trace | llm | core",
"target": "routing_compression | memory | logic_trace | llm | core",
"tensor_type": "ref_tensor | memory_query | memory_result | reasoning_path | expression_verify | memory_update | control",
"timestamp": 1717200000,
"trace_id": "uuid-会话追踪ID"
},
"metadata": {
"dimensions": [768, 128], // 张量维度
"precision": "float32", // 精度类型
"compression_ratio": 0.85, // 压缩率
"confidence": 0.92, // 置信度
"encoding_model": "transformer_v3" // 编码模型标识
},
"payload": {
"data": [0.12, -0.34, ...], // 实际数据(向量/结构化数据)
"indices": [1024, 2048, ...], // 冷区索引(可选)
"metadata_map": {} // 附加元数据
},
"signature": "sha256-checksum" // 完整性校验
```python
PROTOCOL_REGISTRY = {
"fmt": {
0: "ref_tensor", # 路由引用张量
1: "memory_result", # 记忆检索结果
2: "memory_query", # 记忆查询
3: "memory_update", # 新信息更新
4: "reasoning_path", # 推理路径
5: "expression_verify", # 表达验证
6: "control", # 控制信号
},
"parser": {
0: "ref_parser",
1: "memory_result_parser",
2: "memory_query_parser",
3: "memory_update_parser",
4: "reasoning_parser",
5: "expression_parser",
6: "control_parser",
},
}
```
### 张量类型
### 编解码机制(适配器模式)
| 张量类型 | 说明 | 发送方 → 接收方 |
|----------|------|----------------|
| `ref_tensor` | 引用张量,压缩后的多模态特征 | 路由子网 → Core |
| `memory_query` | 记忆查询张量 | Core → 记忆子网 |
| `memory_result` | 记忆检索结果张量 | 记忆子网 → Core |
| `reasoning_path` | 推理路径与验证结果张量 | 逻辑子网 → Core |
| `expression_verify` | 表达验证张量(含语义编码、置信度) | LLM 子网 → Core |
| `memory_update` | 记忆更新参数包张量 | Core → 记忆子网 |
| `control` | 控制信号张量(启停、参数调整) | Core → 任意子网 |
### 编/解码机制
每个子网都配备适配器式的编解码层:
每个子网实现编解码适配器,在其内部张量格式与统一通信张量之间转换:
```
┌─────────────┐ ┌───────────────────┐ ┌─────────────┐
│ 子网内部 │ ──→ │ 编码适配器 │ ──→ │ 标准张量 │
│ 内部格式 │ │ Adapter: 内部→标准 │ │ (统一格式) │
└─────────────┘ └───────────────────┘ └─────────────┘
│
▼
┌─────────────┐ ┌───────────────────┐ ┌─────────────┐
│ 子网内部 │ ←── │ 解码适配器 │ ←── │ 标准张量 │
│ 内部格式 │ │ Adapter: 标准→内部 │ │ (统一格式) │
└─────────────┘ └───────────────────┘ └─────────────┘
┌─────────────┐ ┌─────────────────────┐ ┌──────────────────┐
│ 子网内部 │ ──→ │ 编码适配器 │ ──→ │ 统一通信张量 │
│ 内部张量 │ │ 查注册表 → │ │ [fmt][prs][数据] │
│ │ │ 拼接 fmt+parser 前缀 │ │ │
└─────────────┘ └─────────────────────┘ └──────────────────┘
│
▼
┌─────────────┐ ┌─────────────────────┐ ┌──────────────────┐
│ 子网内部 │ ←── │ 解码适配器 │ ←── │ 统一通信张量 │
│ 内部张量 │ │ 解析 prefix → │ │ [fmt][prs][数据] │
│ │ │ 路由对应解码器 → 剥离 │ │ │
└─────────────┘ └─────────────────────┘ └──────────────────┘
```
#### 编解码接口
```python
class CodecAdapter:
"""每个子网实现此接口,注册到 TensorBus"""
def encode(self, internal_tensor: Tensor,
target: str, reply_req: bool) -> Tensor:
"""内部格式 → 统一通信张量
1. 查协议注册表,获取内部格式对应的 fmt_id
2. 查协议注册表,获取本子网的 parser_id
3. 在数据前拼接 [fmt_id, parser_id, padding..., data...]
"""
pass
def decode(self, comm_tensor: Tensor) -> Tensor:
"""统一通信张量 → 内部格式
1. 读取 comm_tensor[0] 获取 parser_id
2. 根据 parser_id 路由到对应解码器
3. 剥离前缀,返回内部张量格式
"""
pass
class TensorBus:
"""通信总线(门面模式),屏蔽底层实现"""
def send(self, tensor: Tensor, source: str, target: str, reply_req: bool) -> None: ...
def recv(self, source: str) -> tuple[Tensor, str, str, bool]: ...
def register(self, subnet_id: str, codec: CodecAdapter) -> None: ...
```
### 核心优势
#### 动态并网与即插即用
- 新的子网只需实现编解码适配器,通过 `TensorBus.register()` 向 Core 注册即可接入
- 支持**运行时热插拔**:不中断系统运行即可添加/移除子网
- 各子网独立迭代算法,只要编解码接口不变,不影响系统稳定性
- 支持运行时根据需求训练新的子网并动态接入
| 优势 | 说明 |
|------|------|
| 纯粹张量 | 无 JSON/嵌套结构,全部为数值数据,NPU/GPU 原生处理 |
| 自描述 | fmt_id + parser_id 编码在张量头部,接收方即知如何解码 |
| 即插即用 | 新子网只需实现编解码适配器,通过 `TensorBus.register()` 注册 |
| 零拷贝 | 张量可直接在设备内存池流转,无需 CPU 搬运 |
| 高保真 | 无损传递高维特征图、注意力权重,不降维 |
| 硬件亲和 | 张量是 NPU/GPU 原生计算单元,无需格式转换 |
#### 零拷贝(Zero-Copy)
支持分布式能力的系统上,张量可通过**内核旁路技术**直接在设备间传递内存地址,避免将数据从 NPU/GPU 搬回 CPU 再打包的昂贵序列化开销。
### 通信示例
#### 硬件亲和性
张量是 NPU/GPU 的**原生计算单元**。协议允许数据在设备内存池中直接流转,无需经过格式转换,大幅降低延迟和 CPU 占用。
```
路由子网处理完用户输入后,调用编码适配器:
内部张量 (形状 [768]) → 编码器
→ 查表: 路由输出格式 → fmt_id=0, parser_id=0
→ 拼接: [0, 0, 0, ..., 768维数据] → 形状 [770]
→ TensorBus.send(张量, source="routing", target="core", reply_req=True)
#### 高数据保真度
相比于文本或 JSON 协议,张量能**无损传递高维特征图、注意力权重矩阵**等复杂数值结构,确保传递的是完整的
Core 收到后,调用解码适配器:
→ 读取 tensor[0]=0 → 格式是 ref_tensor
→ 读取 tensor[1]=0 → 用 ref_parser 解码
→ 剥离前 2 维 → 得到 768 维内部张量
```
## 调度 Core(前额叶皮层)
@ -282,12 +311,14 @@ function heuristic_process(raw_input):
elif next_action.type == "logical_reason":
reasoning_tensor = logic_subnet.reason(state.context)
state.context["reasoning"] = reasoning_tensor
state.confidence = reasoning_tensor.metadata.confidence
# 从张量前缀解析器解码获取置信度
state.confidence = decode_confidence_from_tensor(reasoning_tensor)
elif next_action.type == "llm_express":
verify_tensor = llm_subnet.express(state.context)
state.context["verification"] = verify_tensor
if verify_tensor.metadata.confidence > THRESHOLD:
# 解析表达验证张量获取通过判断
if decode_verified_from_tensor(verify_tensor):
state.verified = True
# 3. 将本轮新信息 + 结果交给记忆子网(自治更新)
@ -320,10 +351,10 @@ LLM 子网的输出有两个去向,这是核心设计之一:
│ → 自然语言文本 │ ──→ 直接输出给用户
│ │
├── 通道2 ──────────────┤
│ → 表达验证张量 │ ──→ 返回给 Core
│ (包含:语义编码、 │ Core 分析验证张量
│ 置信度、情感标记、 │ 判断 LLM 是否正确表达
│ 表达结构元数据) │ 而非获取自然语言内容
│ → 表达验证张量(纯张量) │ ──→ 返回给 Core
│ 格式约定见协议注册表 │ Core 的验证器解析此张量
│ fmt_id=5: 含语义编码/ │ 判断 LLM 是否正确表达
│ 意图对齐度/实体保留等 │ 而非获取自然语言内容
└──────────────────────┘
```
@ -347,8 +378,10 @@ LLM 子网的输出有两个去向,这是核心设计之一:
│ (不包含任何记忆处理逻辑) │
│ │
│ Core 只做两件事: │
│ 1. 封装查询张量 → 传给记忆子网 │
│ 2. 封装新信息张量 → 传给记忆子网 │
│ 1. 生成查询张量 → 传给记忆子网 │
│ (fmt_id=2, 含编码后的查询) │
│ 2. 生成新信息张量 → 传给记忆子网 │
│ (fmt_id=3, 含编码后的更新数据)│
└──────────┬───────────────────┘
│
▼
@ -370,18 +403,18 @@ LLM 子网的输出有两个去向,这是核心设计之一:
```
Core 需要查询记忆时:
│
├── 封装查询张量(含意图、种子实体、上下文)
├── 封装基础张量(含当前标签、路由结果)
├── 将当前上下文编码为内部查询张量
├── 调用编码适配器: 拼接 fmt_id=2 → 统一通信张量
├── TensorBus.send(张量, "core", "memory", reply_req=True)
│
▼
记忆子网接收后自行检索
│
├── 图数据库 Cypher 查询
├── 向量数据库语义检索
├── 自行关联推理
记忆子网接收后:
├── 解码适配器读取 prefix → fmt_id=2 (memory_query)
├── 自行解析查询意图 → 图DB/向量DB/文本检索
├── 将结果编码: 拼接 fmt_id=1 (memory_result)
│
▼
返回包含记忆的张量给 Core
返回记忆结果张量给 Core
```
## 数据流完整链路(用户视角)

View File

@ -75,42 +75,52 @@ NextAgent LLM 子网:
| **输入** | 结构化推理路径张量(来自 Core,经过张量解码层转换) |
| **输出1** | 自然语言文本(直接发送给用户) |
| **输出2** | 表达验证张量(发送给 Core) |
| **输出2格式** | 标准张量协议,tensor_type=`expression_verify` |
| **输出2格式** | 统一通信张量,fmt_id=5 (expression_verify) |
### 表达验证张量格式
```json
{
"header": {
"source": "llm",
"target": "core",
"tensor_type": "expression_verify",
"trace_id": "trace-uuid",
"timestamp": 1717200000
},
"metadata": {
"confidence": 0.94,
"semantic_encoding_dim": 768,
"expression_structure": "sentence+paragraph",
"sentiment": "positive",
"language": "zh-CN",
"token_count": 128,
"verification_status": "verified"
},
"payload": {
"semantic_encoding": [0.23, -0.56, ...], // 输出文本的语义向量
"intention_alignment": 0.96, // 与输入推理路径的意图对齐度
"factual_consistency": 0.92, // 与输入事实的一致性
"sentence_embeddings": [ // 每句的编码(可选)
{"index": 0, "embedding": [...]},
{"index": 1, "embedding": [...]}
],
"emotion_vector": [0.1, 0.8, 0.3, ...], // 情感多维度向量
"key_entities_preserved": ["天气", "北京", "今天"], // 关键实体保留
"omission_detected": false // 是否有遗漏信息
},
"signature": "sha256-checksum"
}
遵循统一通信张量协议,**无 JSON 结构**,全部为纯数值张量:
```
统一通信张量 (纯数值)
prefix: [fmt_id=5, parser_id=5]
payload:
[semantic_encoding (768)] # 输出文本的语义向量
[intention_alignment (1)] # 与输入推理路径的意图对齐度 (0~1)
[factual_consistency (1)] # 与输入事实的一致性 (0~1)
[emotion_vector (n_dims)] # 情感向量
[entity_preserved_mask (n_entities)] # 关键实体保留掩码 [0/1]
[overall_score (1)] # 综合验证得分 (0~1)
```
**编码示例**:
```python
def encode_expression_verify(output_text: str,
reasoning_path: Tensor) -> Tensor:
"""LLM 自然语言输出 → 表达验证张量 (通道2)
通道1: 自然语言文本直接输出给用户
通道2: 此函数编码验证特征为纯张量返回给 Core
"""
sem_enc = sentence_encoder(output_text) # [768]
alignment = cosine_sim(sem_enc, path_embedding) # [1]
consistency = compute_factual_consistency(
output_text, reasoning_path) # [1]
emotion = emotion_analyzer(output_text) # [n]
entity_mask = check_entities_preserved(output_text) # [n_entities]
overall = compute_overall_score([alignment, consistency,
entity_mask.mean()]) # [1]
payload = concat(sem_enc, alignment, consistency,
emotion, entity_mask, overall)
return codec.encode(fmt_id=5, parser_id=5, data=payload)
# Core 收到后解析
payload = codec.decode(comm_tensor) # 剥离 fmt_id=5
overall = payload[-1]
if overall > THRESHOLD: # 验证通过
user_output = channel_1_text # 允许输出
```
## 子网内部架构

View File

@ -44,64 +44,35 @@
### 推理路径张量格式
```json
{
"header": {
"source": "logic_trace",
"target": "core",
"tensor_type": "reasoning_path",
"trace_id": "trace-uuid",
"timestamp": 1717200000
},
"payload": {
"reasoning_steps": [
{
"step_id": 1,
"type": "deduction",
"input": "用户想要查询天气",
"operation": "提取地点和时间信息",
"output": "地点: 北京, 时间: 今天",
"confidence": 0.95,
"source_evidence": ["memory:偏好_北京天气"]
},
{
"step_id": 2,
"type": "retrieval",
"input": "需要天气数据",
"operation": "查询记忆子网获取北京天气偏好",
"output": "用户关注空气质量>温度",
"confidence": 0.88,
"source_evidence": ["graph:用户_北京_HAS_PERSONA_关注空气质量"]
},
{
"step_id": 3,
"type": "verification",
"input": "步骤1-2结果",
"operation": "交叉验证",
"output": "结论一致",
"confidence": 0.92,
"source_evidence": ["cross_validation"]
}
],
"final_confidence": 0.92,
"alternative_paths": [
{
"path_id": "alt_1",
"conclusion": "用户可能不在北京(上次对话提到出差)",
"confidence": 0.45,
"rejection_reason": "信息过时(3天前的记录)"
}
],
"verified": true,
"error_log": []
},
"metadata": {
"step_count": 3,
"mcts_iterations": 50,
"search_depth": 5,
"verification_status": "passed"
}
}
遵循统一通信张量协议,**无 JSON 结构**,全部为纯数值张量:
```
统一通信张量 (纯数值)
prefix: [fmt_id=4, parser_id=4]
payload:
[n_steps] # 推理步数
[step_embeddings...] # 每步的语义嵌入 [n_steps, 768]
[step_confidences...] # 每步置信度 [n_steps]
[search_depth, mcts_iters, verified] # 元信息 [3]
```
**编码示例(内部推理路径 → 统一张量)**:
```python
def encode_reasoning_path(steps: list[Step],
depth: int, iters: int) -> Tensor:
"""内部推理路径 → 统一通信张量"""
n = len(steps)
step_vecs = [encode_step_io(s) for s in steps] # 每步编码
confs = [s.confidence for s in steps]
payload = concat(
tensor([n]), # 步数
stack(step_vecs).flatten(), # 步嵌入
tensor(confs), # 置信度
tensor([depth, iters, 1.0 if verified]) # 元信息
)
return codec.encode(fmt_id=4, parser_id=4, data=payload)
```
## 子网内部架构

View File

@ -147,9 +147,9 @@ LIMIT 10
| 项目 | 说明 |
|------|------|
| **输入1** | 查询张量(来自 Core,含查询意图、种子实体、上下文) |
| **输入2** | 新信息张量(来自 Core,含本轮标签和子网结果;**不含具体写入指令**) |
| **输出** | `memory_result` 张量(包含检索到的实体/关系/向量结果) |
| **输入1** | 查询张量,fmt_id=2,来自 Core(编码后的查询意图、种子实体) |
| **输入2** | 新信息张量,fmt_id=3,来自 Core(编码后的本轮更新数据;**不含具体写入指令**) |
| **输出** | 记忆结果张量,fmt_id=1,编码检索到的实体/关系向量 |
| **自治** | 记忆子网自行决定如何写入、更新、删除、修改记忆 |
## 训练目标
@ -301,37 +301,63 @@ class Neo4jGraph:
## 新信息张量格式
Core 传递给记忆子网的**新信息张量**(Core 不发送
Core 传递给记忆子网的**新信息张量**遵循统一通信张量协议:
```json
{
"header": {
"source": "core",
"target": "memory",
"tensor_type": "memory_update",
"trace_id": "trace-uuid",
"timestamp": 1717200000
},
"payload": {
### 协议定义
```
格式: 统一通信张量 (纯数值张量)
prefix[0] = fmt_id = 3 (memory_update)
prefix[1] = parser_id = 3 (memory_update_parser)
payload: 编码后的记忆更新数据
实体编码: [entity_ids..., entity_embeddings...]
关系编码: [relation_ids..., relation_embeddings...]
会话标识: [session_tag, turn_tag, ...]
```
实体、关系、会话信息通过子网内部的编解码适配器转为张量表示。
### 记忆子网内部格式(私有)
记忆子网内部维护以下私有数据结构,但**对外通信时始终通过统一协议编解码为纯张量**:
```python
# 内部数据模型(子网私有,不对外暴露)
@Entity
default_graph = {
"entities": [
{"name": "用户_张三", "type": "PersonaNode", "attributes": {"preference": "摇滚音乐"}},
{"name": "项目_NextAgent", "type": "EntityNode", "attributes": {}}
{"name": "用户_张三", "type": "PersonaNode", "preference": "摇滚音乐"},
{"name": "项目_NextAgent", "type": "EntityNode"}
],
"relations": [
{"subject": "用户_张三", "relation": "HAS_PERSONA", "object": "喜欢_摇滚", "confidence": 0.95},
{"subject": "用户_张三", "relation": "RELATES_TO", "object": "项目_NextAgent", "confidence": 0.9}
],
"session_id": "session-20260601-xxxx",
"turn_id": 42,
"metadata": {
"source_subnets": ["routing", "logic_trace", "core"],
"aggregation_strategy": "merge"
}
},
"metadata": {
"entity_count": 3,
"relation_count": 2,
"upsert_mode": "merge"
}
{"subject": "用户_张三", "relation": "HAS_PERSONA", "object": "喜欢_摇滚", "confidence": 0.95},
{"subject": "用户_张三", "relation": "RELATES_TO", "object": "项目_NextAgent", "confidence": 0.9}
]
}
# 编码适配器将内部数据 → 统一通信张量
# 解码适配器将统一通信张量 → 内部查询结构
def encode_memory_update(entities, relations, session_id, turn_id) -> Tensor:
"""内部格式 → 统一通信张量
1. 将实体名/关系名编码为 embedding 向量
2. 拼接会话标识和时间戳
3. 编码适配器: 查表 fmt_id=3, 拼接前缀
"""
entity_embeds = embed_entity_names(entities) # [n_entities, 768]
relation_embeds = encode_relations(relations) # [n_relations, 768]
meta = tensor([session_hash, turn_id]) # [2]
return codec.encode(fmt_id=3, parser_id=3,
data=concat(entity_embeds, relation_embeds, meta))
def decode_memory_query(comm_tensor: Tensor) -> QueryIntent:
"""统一通信张量 → 内部查询意图
1. codec.decode(comm_tensor) 剥离前缀
2. 将 payload 张量解码为查询条件
3. 返回: 图DB检索参数 / 向量检索参数
"""
payload = codec.decode(comm_tensor) # 剥离 fmt_id=2
query_embed = payload[:768]
return self.query_embed_to_params(query_embed)
```

View File

@ -71,7 +71,7 @@
|------|------|
| **输入** | 原始多模态数据(图像/语音/文本)+ 控制张量(来自Core) |
| **输出** | 标准引用张量(发送给Core) |
| **输出格式** | 标准张量协议,tensor_type=`ref_tensor` |
| **输出格式** | 统一通信张量,fmt_id=0 (ref_tensor) |
| **压缩目标** | 10:1 ~ 100:1 压缩率 |
| **延迟目标** | 单次处理 < 50ms |
@ -103,7 +103,8 @@
│ ▼ │
│ ┌─────────────────────┐ │
│ │ 张量编码层 │ │
│ │ 内部格式→标准张量格式 │ │
│ │ 内部格式→统一通信张量 │ │
│ │ 拼接 fmt_id=0 前缀 │ │
│ └──────────┬──────────┘ │
└─────────────────────────┼─────────────────────┘
│
@ -195,20 +196,21 @@
6. 训练 100 epochs,batch size 256
7. 评估:在不同压缩率 (10x, 50x, 100x) 下测量重建质量
### 阶段 4:张量编码层训练
### 阶段 4:编码适配器训练
**目标**:训练编码层将内部张量转换为标准通信张量
**目标**:训练编码层将内部张量转换为统一通信张量(拼接 fmt_id/parser_id 前缀)
**技术路线**:
- 可学习的线性投影层 + LayerNorm
- 维度对齐网络(Projection Head)
**训练步骤**:
1. 定义标准张量协议的目标格式(固定维度、精度等)
1. 定义统一通信张量的目标维度(payload 部分)
2. 收集阶段3的 VQ-VAE 潜在表示作为训练数据
3. 训练投影网络将内部张量映射到标准格式
4. 训练配套的解码网络(供其他子网使用)
5. 验证编码/解码循环的一致性误差
3. 训练投影网络将内部张量映射到 payload 格式
4. 实现编码适配器: payload + 拼接 [fmt_id=0, parser_id=0, padding..., data]
5. 训练配套的解码网络(供其他子网使用)
6. 验证编码/解码循环的一致性误差
### 阶段 5:联合端到端微调

View File

@ -42,21 +42,24 @@ NextAgent 的所有子网间通过**标准张量协议**通信。这些张量是
### 解析目标
| 张量类型 | 输入 | 人类可读输出 |
|----------|------|-------------|
| `ref_tensor` | 路由子网的压缩特征向量 | "用户输入了一张北京故宫的照片,标签: ['建筑', '故宫', '北京'],置信度: 0.92" |
| `memory_result` | 记忆子网的检索结果 | "找到 3 条相关记忆: (1) 用户上次去北京是 2025年3月, (2) 用户偏好空气质量信息, (3) 用户关注pm2.5指数" |
| `reasoning_path` | 推理路径与验证结果 | "推理链: 步骤1(提取实体)→步骤2(查天气数据)→步骤3(推理结论: 北京今天晴, 22°C), 置信度: 0.95, 已验证: true" |
| `expression_verify` | LLM 表达验证张量 | "表达验证: 语义对齐度 0.96, 情感正向, 关键实体保留: [天气,北京,今天], 遗漏检测: 无" |
| `memory_update` | 记忆更新信息 | "新信息: {实体: [用户_张三], 关系: [HAS_PERSONA 喜欢_摇滚], 置信度: 0.9}" |
| `control` | 控制信号 | "控制信号: Core→路由子网, 类型: [从滑动窗口提取详细信息], 参数: {depth: 2}" |
所有输入为**统一通信张量(纯数值)**,解析为人类可读文本。关键区别:解析器不读取 JSON 字段,而是读取张量前缀中的 parser_id 路由到对应解析器。
| 张量 fmt_id | 解析器 | 输入 (纯张量) | 人类可读输出 |
|:----------:|:------:|:------------:|:------------|
| `0` (ref) | ref_parser | payload: 压缩后的多模态特征向量 | "用户输入了一张北京故宫的照片,标签: [建筑, 故宫, 北京]" |
| `1` (memory_result) | memory_parser | payload: 记忆检索结果的实体/关系编码 | "找到 3 条相关记忆: (1) 用户上次去北京是 2025年3月..." |
| `2` (memory_query) | query_parser | payload: 编码后的查询意图 | "Core 查询: 用户上次提到过什么音乐偏好?" |
| `3` (memory_update) | update_parser | payload: 编码后的更新数据 | "新信息: 实体[用户_张三], 关系[HAS_PERSONA 喜欢_摇滚]" |
| `4` (reasoning_path) | reasoning_parser | payload: 推理步嵌入+置信度向量 | "推理链: 步骤1(提取实体: 北京) → 步骤2(查天气...) → 结论" |
| `5` (expression_verify) | expression_parser | payload: 语义编码+对齐度+情感 | "表达验证: 语义对齐度 0.96, 情感正向, 综合得分 0.94" |
| `6` (control) | control_parser | payload: 控制参数编码 | "控制信号: Core→路由子网, 要求从滑动窗口提取详细信息" |
### 输入/输出
| 项目 | 说明 |
|------|------|
| **输入** | 标准张量(任意类型) |
| **输出** | 结构化人类可读文本(Markdown / JSON / 自然语言) |
| **输入** | 统一通信张量(纯数值,前缀含 parser_id,其余为载荷数据) |
| **输出** | 结构化人类可读文本(Markdown / 自然语言) |
| **不参与运行时** | 纯训练/调试工具,不加入推理链路 |
| **延迟要求** | 无(仅离线使用) |
@ -67,88 +70,120 @@ NextAgent 的所有子网间通过**标准张量协议**通信。这些张量是
### 架构总览
```
标准张量 (任意类型)
统一通信张量 (纯数值, [fmt, parser, ...payload])
│
▼
┌──────────────────────────────────────────────┐
│ 张量解析网络 (Tensor Parser) │
│ │
│ ┌──────────────────────────────────────────┐│
│ │ 张量类型分类器 (Tensor Type Classifier) ││
│ │ 识别张量类型 → 路由到对应解析器 ││
│ │ 输出: ref / memory_result / reasoning / ││
│ │ expression / update / control ││
│ └──────────────────┬───────────────────────┘│
│ │ │
│ ┌────────────────┼────────────────────┐ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌────────┐ ┌────────────┐ ┌────────┐│
│ │ 引用 │ │ 记忆结果 │ ... │ 控制 ││
│ │ 解析器 │ │ 解析器 │ │ 解析器 ││
│ └───┬────┘ └─────┬──────┘ └───┬────┘│
│ │ │ │ │
│ └──────────────┼────────────────────┘ │
│ ▼ │
│ ┌──────────────────────────────────────────┐│
│ │ 文本生成器 (Text Generator) ││
│ │ 将结构化解析结果转换为人类可读文本 ││
│ │ 支持: 自然语言 / 结构化表格 / JSON ││
│ └──────────────────────────────────────────┘│
│ │
└──────────────────────────────────────────────┘
┌──────────────────────────────────────────────────┐
│ 张量解析网络 (Tensor Parser) │
│ │
│ 读取 dim[0] = fmt_id
│ 读取 dim[1] = parser_id
│ 剥离前 k 维 → payload 纯数据
│ │
│ ┌──────────────────────────────────────────────┐│
│ │ 解析路由器 (Parser Router) ││
│ │ parser_id → 直接路由到对应解析器 ││
│ │ (查表: 0→ref, 1→memory, 4→reasoning, ...) ││
│ └──────────────────┬───────────────────────────┘│
│ │ │
│ ┌────────────────┼────────────────┐ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌────────┐ ┌────────────┐ ┌──────────┐ │
│ │ 引用 │ │ 记忆结果 │ │ 推理路径 │ │
│ │ 解析器 │ │ 解析器 │ │ 解析器 │ │
│ └───┬────┘ └─────┬──────┘ └────┬─────┘ │
│ │ │ │ │
│ └──────────────┼────────────────┘ │
│ ▼ │
│ ┌────────────────────────────────────────────┐│
│ │ 文本生成器 (Text Generator) ││
│ │ 将解析器输出的结构化数据转为人类可读文本 ││
│ │ 支持: 自然语言 / 结构化表格 ││
│ └────────────────────────────────────────────┘│
│ │
└────────────────────────────────────────────────┘
│
▼
人类可读输出文本
```
### 类型分类器
### 解析路由器
一个轻量级分类器,从张量 header 中提取 `tensor_type` 字段并路由到对应解析器。
输入张量中的 `parser_id` 直接索引到对应解析器,无需分类网络:
```python
class TensorTypeClassifier:
def classify(self, tensor: StandardTensor) -> str:
# 直接从 header 读取(运行时信息) + 向量特征校验
expected_type = tensor.header.tensor_type
payload_hash = hash(tensor.payload.data[:16])
# 可选:用小网络验证类型与负载内容一致
if self.verification_enabled:
type_logits = self.small_classifier(tensor.payload.data)
predicted_type = self.type_names[type_logits.argmax()]
if predicted_type != expected_type:
logger.warning(f"类型不一致: header={expected_type}, payload特征={predicted_type}")
return expected_type
PARSER_MAP = {
0: RefParser(), # ref_tensor
1: MemoryResultParser(), # memory_result
2: MemoryQueryParser(), # memory_query
3: MemoryUpdateParser(), # memory_update
4: ReasoningParser(), # reasoning_path
5: ExpressionParser(), # expression_verify
6: ControlParser(), # control
}
def route_and_parse(comm_tensor: Tensor) -> str:
"""
1. 读取张量前缀中的 parser_id
2. 路由到对应解析器
3. 解析器处理 payload 部分(剥离前缀后)
"""
parser_id = int(comm_tensor[1]) # dim[1]
parser = PARSER_MAP[parser_id]
payload = comm_tensor[PREFIX_LEN:] # 剥离前缀
return parser.parse(payload)
```
### 类型一致性校验(可选调试功能)
```python
def verify_type_consistency(comm_tensor: Tensor) -> bool:
"""校验 fmt_id 与 payload 内容一致(可选)"""
fmt_id = int(comm_tensor[0])
parser_id = int(comm_tensor[1])
payload = comm_tensor[PREFIX_LEN:]
# 用小网络验证负载特征与 fmt_id 匹配
if verification_enabled:
predicted = small_classifier(payload[:64]) # 只看前64维
if predicted != fmt_id:
logger.warning(f"类型不一致: header fmt_id={fmt_id}, payload特征={predicted}")
return fmt_id == parser_id # 正常情况下 fmt_id == parser_id
### 专用解析器
所有解析器接收的是**剥离前缀后的纯载荷张量**,需自己解析其内部结构。
#### 引用解析器 (Ref Parser)
将路由子网的压缩特征向量解码为标签信息:
路由子网的压缩特征向量解码:
```python
class RefParser(nn.Module):
"""引用张量解析器"""
"""引用张量解析器
输入: 剥离前缀后的 payload 张量 [768]
输出: 标签文本
"""
def __init__(self, latent_dim=768, label_vocab_size=10000):
super().__init__()
self.encoder = nn.Linear(latent_dim, 512)
self.label_predictor = nn.Linear(512, label_vocab_size) # 标签分类
self.confidence_head = nn.Linear(512, 1) # 置信度估计
self.category_head = nn.Linear(512, 50) # 类别预测
self.label_predictor = nn.Linear(512, label_vocab_size)
self.confidence_head = nn.Linear(512, 1)
def forward(self, ref_tensor):
h = self.encoder(ref_tensor)
label_logits = self.label_predictor(h) # 预测标签
confidence = torch.sigmoid(self.confidence_head(h)) # 置信度
category = self.category_head(h) # 类别
def forward(self, payload: Tensor):
"""payload: 剥离前缀后的纯数据张量"""
h = self.encoder(payload)
label_logits = self.label_predictor(h)
confidence = torch.sigmoid(self.confidence_head(h))
top_labels = label_logits.topk(5).indices
return {
"predicted_labels": label_logits.topk(5).indices,
"predicted_labels": decode_label_ids(top_labels),
"confidence": confidence.item(),
"category": category.argmax().item()
}
```
@ -156,25 +191,33 @@ class RefParser(nn.Module):
```python
class ReasoningParser(nn.Module):
"""推理路径张量→推理链文本"""
"""推理路径张量 → 推理链文本
输入 payload 结构:
[n_steps, step_emb_1(768), step_conf_1,
step_emb_2(768), step_conf_2, ..., meta(3)]
输出: 步骤描述文本
"""
def __init__(self, hidden_dim=768, max_steps=20):
super().__init__()
self.step_encoder = nn.Linear(hidden_dim, 256)
self.step_decoder = nn.TransformerDecoderLayer(
d_model=256, nhead=4
)
# 将每一步解码为文本 token
self.text_head = nn.Linear(256, vocab_size)
def forward(self, reasoning_tensor):
steps = reasoning_tensor["reasoning_steps"]
def forward(self, payload: Tensor):
# 解析 payload 结构: 先读步数,再拆出每步
n_steps = int(payload[0])
decoded = []
for step_emb in steps:
h = self.step_encoder(step_emb)
# 自回归解码为文本
offset = 1 # 跳过步数
for _ in range(n_steps):
step_emb = payload[offset:offset + 768]
conf = payload[offset + 768]
offset += 769
h = self.step_decoder(step_emb)
step_text = self.decode_step(h)
decoded.append(step_text)
decoded.append(f"{step_text} (conf={conf:.2f})")
return "\n".join(decoded)
```
@ -204,32 +247,36 @@ class ReasoningParser(nn.Module):
对于无法自然获得标注的场景,使用 LLM 生成合成数据:
```python
def synthesize_training_pair(tensor_type: str) -> tuple:
"""生成 (标准张量, 人类可读文本) 配对"""
def synthesize_training_pair(fmt_id: int) -> tuple:
"""生成 (统一通信张量, 人类可读文本) 配对"""
# 1. 随机生成或从真实数据采样标准张量
tensor = sample_tensor(tensor_type)
# 1. 随机生成或从真实数据采样统一通信张量
comm_tensor = sample_comm_tensor(fmt_id)
# 2. 使用 LLM 将张量转为人类可读描述
metadata_json = json.dumps(tensor.header, ensure_ascii=False)
payload_preview = summarize_payload(tensor.payload.data)
# 2. 解析张量属性,不使用 JSON header(协议中不存在)
parser_id = int(comm_tensor[1])
payload = comm_tensor[PREFIX_LEN:]
payload_stats = {
"shape": list(payload.shape),
"mean": float(payload.mean()),
"std": float(payload.std()),
}
fmt_name = PROTOCOL_REGISTRY["fmt"][fmt_id]
llm_prompt = f"""
你是一个张量解析专家。以下是一个 {tensor_type} 类型的标准张量:
你是一个张量解析专家。以下是一个 {fmt_name} 类型的统一通信张量。
这是一个纯数值张量,不包含任何 JSON 元数据。
头: {metadata_json}
负载摘要: {payload_preview}
维度: {tensor.metadata.dimensions}
置信度: {tensor.metadata.confidence}
fmt_id={fmt_id} ({fmt_name})
parser_id={parser_id}
载荷形状: {payload_stats['shape']}
载荷统计: mean={payload_stats['mean']:.4f}, std={payload_stats['std']:.4f}
请用自然语言描述这个张量代表什么。
例如,如果是 ref_tensor,描述"用户输入了关于XXX的查询,提取到的关键标签是..."
如果是 reasoning_path,描述推理链的每一步和最终结论。
请根据已知的数据格式约定,用自然语言描述此张量的含义。
"""
human_readable = llm.generate(llm_prompt)
return (tensor, human_readable)
return (comm_tensor, human_readable)
```
#### 方法三:端到端采集
@ -363,34 +410,54 @@ def regression_test(subnet, test_cases):
```python
class RuleBasedParser:
"""基于规则的张量解析器(MVP)"""
"""基于规则的张量解析器(MVP)
def parse(self, tensor: StandardTensor) -> str:
tensor_type = tensor.header.tensor_type
读取张量前缀中的 parser_id 确定类型,
然后根据预定格式输出可读文本。
"""
PARSER_NAMES = {
0: "引用张量 (ref_tensor)",
1: "记忆结果 (memory_result)",
2: "记忆查询 (memory_query)",
3: "记忆更新 (memory_update)",
4: "推理路径 (reasoning_path)",
5: "表达验证 (expression_verify)",
6: "控制信号 (control)",
}
def parse(self, comm_tensor: Tensor) -> str:
fmt_id = int(comm_tensor[0])
parser_id = int(comm_tensor[1])
payload = comm_tensor[PREFIX_LEN:]
if tensor_type == "control":
return self.parse_control(tensor)
elif tensor_type == "memory_result":
return self.parse_memory(tensor)
type_name = self.PARSER_NAMES.get(fmt_id, f"未知类型:{fmt_id}")
if fmt_id == 6: # control
return self.parse_control(payload)
elif fmt_id in (0, 1): # ref / memory_result
return self.parse_vector(payload, type_name)
elif fmt_id == 4: # reasoning
return self.parse_reasoning(payload)
elif fmt_id == 5: # expression
return self.parse_expression(payload)
else:
# 其他类型走模板
return f"[{tensor_type}] 源={tensor.header.source}, 目标={tensor.header.target}, " \
f"置信度={tensor.metadata.confidence:.2f}, " \
f"维度={tensor.metadata.dimensions}"
return f"[{type_name}] 维度={list(payload.shape)}, " \
f"均值={float(payload.mean()):.4f}"
def parse_control(self, tensor):
params = tensor.payload.metadata_map
return f"Core 发往 {tensor.header.target}: " \
f"控制类型={params.get('control_type', 'unknown')}, " \
f"参数={params}"
def parse_control(self, payload):
return f"控制信号张量 (维度{list(payload.shape)})"
def parse_memory(self, tensor):
entities = tensor.payload.metadata_map.get('entities', [])
rels = tensor.payload.metadata_map.get('relations', [])
text = f"记忆检索结果: 找到 {len(entities)} 个实体, {len(rels)} 条关系\n"
for r in rels[:3]:
text += f" · {r['subject']} -[{r['type']}]-> {r['object']} (置信度 {r.get('confidence', 1.0)})\n"
return text
def parse_vector(self, payload, type_name):
return f"[{type_name}] {list(payload.shape)}维向量"
def parse_reasoning(self, payload):
n_steps = int(payload[0])
return f"推理路径: {n_steps}步, 张量维度{list(payload.shape)}"
def parse_expression(self, payload):
alignment = float(payload[768]) # 约定的位置
return f"表达验证: 意图对齐度={alignment:.2f}"
```
规则版不需要训练,可以立即使用。随着需求积累,逐步加入可学习的解析器替换规则路径。

1
requirements.txt Normal file
View File

@ -0,0 +1 @@
numpy>=1.24.0

3
src/parsers/__init__.py Normal file
View File

@ -0,0 +1,3 @@
from .rule_based import RuleBasedParser
__all__ = ["RuleBasedParser"]

Binary file not shown.

Binary file not shown.

46
src/parsers/rule_based.py Normal file
View File

@ -0,0 +1,46 @@
import numpy as np
from protocol.registry import PROTOCOL_REGISTRY
class RuleBasedParser:
"""规则解析器 MVP,不依赖 ML。
读取通信张量前缀中的 parser_id 确定类型,
根据各类型约定的 payload 结构输出可读文本。
"""
def parse(self, tensor: np.ndarray) -> str:
if tensor.shape[0] < PROTOCOL_REGISTRY.prefix_len:
return (f"张量过短: {tensor.shape[0]} < "
f"{PROTOCOL_REGISTRY.prefix_len}")
parser_id = int(tensor[1])
payload = tensor[PROTOCOL_REGISTRY.prefix_len:]
return self._parse_by_type(parser_id, payload)
def _parse_by_type(self, parser_id: int, payload: np.ndarray) -> str:
if parser_id == 0:
n_labels = int(payload[0]) if len(payload) > 0 else 0
return f"引用张量: {n_labels}个标签"
elif parser_id == 1:
n_entities = int(payload[0]) if len(payload) > 0 else 0
n_relations = int(payload[1]) if len(payload) > 1 else 0
return f"记忆结果: {n_entities}个实体, {n_relations}条关系"
elif parser_id == 2:
return "记忆查询"
elif parser_id == 3:
n_entities = int(payload[0]) if len(payload) > 0 else 0
n_relations = int(payload[1]) if len(payload) > 1 else 0
return f"记忆更新: {n_entities}个实体, {n_relations}条关系"
elif parser_id == 4:
n_steps = int(payload[0]) if len(payload) > 0 else 0
return f"推理路径: {n_steps}步"
elif parser_id == 5:
idx = 768
intent_alignment = payload[idx] if len(payload) > idx else 0.0
return f"表达验证: 意图对齐度={intent_alignment:.2f}"
elif parser_id == 6:
return "控制信号"
else:
name = PROTOCOL_REGISTRY.parser_name(parser_id)
return f"未知解析器: {parser_id} ({name})"

13
src/protocol/__init__.py Normal file
View File

@ -0,0 +1,13 @@
from .registry import PROTOCOL_REGISTRY, ProtocolRegistry
from .codec import CodecAdapter, IdentityCodec, EncodingError, DecodingError
from .tensor_bus import TensorBus
__all__ = [
"PROTOCOL_REGISTRY",
"ProtocolRegistry",
"CodecAdapter",
"IdentityCodec",
"EncodingError",
"DecodingError",
"TensorBus",
]

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

150
src/protocol/codec.py Normal file
View File

@ -0,0 +1,150 @@
"""
CodecAdapter — 编解码适配器基类
每个子网继承此类,实现 encode / decode 方法,完成
内部张量格式 ↔ 统一通信张量(带 fmt_id/parser_id 前缀)的转换。
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Any
import numpy as np
from .registry import PROTOCOL_REGISTRY
class EncodingError(Exception):
"""编码失败"""
pass
class DecodingError(Exception):
"""解码失败"""
pass
class CodecAdapter(ABC):
"""
编解码适配器基类
子类需实现:
- encode(internal_tensor, target, reply_req) -> comm_tensor
- decode(comm_tensor) -> internal_tensor
其中 comm_tensor 是形状为 [prefix_len + payload_dim] 的 numpy ndarray
"""
def __init__(self, subnet_id: str, fmt_id: int, parser_id: int):
PROTOCOL_REGISTRY.validate_subnet_id(subnet_id)
PROTOCOL_REGISTRY.validate_fmt_id(fmt_id)
PROTOCOL_REGISTRY.validate_parser_id(parser_id)
self.subnet_id = subnet_id
self.fmt_id = fmt_id
self.parser_id = parser_id
self.prefix_len = PROTOCOL_REGISTRY.prefix_len
# ------------------------------------------------------------------
# 编码:子网内部格式 → 统一通信张量
# ------------------------------------------------------------------
def encode(self, internal_tensor: np.ndarray, target: str, reply_req: bool = True) -> np.ndarray:
"""
内部张量 → 统一通信张量
Parameters
----------
internal_tensor : np.ndarray
子网内部格式的特征向量 (1D 或 2D)
target : str
目标子网 ID
reply_req : bool
是否需要回复
Returns
-------
comm_tensor : np.ndarray
形状为 [prefix_len + ...] 的统一通信张量
"""
PROTOCOL_REGISTRY.validate_subnet_id(target)
self._validate_internal(internal_tensor)
# 子类可重写此方法实现更复杂的编码逻辑
return self._build_prefix(target, reply_req, internal_tensor)
# ------------------------------------------------------------------
# 解码:统一通信张量 → 内部格式
# ------------------------------------------------------------------
def decode(self, comm_tensor: np.ndarray) -> np.ndarray:
"""
统一通信张量 → 内部张量
Parameters
----------
comm_tensor : np.ndarray
统一通信张量,前 prefix_len 维为前缀
Returns
-------
internal_tensor : np.ndarray
剥离前缀后的 payload 数据
"""
self._validate_prefix(comm_tensor)
return comm_tensor[self.prefix_len:] # 剥离前缀
# ------------------------------------------------------------------
# 校验内部张量格式(子类可重写)
# ------------------------------------------------------------------
def _validate_internal(self, tensor: np.ndarray):
if not isinstance(tensor, np.ndarray):
raise EncodingError(f"内部类型必须是 np.ndarray, 得到 {type(tensor)}")
if tensor.ndim not in (1, 2):
raise EncodingError(f"内部张量维度必须是 1 或 2, 得到 {tensor.ndim}")
# ------------------------------------------------------------------
# 校验统一通信张量的前缀
# ------------------------------------------------------------------
def _validate_prefix(self, tensor: np.ndarray):
if not isinstance(tensor, np.ndarray):
raise DecodingError(f"通信张量必须是 np.ndarray, 得到 {type(tensor)}")
if tensor.shape[0] < self.prefix_len:
raise DecodingError(
f"通信张量长度 ({tensor.shape[0]}) 不足前缀长度 ({self.prefix_len})"
)
fmt_id = int(tensor[0])
parser_id = int(tensor[1])
if fmt_id != self.fmt_id and fmt_id != -1: # -1 表示未知(测试用)
raise DecodingError(
f"fmt_id 不匹配: 预期 {self.fmt_id}, 实际 {fmt_id}"
)
if parser_id != self.parser_id and parser_id != -1:
raise DecodingError(
f"parser_id 不匹配: 预期 {self.parser_id}, 实际 {parser_id}"
)
# ------------------------------------------------------------------
# 构建前缀
# ------------------------------------------------------------------
def _build_prefix(self, target: str, reply_req: bool,
payload: np.ndarray) -> np.ndarray:
padding = self.prefix_len - 2
prefix = np.array([self.fmt_id, self.parser_id] + [0.0] * padding,
dtype=np.float64)
return np.concatenate([prefix, payload.ravel()])
class IdentityCodec(CodecAdapter):
"""
恒等编解码器(测试用):fmt_id/parser_id = -1
不做格式校验,直接拼接/剥离前缀。
"""
def __init__(self):
super().__init__(subnet_id="core", fmt_id=-1, parser_id=-1)
def _validate_prefix(self, tensor: np.ndarray):
if tensor.shape[0] < self.prefix_len:
raise DecodingError(f"张量过短: {tensor.shape[0]} < {self.prefix_len}")
def _validate_internal(self, tensor: np.ndarray):
if not isinstance(tensor, np.ndarray):
raise EncodingError(f"内部类型必须是 np.ndarray, 得到 {type(tensor)}")

83
src/protocol/registry.py Normal file
View File

@ -0,0 +1,83 @@
"""
Protocol Registry — 统一张量通信协议的 fmt_id / parser_id 注册表
所有 fmt_id 和 parser_id 均为系统设计时预定义,通信时编码在张量前缀中。
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Dict
# ---------------------------------------------------------------------------
# fmt_id — 数据格式标识
# ---------------------------------------------------------------------------
FMT_NAMES: Dict[int, str] = {
0: "ref_tensor", # 路由引用张量 (压缩后的多模态特征)
1: "memory_result", # 记忆检索结果 (实体/关系编码)
2: "memory_query", # 记忆查询 (编码后的查询意图)
3: "memory_update", # 新信息更新 (编码后的更新数据)
4: "reasoning_path", # 推理路径 (步嵌入 + 置信度)
5: "expression_verify", # 表达验证 (语义编码 + 对齐度 + 情感)
6: "control", # 控制信号 (控制参数编码)
}
# ---------------------------------------------------------------------------
# parser_id — 解析网络标识
# ---------------------------------------------------------------------------
PARSER_NAMES: Dict[int, str] = {
0: "ref_parser", # 引用张量解析器
1: "memory_result_parser", # 记忆结果解析器
2: "memory_query_parser", # 记忆查询解析器
3: "memory_update_parser", # 记忆更新解析器
4: "reasoning_parser", # 推理路径解析器
5: "expression_parser", # 表达验证解析器
6: "control_parser", # 控制信号解析器
}
# ---------------------------------------------------------------------------
# payload_prefix_len — 张量前缀长度(统一为 2: fmt_id + parser_id)
# 后续扩展可增加 padding,此时修改此值即可
# ---------------------------------------------------------------------------
PAYLOAD_PREFIX_LEN: int = 2
# ---------------------------------------------------------------------------
# 子网 ID 列表
# ---------------------------------------------------------------------------
SUBNET_IDS = ["routing", "memory", "logic_trace", "llm", "core"]
# ---------------------------------------------------------------------------
# 校验
# ---------------------------------------------------------------------------
assert FMT_NAMES.keys() == PARSER_NAMES.keys(), "fmt_id 与 parser_id 必须一一对应"
@dataclass
class ProtocolRegistry:
"""协议注册表 — 封装查表逻辑"""
fmt_names: Dict[int, str] = field(default_factory=lambda: dict(FMT_NAMES))
parser_names: Dict[int, str] = field(default_factory=lambda: dict(PARSER_NAMES))
prefix_len: int = PAYLOAD_PREFIX_LEN
subnet_ids: list = field(default_factory=lambda: list(SUBNET_IDS))
def fmt_name(self, fmt_id: int) -> str:
return self.fmt_names.get(fmt_id, f"unknown_fmt_{fmt_id}")
def parser_name(self, parser_id: int) -> str:
return self.parser_names.get(parser_id, f"unknown_parser_{parser_id}")
def validate_fmt_id(self, fmt_id: int):
if fmt_id not in self.fmt_names:
raise ValueError(f"未知的 fmt_id: {fmt_id}, 可选: {list(self.fmt_names.keys())}")
def validate_parser_id(self, parser_id: int):
if parser_id not in self.parser_names:
raise ValueError(f"未知的 parser_id: {parser_id}, 可选: {list(self.parser_names.keys())}")
def validate_subnet_id(self, subnet_id: str):
if subnet_id not in self.subnet_ids:
raise ValueError(f"未知的子网 ID: {subnet_id}, 可选: {self.subnet_ids}")
# 全局单例
PROTOCOL_REGISTRY = ProtocolRegistry()

View File

@ -0,0 +1,37 @@
from typing import Dict
import numpy as np
from .registry import PROTOCOL_REGISTRY
from .codec import CodecAdapter, DecodingError
class TensorBus:
"""TensorBus 通信总线(门面模式)
维护子网注册表 {subnet_id: CodecAdapter},
提供统一的通信入口。
"""
def __init__(self):
self._subnets: Dict[str, CodecAdapter] = {}
def register(self, subnet_id: str, codec: CodecAdapter):
PROTOCOL_REGISTRY.validate_subnet_id(subnet_id)
self._subnets[subnet_id] = codec
def send(self, tensor: np.ndarray, source: str, target: str,
reply_req: bool = True) -> np.ndarray:
if source not in self._subnets:
raise KeyError(f"源子网未注册: {source}")
if target not in self._subnets:
raise KeyError(f"目标子网未注册: {target}")
source_codec = self._subnets[source]
target_codec = self._subnets[target]
comm_tensor = source_codec.encode(tensor, target, reply_req)
return target_codec.decode(comm_tensor)
def recv(self, source: str, tensor: np.ndarray) -> np.ndarray:
if source not in self._subnets:
raise KeyError(f"源子网未注册: {source}")
return self._subnets[source].decode(tensor)

Binary file not shown.

211
tests/test_protocol.py Normal file
View File

@ -0,0 +1,211 @@
import sys
import unittest
import numpy as np
sys.path.insert(0, "/home/program/NextAgent/src")
from protocol.registry import PROTOCOL_REGISTRY, FMT_NAMES, PARSER_NAMES
from protocol.codec import CodecAdapter, IdentityCodec, EncodingError, DecodingError
from protocol.tensor_bus import TensorBus
from parsers.rule_based import RuleBasedParser
class _TestCodec(CodecAdapter):
"""测试用编解码器:fmt_id=0 (ref_tensor), parser_id=0 (ref_parser)"""
def __init__(self):
super().__init__(subnet_id="core", fmt_id=0, parser_id=0)
class _LaxCodec(CodecAdapter):
"""宽松编解码器(类似 IdentityCodec 但使用合法 id)"""
def __init__(self):
super().__init__(subnet_id="core", fmt_id=0, parser_id=0)
def _validate_prefix(self, tensor):
if tensor.shape[0] < self.prefix_len:
raise DecodingError(f"张量过短: {tensor.shape[0]} < {self.prefix_len}")
def _validate_internal(self, tensor):
if not isinstance(tensor, np.ndarray):
raise EncodingError(f"内部类型必须是 np.ndarray, 得到 {type(tensor)}")
class TestProtocolRegistry(unittest.TestCase):
def test_validate_fmt_id_valid(self):
for fid in FMT_NAMES:
PROTOCOL_REGISTRY.validate_fmt_id(fid)
def test_validate_fmt_id_invalid(self):
with self.assertRaises(ValueError):
PROTOCOL_REGISTRY.validate_fmt_id(999)
def test_validate_parser_id_valid(self):
for pid in PARSER_NAMES:
PROTOCOL_REGISTRY.validate_parser_id(pid)
def test_validate_parser_id_invalid(self):
with self.assertRaises(ValueError):
PROTOCOL_REGISTRY.validate_parser_id(999)
def test_validate_subnet_id_valid(self):
for sid in PROTOCOL_REGISTRY.subnet_ids:
PROTOCOL_REGISTRY.validate_subnet_id(sid)
def test_validate_subnet_id_invalid(self):
with self.assertRaises(ValueError):
PROTOCOL_REGISTRY.validate_subnet_id("nonexistent")
def test_fmt_name_known(self):
self.assertEqual(PROTOCOL_REGISTRY.fmt_name(0), "ref_tensor")
def test_fmt_name_unknown(self):
self.assertEqual(PROTOCOL_REGISTRY.fmt_name(999), "unknown_fmt_999")
def test_parser_name_known(self):
self.assertEqual(PROTOCOL_REGISTRY.parser_name(0), "ref_parser")
def test_parser_name_unknown(self):
self.assertEqual(PROTOCOL_REGISTRY.parser_name(999),
"unknown_parser_999")
class TestCodecAdapter(unittest.TestCase):
def test_encode_decode_roundtrip(self):
codec = _TestCodec()
payload = np.array([1.0, 2.0, 3.0, 4.0])
comm = codec.encode(payload, target="routing", reply_req=True)
decoded = codec.decode(comm)
np.testing.assert_array_equal(decoded, payload)
def test_encode_raises_on_non_ndarray(self):
codec = _TestCodec()
with self.assertRaises(EncodingError):
codec.encode([1, 2, 3], target="routing", reply_req=True)
def test_decode_raises_on_short_tensor(self):
codec = _TestCodec()
short = np.array([0.0])
with self.assertRaises(DecodingError):
codec.decode(short)
def test_decode_raises_on_wrong_fmt_id(self):
codec = _TestCodec()
wrong = np.array([99.0, 0.0, 1.0, 2.0])
with self.assertRaises(DecodingError):
codec.decode(wrong)
class TestLaxCodec(unittest.TestCase):
def test_encode_decode_roundtrip(self):
codec = _LaxCodec()
payload = np.array([10.0, 20.0, 30.0])
comm = codec.encode(payload, target="core", reply_req=False)
decoded = codec.decode(comm)
np.testing.assert_array_equal(decoded, payload)
def test_decode_allows_any_prefix(self):
codec = _LaxCodec()
comm = np.array([99.0, 99.0, 42.0])
decoded = codec.decode(comm)
np.testing.assert_array_equal(decoded, np.array([42.0]))
def test_decode_raises_on_short_tensor(self):
codec = _LaxCodec()
with self.assertRaises(DecodingError):
codec.decode(np.array([0.0]))
def test_encode_raises_on_non_ndarray(self):
codec = _LaxCodec()
with self.assertRaises(EncodingError):
codec.encode("bad", target="core", reply_req=True)
class TestRuleBasedParser(unittest.TestCase):
def setUp(self):
self.parser = RuleBasedParser()
def test_ref_tensor(self):
tensor = np.array([0.0, 0.0, 5.0, 1.0, 2.0, 3.0])
result = self.parser.parse(tensor)
self.assertEqual(result, "引用张量: 5个标签")
def test_memory_result(self):
tensor = np.array([1.0, 1.0, 3.0, 7.0])
result = self.parser.parse(tensor)
self.assertEqual(result, "记忆结果: 3个实体, 7条关系")
def test_memory_query(self):
tensor = np.array([2.0, 2.0, 1.0, 2.0, 3.0])
result = self.parser.parse(tensor)
self.assertEqual(result, "记忆查询")
def test_memory_update(self):
tensor = np.array([3.0, 3.0, 2.0, 4.0])
result = self.parser.parse(tensor)
self.assertEqual(result, "记忆更新: 2个实体, 4条关系")
def test_reasoning_path(self):
tensor = np.array([4.0, 4.0, 6.0])
result = self.parser.parse(tensor)
self.assertEqual(result, "推理路径: 6步")
def test_expression_verify(self):
tensor = np.zeros(2 + 769)
tensor[0] = 5.0
tensor[1] = 5.0
tensor[2 + 768] = 0.85
result = self.parser.parse(tensor)
self.assertEqual(result, "表达验证: 意图对齐度=0.85")
def test_control(self):
tensor = np.array([6.0, 6.0])
result = self.parser.parse(tensor)
self.assertEqual(result, "控制信号")
def test_short_tensor(self):
tensor = np.array([0.0])
result = self.parser.parse(tensor)
self.assertIn("过短", result)
class TestTensorBus(unittest.TestCase):
def setUp(self):
self.bus = TensorBus()
self.codec_a = _TestCodec()
self.codec_b = _LaxCodec()
self.bus.register("routing", self.codec_a)
self.bus.register("core", self.codec_b)
def test_register_and_send(self):
payload = np.array([7.0, 8.0, 9.0])
result = self.bus.send(payload, source="routing", target="core")
np.testing.assert_array_equal(result, payload)
def test_recv(self):
payload = np.array([1.0, 2.0])
comm = self.codec_a.encode(payload, target="routing", reply_req=True)
result = self.bus.recv("routing", comm)
np.testing.assert_array_equal(result, payload)
def test_send_unregistered_source(self):
with self.assertRaises(KeyError):
self.bus.send(np.array([1.0]), source="nonexistent", target="core")
def test_send_unregistered_target(self):
with self.assertRaises(KeyError):
self.bus.send(np.array([1.0]), source="routing",
target="nonexistent")
def test_recv_unregistered_source(self):
with self.assertRaises(KeyError):
self.bus.recv("nonexistent", np.array([1.0, 2.0]))
if __name__ == "__main__":
unittest.main()