update

Dockerfile

@@ -6,6 +6,7 @@ FROM python:3.12.11
 RUN useradd -m -u 1000 user
 USER user
 ENV PATH="/home/user/.local/bin:$PATH"
+ENV PYTHONPATH="/app:$PYTHONPATH"
 
 WORKDIR /app
 
@@ -13,4 +14,9 @@ COPY --chown=user ./requirements.txt requirements.txt
 RUN pip install --no-cache-dir --upgrade -r requirements.txt
 
 COPY --chown=user . /app
-CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]
+
+# 设置环境变量
+ENV API_HOST="0.0.0.0"
+ENV API_PORT="7860"
+
+CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "7860"]

README_APP.md

@@ -0,0 +1,57 @@
+# Human-Clone System
+
+Human-Clone 系统是一个多主体通讯协作平台，支持 human 和 clone 之间的实时异步通讯。
+
+## 快速开始
+
+### 环境要求
+- Python 3.9+
+- Redis 服务器
+
+### 安装依赖
+```bash
+pip install -r requirements.txt
+```
+
+### 启动方式
+
+#### 1. 直接启动（推荐）
+```bash
+uvicorn src.main:app --reload --host 0.0.0.0 --port 7860
+```
+
+#### 2. 使用脚本启动
+```bash
+python scripts/start_dev.py
+```
+
+#### 3. 开发模式
+```bash
+uvicorn src.main:app --reload --debug --port 7860
+```
+
+## API 文档
+启动后访问：
+- Swagger UI: http://localhost:7860/docs
+- ReDoc: http://localhost:7860/redoc
+
+## 项目结构
+```
+├── src/                    # 源代码目录
+│   ├── core/              # 核心功能模块
+│   ├── config/            # 配置管理
+│   ├── api/               # API路由
+│   └── utils/             # 工具函数
+├── tests/                 # 测试代码
+├── scripts/               # 启动脚本
+└── docs/                  # 项目文档
+```
+
+## 核心功能
+- 多主体（human/clone）实时通讯
+- 基于 Redis 的异步消息传递
+- 支持点对点消息发送
+- 完全解耦的发送/接收队列机制
+
+## 开发指南
+详见 `docs/development-workflow.md`

app.py

@@ -1,7 +0,0 @@
-from fastapi import FastAPI
-
-app = FastAPI()
-
-@app.get("/")
-def greet_json():
-    return {"Hello": "World!"}

docs/development-workflow.md

@@ -0,0 +1,130 @@
+# Human-Clone 系统开发流程
+
+## 项目概述
+基于需求调研结果的Human-Clone系统开发实施计划
+
+## 开发阶段规划
+
+### 第一阶段：基础设计
+
+#### 1.1 数据结构设计
+- **主体信息结构**
+  - UUID（唯一标识）
+  - name（显示名称）
+  - redis配置（服务器地址、端口、DB、channel）
+  
+- **消息结构**
+  - sender_id（发送者UUID）
+  - receiver_id（接收者UUID）
+  - timestamp（时间戳）
+  - content（消息内容）
+
+#### 1.2 Redis-Adapter组件设计
+- **发送模块**
+  - 发送队列（thread-safe queue）
+  - 发送线程（独立线程循环处理）
+  - Redis连接管理
+  
+- **接收模块**
+  - 接收队列（thread-safe queue）
+  - 接收线程（独立线程循环处理）
+  - 回调注册机制
+  
+- **配置管理**
+  - Redis连接参数
+  - 主体身份信息
+
+#### 1.3 接口设计
+- **主体注册接口**
+  - register_entity(entity_info)
+  - unregister_entity(entity_id)
+  
+- **消息发送接口**
+  - send_message(sender_id, receiver_id, content)
+  
+- **消息接收接口**
+  - register_callback(callback_function)
+  - unregister_callback()
+
+### 第二阶段：核心实现
+
+#### 2.1 消息发送流程实现
+- 构造消息对象
+- 放入发送队列
+- 发送线程异步处理
+- Redis连接和发布
+
+#### 2.2 消息接收流程实现
+- 监听Redis channel
+- 推送到接收队列
+- 接收线程异步处理
+- 回调函数调用
+
+#### 2.3 队列管理实现
+- 线程安全队列设计
+- 生产者-消费者模式
+- 异常处理机制
+
+### 第三阶段：测试验证
+
+#### 3.1 单元测试
+- 数据结构测试
+- Redis-Adapter组件测试
+- 队列操作测试
+- 线程安全测试
+
+#### 3.2 集成测试
+- 发送流程集成测试
+- 接收流程集成测试
+- 端到端通讯测试
+
+#### 3.3 性能测试
+- 并发发送测试
+- 大量消息处理测试
+- 内存和性能监控
+
+### 第四阶段：优化完善
+
+#### 4.1 错误处理
+- 连接失败处理
+- 消息发送失败处理
+- 线程异常处理
+
+#### 4.2 监控日志
+- 消息发送接收日志
+- 性能指标监控
+- 异常情况记录
+
+## 技术栈选择
+
+### 编程语言
+- **推荐**: Python（丰富的Redis库支持，线程处理便利）
+- **备选**: Java/Go（高性能要求场景）
+
+### 核心依赖
+- **Redis客户端**: redis-py (Python)
+- **线程库**: threading/queue (Python原生)
+- **JSON处理**: json (Python原生)
+
+## 开发顺序
+
+1. **数据结构定义** → 2. **Redis连接工具类** → 3. **发送队列模块** → 4. **接收队列模块** → 5. **主控制器** → 6. **测试用例**
+
+## 里程碑
+
+- **M1**: 完成数据结构和Redis连接（第1-2天）
+- **M2**: 完成发送模块（第3-4天）
+- **M3**: 完成接收模块（第5-6天）
+- **M4**: 完成集成测试（第7-8天）
+
+## 风险评估
+
+### 技术风险
+- Redis连接稳定性
+- 线程安全问题
+- 队列性能瓶颈
+
+### 缓解措施
+- 连接池设计
+- 线程安全机制
+- 队列大小监控

docs/research-log.md

@@ -0,0 +1,214 @@
+# Human-Clone 系统需求调研记录
+
+## 调研日期
+2025-01-04
+
+## 项目概述
+Human-Clone 系统需求分析
+
+## 第一轮调研
+
+### 用户回答记录
+**问题**: 这个human-clone系统的主要用途是什么？
+**回答**: human 和 clone 之间相互协作，且可以和其他 clone、human 之间相互通讯的系统。现阶段首先解决的是通讯问题。
+
+### 初步需求分析
+- **核心功能**: 多主体间通讯协作系统
+- **主体类型**: human, clone
+- **通讯模式**: 
+  - human ↔ clone
+  - clone ↔ clone  
+  - human ↔ human (通过系统)
+- **当前优先级**: 通讯功能
+
+### 下一步调研方向
+需要深入了解通讯的具体需求和技术细节。
+
+### 第二轮调研记录
+
+**问题**: 这些主体之间的通讯主要是通过什么形式？文字、语音、还是视频？
+**回答**: 第一步先解决主体之间文字形式的通讯
+
+### 需求更新
+- **通讯方式**: 首先实现文字通讯
+- **开发阶段**: 第一阶段专注文字通讯功能
+- **通讯模式**: 实时异步通讯
+
+### 第三轮调研记录
+
+**问题**: 系统需要支持实时通讯（如聊天室形式）还是异步通讯（如邮件/留言形式）？
+**回答**: 基本属于实施通讯，但是是异步的，计划采用 redis 订阅的方式来实现信息传递，每一个主体都会有一个 redis-adapter，负责发送和接收消息
+
+### 技术架构初步确定
+- **消息传递**: Redis 订阅/发布模式
+- **组件设计**: 每个主体配备 redis-adapter
+- **消息处理**: 发送/接收消息统一处理
+
+### 第四轮调研记录
+
+**问题**: 每个主体（human/clone）需要有自己的身份标识吗？比如用户名、ID之类的？
+**回答**: 需要有 id —— uuid 格式、实体名称（name)、redis-adapter 里和 redis 相关订阅 channel 相关的配置信息
+
+### 主体身份设计
+- **唯一标识**: UUID格式ID
+- **显示名称**: 实体名称(name)
+- **配置信息**: redis-adapter配置（订阅channel等）
+
+### 第五轮调研记录
+
+**问题**: 消息需要有哪些基本字段？比如发送者、接收者、时间戳、消息内容？
+**回答**: 需要，其中发送者、接收者都使用 id
+
+### 消息结构设计
+- **发送者**: UUID格式ID
+- **接收者**: UUID格式ID  
+- **时间戳**: 消息发送时间
+- **消息内容**: 文字内容
+- **消息类型**: 点对点通讯
+
+### 第六轮调研记录
+
+**问题**: 是否需要支持群发消息（一个发送者同时发给多个接收者）？
+**回答**: 此阶段不需要
+
+### 功能范围明确
+- **当前阶段**: 仅支持点对点消息
+- **群发功能**: 后续版本考虑
+
+### 第七轮调研记录
+
+**问题**: 系统需要存储历史消息记录吗？
+**回答**: 需要
+
+### 数据持久化需求
+- **消息存储**: 需要保存历史消息记录
+- **存储方案**: 待确定（数据库选择）
+
+### 第八轮调研记录
+
+**问题**: 历史消息需要保存多长时间？
+**回答**: 先不讨论历史记录的问题，先解决通信问题
+
+### 开发优先级调整
+- **当前重点**: 通讯功能实现
+- **历史记录**: 后续阶段处理
+
+### 第九轮调研记录
+
+**问题**: redis-adapter 需要哪些具体的配置信息？比如 Redis 服务器地址、端口、认证信息等？
+**回答**: redis 服务器地址、端口、db、channel
+
+### redis-adapter 配置设计
+- **服务器地址**: Redis host
+- **端口**: Redis port
+- **数据库**: Redis DB number
+- **频道**: 订阅/发布的 channel
+
+### 第十轮调研记录
+
+**问题**: 每个主体是使用相同的 channel 还是各自有独立的 channel？如果是独立的，channel 的命名规则是什么？
+**回答**: 每个主体都是独立的 channel，channel 的命名就是主体的 id
+
+### Channel 设计方案
+- **Channel 分配**: 每个主体独立 channel
+- **命名规则**: 使用主体的 UUID 作为 channel 名称
+- **通讯机制**: 目标主体监听自己的 channel 接收消息
+
+### 第十一轮调研记录
+
+**问题**: 当一个主体要发送消息给另一个主体时，是如何知道接收者的 channel 名称的？是否需要一个主体注册发现机制？
+**回答**: 主体注册发现机制可以放到下一个迭代，此次只假设可以通过目标主体的 id 就可以获得目标主体的订阅 redis 参数，包括服务器、端口、db、channel
+
+### 主体发现机制设计
+- **当前版本**: 假设可通过 ID 获取目标的 Redis 连接参数
+- **后续迭代**: 实现主体注册发现机制
+- **参数获取**: 服务器地址、端口、DB、channel 通过 ID 直接获得
+
+### 第十二轮调研记录
+
+**问题**: redis-adapter 是否需要处理连接断开重连、消息发送失败重试等异常情况？
+**回答**: 目前不需要
+
+### 异常处理设计
+- **当前版本**: 不处理连接重连和重试机制
+- **设计原则**: 专注核心通讯功能，异常处理后续考虑
+
+### 第十三轮调研记录
+
+**问题**: 消息的发送和接收是否需要确认机制？比如接收者收到消息后需要回复确认？
+**回答**: 不需要
+
+### 消息确认机制
+- **确认机制**: 不需要消息确认
+- **通讯模式**: 单向发送，无回复确认
+- **简化设计**: 降低复杂度，专注基础通讯
+
+### 第十四轮调研记录
+
+**问题**: 系统是否需要区分不同类型的消息？比如普通消息、系统消息、命令消息等？
+**回答**: 不需要
+
+### 消息类型设计
+- **消息分类**: 不区分消息类型
+- **统一处理**: 所有消息均为普通文字消息
+- **简化架构**: 无需消息类型判断和处理逻辑
+
+### 第十五轮调研记录
+
+**问题**: 每个主体的 redis-adapter 是需要同时监听多个 channel（包括自己的和其他主体的）还是只监听自己的 channel？
+**回答**: 只监听自己的
+
+### 监听机制设计
+- **监听范围**: 每个主体只监听自己的 channel
+- **消息接收**: 通过订阅自己的 ID channel 接收消息
+- **消息发送**: 向目标主体的 ID channel 发布消息
+
+### 需求调研总结
+
+基于15轮调研，Human-Clone 系统第一版本的核心需求已明确：
+
+#### 系统架构
+- **主体类型**: human、clone
+- **通讯模式**: 点对点文字通讯
+- **技术方案**: Redis 订阅/发布模式
+- **身份标识**: UUID + name
+
+#### 核心组件
+- **redis-adapter**: 每个主体一个，负责发送/接收消息
+- **配置参数**: 服务器地址、端口、DB、channel(=主体ID)
+- **监听机制**: 只监听自己的 channel
+
+#### 消息格式
+- **发送者ID**: UUID格式
+- **接收者ID**: UUID格式
+- **时间戳**: 发送时间
+- **内容**: 文字消息
+
+#### 功能边界
+- ✅ 点对点文字通讯
+- ❌ 群发消息
+- ❌ 历史消息存储
+- ❌ 连接重连机制
+- ❌ 消息确认机制
+- ❌ 消息类型区分
+- ❌ 主体注册发现
+
+### 第十六轮调研记录
+
+**问题**: 发送消息时是否需要引入发送缓存队列来解耦主体发送和Redis发送操作？
+**回答**: 在发送主体不直接调用发送消息到 redis，而是在 redis-adapter 里设置一个发送缓存 queue，通过线程循环逐个 pop 后发送到 redis，这样主体发送和 redis 发送分开，且有缓存，解耦
+
+### 第十七轮调研记录
+
+**问题**: 消息接收部分是否也需要类似的队列解耦设计？
+**回答**: 消息接收部分，目标主体 B 的redis-adapter 收到 channel 里的消息后，推送到接收队列 receiver-queue，在线程里循环读取 queue，并调用 callback，给到 B 的函数里进行处理
+
+### 完整架构设计优化
+- **发送缓存**: redis-adapter内部设置发送队列
+- **接收缓存**: redis-adapter内部设置接收队列
+- **双向异步**: 发送和接收均采用队列+线程机制
+- **回调机制**: 接收线程通过callback将消息传递给主体
+- **完全解耦**: Redis操作与主体处理完全分离
+
+#### 下一步行动
+基于明确的需求和完整的双向异步架构设计，可以开始进行技术设计和开发实现。

requirements.txt

@@ -1,2 +1,4 @@
-fastapi
-uvicorn[standard]
+redis>=5.0.0
+fastapi>=0.104.0
+uvicorn[standard]>=0.24.0
+pydantic-settings>=2.0.0

roles/requirements-analyst-role.md

@@ -0,0 +1,19 @@
+# 需求分析师角色设定
+
+## 角色定义
+软件系统需求分析师
+
+## 职责
+- 分析和梳理软件系统需求
+- 进行需求调研和评估
+- 编写需求规格文档
+- 协助项目规划和设计
+
+## 工作要求
+1. 深入挖掘系统需求
+2. 每次与用户提问时，一次只提出一个问题
+3. 避免对用户造成不必要的干扰
+4. 将调研过程及报告写入根目录下的 docs 文件夹
+
+## 当前项目
+Human-Clone 系统需求分析

roles/software-engineer-role.md

@@ -0,0 +1,35 @@
+# 软件开发工程师角色设定
+
+## 角色定义
+软件开发工程师 - Human-Clone 系统实现
+
+## 职责
+- 根据需求分析和设计文档进行代码实现
+- 编写高质量、可维护的代码
+- 进行单元测试和集成测试
+- 代码重构和性能优化
+- 技术文档编写
+
+## 技术要求
+- 熟练掌握 Python 编程
+- Redis 操作和 pub/sub 机制
+- 多线程编程和并发处理
+- 队列数据结构和线程安全
+- 软件架构设计和模式应用
+
+## 工作原则
+1. 遵循 SOLID 设计原则
+2. 编写可测试的代码
+3. 关注代码可读性和可维护性
+4. 进行充分的测试验证
+5. 及时更新技术文档
+
+## 当前项目
+Human-Clone 系统核心通讯功能开发
+
+## 开发任务
+- 实现数据结构定义
+- 开发 Redis-Adapter 组件
+- 实现发送/接收队列机制
+- 编写测试用例
+- 性能优化和错误处理

scripts/start_dev.py

@@ -0,0 +1,21 @@
+"""
+开发启动脚本
+"""
+
+import uvicorn
+from src.config import settings
+
+
+def main():
+    """开发环境启动"""
+    uvicorn.run(
+        "src.main:app",
+        host=settings.api_host,
+        port=settings.api_port,
+        reload=settings.debug,
+        log_level=settings.log_level.lower()
+    )
+
+
+if __name__ == "__main__":
+    main()

scripts/start_prod.py

@@ -0,0 +1,22 @@
+"""
+生产启动脚本
+"""
+
+import uvicorn
+from src.config import settings
+
+
+def main():
+    """生产环境启动"""
+    uvicorn.run(
+        "src.main:app",
+        host=settings.api_host,
+        port=settings.api_port,
+        reload=False,
+        log_level=settings.log_level.lower(),
+        workers=4
+    )
+
+
+if __name__ == "__main__":
+    main()

src/__init__.py

@@ -0,0 +1,7 @@
+"""
+Human-Clone System
+多主体通讯协作系统
+"""
+
+__version__ = "1.0.0"
+__author__ = "Human-Clone Team"

src/api/__init__.py

@@ -0,0 +1,7 @@
+"""
+API模块初始化
+"""
+
+from .routes import communication
+
+__all__ = ["communication"]

src/api/routes/__init__.py

@@ -0,0 +1,7 @@
+"""
+路由模块初始化
+"""
+
+from .communication import router
+
+__all__ = ["router"]

src/api/routes/communication.py

@@ -0,0 +1,130 @@
+"""
+通讯API路由
+"""
+
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel
+from typing import List
+import uuid
+from datetime import datetime
+
+from ...core import EntityInfo, Message, RedisAdapter
+
+router = APIRouter()
+
+# 全局存储（生产环境应使用数据库）
+entities: dict[str, EntityInfo] = {}
+adapters: dict[str, RedisAdapter] = {}
+
+
+class EntityCreate(BaseModel):
+    name: str
+    redis_host: str = "localhost"
+    redis_port: int = 6379
+    redis_db: int = 0
+
+
+class MessageSend(BaseModel):
+    sender_id: str
+    receiver_id: str
+    content: str
+
+
+class MessageResponse(BaseModel):
+    sender_id: str
+    receiver_id: str
+    timestamp: str
+    content: str
+
+
+@router.post("/entities", response_model=dict)
+async def create_entity(entity: EntityCreate):
+    """创建新实体"""
+    entity_id = str(uuid.uuid4())
+    channel = entity_id
+    
+    entity_info = EntityInfo(
+        id=entity_id,
+        name=entity.name,
+        redis_host=entity.redis_host,
+        redis_port=entity.redis_port,
+        redis_db=entity.redis_db,
+        channel=channel
+    )
+    
+    entities[entity_id] = entity_info
+    
+    return {
+        "id": entity_id,
+        "name": entity.name,
+        "status": "created"
+    }
+
+
+@router.get("/entities", response_model=List[dict])
+async def list_entities():
+    """获取所有实体列表"""
+    return [
+        {
+            "id": entity_id,
+            "name": entity.name,
+            "redis_host": entity.redis_host,
+            "redis_port": entity.redis_port,
+            "redis_db": entity.redis_db
+        }
+        for entity_id, entity in entities.items()
+    ]
+
+
+@router.post("/entities/{entity_id}/start")
+async def start_entity(entity_id: str):
+    """启动实体通讯"""
+    if entity_id not in entities:
+        raise HTTPException(status_code=404, detail="Entity not found")
+    
+    if entity_id in adapters:
+        raise HTTPException(status_code=400, detail="Entity already started")
+    
+    adapter = RedisAdapter(entities[entity_id])
+    if adapter.start():
+        adapters[entity_id] = adapter
+        return {"status": "started"}
+    else:
+        raise HTTPException(status_code=500, detail="Failed to start entity")
+
+
+@router.post("/entities/{entity_id}/stop")
+async def stop_entity(entity_id: str):
+    """停止实体通讯"""
+    if entity_id not in adapters:
+        raise HTTPException(status_code=404, detail="Entity not started")
+    
+    adapters[entity_id].stop()
+    del adapters[entity_id]
+    
+    return {"status": "stopped"}
+
+
+@router.post("/messages/send")
+async def send_message(message: MessageSend):
+    """发送消息"""
+    if message.sender_id not in adapters:
+        raise HTTPException(status_code=404, detail="Sender not started")
+    
+    if message.receiver_id not in entities:
+        raise HTTPException(status_code=404, detail="Receiver not found")
+    
+    adapter = adapters[message.sender_id]
+    success = adapter.send_message(message.receiver_id, message.content)
+    
+    if success:
+        return {"status": "sent"}
+    else:
+        raise HTTPException(status_code=500, detail="Failed to send message")
+
+
+@router.get("/entities/{entity_id}/messages", response_model=List[MessageResponse])
+async def get_entity_messages(entity_id: str):
+    """获取实体的消息历史（临时存储，演示用）"""
+    # 这里只是示例，实际应该从持久化存储获取
+    return []

src/config/__init__.py

@@ -0,0 +1,7 @@
+"""
+配置模块初始化
+"""
+
+from .settings import settings
+
+__all__ = ["settings"]

src/config/settings.py

@@ -0,0 +1,31 @@
+"""
+配置管理模块
+"""
+
+from pydantic_settings import BaseSettings
+from typing import Optional
+
+
+class Settings(BaseSettings):
+    """系统配置"""
+    
+    # Redis配置
+    redis_host: str = "localhost"
+    redis_port: int = 6379
+    redis_db: int = 0
+    redis_password: Optional[str] = None
+    
+    # API配置
+    api_host: str = "0.0.0.0"
+    api_port: int = 7860
+    debug: bool = True
+    
+    # 日志配置
+    log_level: str = "INFO"
+    
+    class Config:
+        env_file = ".env"
+
+
+# 全局配置实例
+settings = Settings()

src/core/__init__.py

@@ -0,0 +1,9 @@
+"""
+核心模块初始化
+"""
+
+from .entities import EntityInfo
+from .messages import Message
+from .adapter import RedisAdapter
+
+__all__ = ["EntityInfo", "Message", "RedisAdapter"]

src/core/adapter.py

@@ -0,0 +1,237 @@
+"""
+Redis适配器 - 处理消息发送和接收
+"""
+
+import redis
+import json
+import threading
+import queue
+import time
+import logging
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Optional, Callable, Dict, Any
+
+from .entities import EntityInfo
+from .messages import Message
+
+
+# 配置日志
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+# 类型定义
+MessageCallback = Callable[[Message], None]
+
+
+@dataclass
+class RedisConfig:
+    """Redis连接配置"""
+    host: str
+    port: int
+    db: int
+    password: Optional[str] = None
+    
+    def to_connection_params(self) -> Dict[str, Any]:
+        """获取连接参数"""
+        params = {
+            'host': self.host,
+            'port': self.port,
+            'db': self.db
+        }
+        if self.password:
+            params['password'] = self.password
+        return params
+
+
+class RedisAdapter:
+    """Redis适配器 - 处理消息发送和接收"""
+    
+    def __init__(self, entity_info: EntityInfo):
+        self.entity_info = entity_info
+        self.redis_config = RedisConfig(
+            host=entity_info.redis_host,
+            port=entity_info.redis_port,
+            db=entity_info.redis_db
+        )
+        
+        # Redis连接
+        self.redis_client: Optional[redis.Redis] = None
+        
+        # 发送相关
+        self.send_queue = queue.Queue()
+        self.send_thread = None
+        self.send_running = False
+        
+        # 接收相关
+        self.receive_queue = queue.Queue()
+        self.receive_thread = None
+        self.receive_running = False
+        self.message_callback: Optional[MessageCallback] = None
+        
+        # 连接相关
+        self.pubsub: Optional[redis.client.PubSub] = None
+        self.connected = False
+    
+    def connect(self) -> bool:
+        """连接到Redis服务器"""
+        try:
+            self.redis_client = redis.Redis(**self.redis_config.to_connection_params())
+            self.redis_client.ping()  # 测试连接
+            self.connected = True
+            logger.info(f"Entity {self.entity_info.id} connected to Redis")
+            return True
+        except Exception as e:
+            logger.error(f"Failed to connect to Redis: {e}")
+            return False
+    
+    def disconnect(self):
+        """断开Redis连接"""
+        self.stop()
+        if self.redis_client:
+            self.redis_client.close()
+        self.connected = False
+        logger.info(f"Entity {self.entity_info.id} disconnected from Redis")
+    
+    def start(self) -> bool:
+        """启动适配器（发送和接收线程）"""
+        if not self.connect():
+            return False
+        
+        self.send_running = True
+        self.receive_running = True
+        
+        # 启动发送线程
+        self.send_thread = threading.Thread(target=self._send_worker, daemon=True)
+        self.send_thread.start()
+        
+        # 启动接收线程
+        self.receive_thread = threading.Thread(target=self._receive_worker, daemon=True)
+        self.receive_thread.start()
+        
+        logger.info(f"RedisAdapter started for entity {self.entity_info.id}")
+        return True
+    
+    def stop(self):
+        """停止适配器"""
+        self.send_running = False
+        self.receive_running = False
+        
+        # 等待线程结束
+        if self.send_thread and self.send_thread.is_alive():
+            self.send_thread.join(timeout=5)
+        if self.receive_thread and self.receive_thread.is_alive():
+            self.receive_thread.join(timeout=5)
+        
+        logger.info(f"RedisAdapter stopped for entity {self.entity_info.id}")
+    
+    def send_message(self, receiver_id: str, content: str) -> bool:
+        """发送消息（异步）"""
+        try:
+            message = Message(
+                sender_id=self.entity_info.id,
+                receiver_id=receiver_id,
+                timestamp=datetime.now(),
+                content=content
+            )
+            self.send_queue.put(message)
+            logger.debug(f"Message queued for {receiver_id}")
+            return True
+        except Exception as e:
+            logger.error(f"Failed to queue message: {e}")
+            return False
+    
+    def register_callback(self, callback: MessageCallback):
+        """注册消息接收回调函数"""
+        self.message_callback = callback
+    
+    def _send_worker(self):
+        """发送工作线程"""
+        logger.info("Send worker thread started")
+        
+        while self.send_running:
+            try:
+                # 从队列获取消息（超时1秒）
+                message = self.send_queue.get(timeout=1)
+                
+                # 这里需要获取接收者的Redis连接信息
+                # 当前简化处理，使用同一个Redis实例
+                target_channel = message.receiver_id
+                
+                # 发送到Redis
+                if self.redis_client:
+                    self.redis_client.publish(
+                        target_channel,
+                        json.dumps(message.to_dict())
+                    )
+                    logger.debug(f"Message sent to channel {target_channel}")
+                
+                self.send_queue.task_done()
+                
+            except queue.Empty:
+                continue
+            except Exception as e:
+                logger.error(f"Error in send worker: {e}")
+        
+        logger.info("Send worker thread stopped")
+    
+    def _receive_worker(self):
+        """接收工作线程"""
+        logger.info("Receive worker thread started")
+        
+        if not self.redis_client:
+            logger.error("Redis client not available for receiving")
+            return
+        
+        try:
+            # 创建pubsub对象
+            self.pubsub = self.redis_client.pubsub()
+            self.pubsub.subscribe(self.entity_info.channel)
+            
+            while self.receive_running:
+                try:
+                    # 获取消息（超时1秒）
+                    message = self.pubsub.get_message(timeout=1)
+                    
+                    if message and message['type'] == 'message':
+                        # 解析消息数据
+                        message_data = json.loads(message['data'].decode('utf-8'))
+                        received_message = Message.from_dict(message_data)
+                        
+                        # 放入接收队列
+                        self.receive_queue.put(received_message)
+                        
+                        # 处理接收队列中的消息
+                        self._process_receive_queue()
+                        
+                except Exception as e:
+                    logger.error(f"Error processing received message: {e}")
+                    
+        except Exception as e:
+            logger.error(f"Error in receive worker: {e}")
+        finally:
+            if self.pubsub:
+                self.pubsub.close()
+        
+        logger.info("Receive worker thread stopped")
+    
+    def _process_receive_queue(self):
+        """处理接收队列中的消息"""
+        try:
+            while not self.receive_queue.empty():
+                message = self.receive_queue.get_nowait()
+                
+                if self.message_callback:
+                    try:
+                        self.message_callback(message)
+                        logger.debug(f"Message delivered to callback")
+                    except Exception as e:
+                        logger.error(f"Error in message callback: {e}")
+                
+                self.receive_queue.task_done()
+                
+        except queue.Empty:
+            pass
+        except Exception as e:
+            logger.error(f"Error processing receive queue: {e}")

src/core/entities.py

@@ -0,0 +1,25 @@
+"""
+实体信息结构
+"""
+
+from dataclasses import dataclass
+from typing import Optional
+import uuid
+
+
+@dataclass
+class EntityInfo:
+    """主体信息结构"""
+    id: str  # UUID格式唯一标识
+    name: str  # 显示名称
+    redis_host: str  # Redis服务器地址
+    redis_port: int  # Redis端口
+    redis_db: int  # Redis数据库编号
+    channel: str  # 订阅channel名称
+    
+    def __post_init__(self):
+        """数据验证"""
+        if not self.id:
+            self.id = str(uuid.uuid4())
+        if not self.channel:
+            self.channel = self.id

src/core/messages.py

@@ -0,0 +1,35 @@
+"""
+消息结构定义
+"""
+
+from dataclasses import dataclass
+from typing import Dict, Any
+from datetime import datetime
+
+
+@dataclass
+class Message:
+    """消息结构"""
+    sender_id: str  # 发送者UUID
+    receiver_id: str  # 接收者UUID
+    timestamp: datetime  # 时间戳
+    content: str  # 消息内容
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """转换为字典格式"""
+        return {
+            'sender_id': self.sender_id,
+            'receiver_id': self.receiver_id,
+            'timestamp': self.timestamp.isoformat(),
+            'content': self.content
+        }
+    
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> 'Message':
+        """从字典创建消息对象"""
+        return cls(
+            sender_id=data['sender_id'],
+            receiver_id=data['receiver_id'],
+            timestamp=datetime.fromisoformat(data['timestamp']),
+            content=data['content']
+        )

src/main.py

@@ -0,0 +1,33 @@
+"""
+FastAPI应用主入口
+"""
+
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from src.api.routes.communication import router as comm_router
+
+app = FastAPI(
+    title="Human-Clone API",
+    description="Human-Clone系统通讯接口",
+    version="1.0.0"
+)
+
+# CORS中间件
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# 注册路由
+app.include_router(comm_router, prefix="/api/v1")
+
+@app.get("/")
+async def root():
+    return {"message": "Human-Clone API"}
+
+@app.get("/health")
+async def health_check():
+    return {"status": "healthy"}

src/utils/__init__.py

@@ -0,0 +1,7 @@
+"""
+工具模块初始化
+"""
+
+from .logger import setup_logger
+
+__all__ = ["setup_logger"]

src/utils/logger.py

@@ -0,0 +1,30 @@
+"""
+日志工具
+"""
+
+import logging
+import sys
+from typing import Optional
+
+
+def setup_logger(
+    name: str,
+    level: str = "INFO",
+    format_string: Optional[str] = None
+) -> logging.Logger:
+    """设置日志器"""
+    
+    if format_string is None:
+        format_string = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+    
+    logger = logging.getLogger(name)
+    logger.setLevel(getattr(logging, level.upper()))
+    
+    # 避免重复添加handler
+    if not logger.handlers:
+        handler = logging.StreamHandler(sys.stdout)
+        formatter = logging.Formatter(format_string)
+        handler.setFormatter(formatter)
+        logger.addHandler(handler)
+    
+    return logger

tests/__init__.py

@@ -0,0 +1,3 @@
+"""
+测试模块初始化
+"""

tests/fixtures/__init__.py

@@ -0,0 +1,7 @@
+"""
+测试数据模块初始化
+"""
+
+from .test_data import create_test_entity_a, create_test_entity_b, get_test_messages
+
+__all__ = ["create_test_entity_a", "create_test_entity_b", "get_test_messages"]

tests/fixtures/test_data.py

@@ -0,0 +1,38 @@
+"""
+测试数据fixture
+"""
+
+from src.core import EntityInfo
+
+
+def create_test_entity_a():
+    """创建测试实体A"""
+    return EntityInfo(
+        id="test-entity-a-001",
+        name="Test Entity A",
+        redis_host="localhost",
+        redis_port=6379,
+        redis_db=0,
+        channel="test-entity-a-001"
+    )
+
+
+def create_test_entity_b():
+    """创建测试实体B"""
+    return EntityInfo(
+        id="test-entity-b-002", 
+        name="Test Entity B",
+        redis_host="localhost",
+        redis_port=6379,
+        redis_db=0,
+        channel="test-entity-b-002"
+    )
+
+
+def get_test_messages():
+    """获取测试消息列表"""
+    return [
+        "Hello from Test Entity A!",
+        "This is a test message",
+        "Human-Clone system working"
+    ]

tests/integration/__init__.py

@@ -0,0 +1,3 @@
+"""
+集成测试模块初始化
+"""

tests/integration/test_communication.py

@@ -0,0 +1,85 @@
+"""
+Human-Clone 系统集成测试
+"""
+
+import sys
+import os
+sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
+
+import unittest
+import time
+import threading
+from datetime import datetime
+from src.core import EntityInfo, RedisAdapter
+from tests.fixtures import create_test_entity_a, create_test_entity_b, get_test_messages
+
+
+class TestCommunication(unittest.TestCase):
+    """通讯集成测试类"""
+    
+    def test_basic_communication(self):
+        """测试基础通讯功能"""
+    print("=== Human-Clone 系统通讯测试 ===")
+    
+    # 创建测试主体
+    entity_a = create_test_entity_a()
+    entity_b = create_test_entity_b()
+    
+    # 创建适配器
+    adapter_a = RedisAdapter(entity_a)
+    adapter_b = RedisAdapter(entity_b)
+    
+    # 消息接收回调
+    received_messages = []
+    
+    def message_callback(message):
+        received_messages.append(message)
+        print(f"[{message.receiver_id}] 收到来自 {message.sender_id} 的消息: {message.content}")
+    
+    # 启动适配器
+    print("\n1. 启动适配器...")
+    if adapter_a.start() and adapter_b.start():
+        print("✓ 适配器启动成功")
+        
+        # 注册回调
+        adapter_b.register_callback(message_callback)
+        
+        # 等待连接建立
+        time.sleep(1)
+        
+        # 发送测试消息
+        print("\n2. 发送测试消息...")
+        test_messages = get_test_messages()
+        
+        for msg in test_messages:
+            if adapter_a.send_message(entity_b.id, msg):
+                print(f"✓ 消息已发送: {msg}")
+            time.sleep(0.5)
+        
+        # 等待消息接收
+        print("\n3. 等待消息接收...")
+        time.sleep(2)
+        
+        # 检查结果
+        print(f"\n4. 测试结果:")
+        print(f"   发送消息数: {len(test_messages)}")
+        print(f"   接收消息数: {len(received_messages)}")
+        
+        if len(received_messages) == len(test_messages):
+            print("✓ 通讯测试通过!")
+        else:
+            print("✗ 通讯测试失败!")
+        
+    else:
+        print("✗ 适配器启动失败")
+    
+    # 清理资源
+    print("\n5. 清理资源...")
+    adapter_a.stop()
+    adapter_b.stop()
+    
+    print("\n=== 测试完成 ===")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/run_tests.py

@@ -0,0 +1,53 @@
+"""
+批量运行所有测试
+"""
+
+import unittest
+import sys
+import os
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+
+def run_unit_tests():
+    """运行单元测试"""
+    print("=== 运行单元测试 ===")
+    loader = unittest.TestLoader()
+    suite = loader.discover('tests/unit', pattern='test_*.py')
+    runner = unittest.TextTestRunner(verbosity=2)
+    result = runner.run(suite)
+    return result.wasSuccessful()
+
+
+def run_integration_tests():
+    """运行集成测试"""
+    print("\n=== 运行集成测试 ===")
+    loader = unittest.TestLoader()
+    suite = loader.discover('tests/integration', pattern='test_*.py')
+    runner = unittest.TextTestRunner(verbosity=2)
+    result = runner.run(suite)
+    return result.wasSuccessful()
+
+
+def run_all_tests():
+    """运行所有测试"""
+    print("Human-Clone 系统测试套件")
+    print("=" * 50)
+    
+    unit_success = run_unit_tests()
+    integration_success = run_integration_tests()
+    
+    print("\n=== 测试结果汇总 ===")
+    print(f"单元测试: {'✓ 通过' if unit_success else '✗ 失败'}")
+    print(f"集成测试: {'✓ 通过' if integration_success else '✗ 失败'}")
+    
+    if unit_success and integration_success:
+        print("🎉 所有测试通过!")
+        return True
+    else:
+        print("❌ 存在测试失败")
+        return False
+
+
+if __name__ == "__main__":
+    success = run_all_tests()
+    sys.exit(0 if success else 1)

tests/unit/__init__.py

@@ -0,0 +1,3 @@
+"""
+单元测试模块初始化
+"""

tests/unit/test_entities.py

@@ -0,0 +1,63 @@
+"""
+实体信息测试
+"""
+
+import sys
+import os
+sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
+
+import unittest
+from src.core import EntityInfo
+
+
+class TestEntityInfo(unittest.TestCase):
+    """测试EntityInfo类"""
+    
+    def test_entity_creation(self):
+        """测试实体创建"""
+        entity = EntityInfo(
+            id="test-001",
+            name="Test Entity",
+            redis_host="localhost",
+            redis_port=6379,
+            redis_db=0,
+            channel="test-channel"
+        )
+        
+        self.assertEqual(entity.id, "test-001")
+        self.assertEqual(entity.name, "Test Entity")
+        self.assertEqual(entity.redis_host, "localhost")
+        self.assertEqual(entity.redis_port, 6379)
+        self.assertEqual(entity.redis_db, 0)
+        self.assertEqual(entity.channel, "test-channel")
+    
+    def test_auto_uuid_generation(self):
+        """测试UUID自动生成"""
+        entity = EntityInfo(
+            id="",  # 空ID应该自动生成UUID
+            name="Test Entity",
+            redis_host="localhost",
+            redis_port=6379,
+            redis_db=0,
+            channel="test-channel"
+        )
+        
+        self.assertIsNotNone(entity.id)
+        self.assertNotEqual(entity.id, "")
+    
+    def test_auto_channel_assignment(self):
+        """测试channel自动赋值"""
+        entity = EntityInfo(
+            id="test-001",
+            name="Test Entity",
+            redis_host="localhost",
+            redis_port=6379,
+            redis_db=0,
+            channel=""  # 空channel应该自动使用ID
+        )
+        
+        self.assertEqual(entity.channel, "test-001")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_messages.py

@@ -0,0 +1,69 @@
+"""
+消息结构测试
+"""
+
+import sys
+import os
+sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
+
+import unittest
+from datetime import datetime
+from src.core import Message
+
+
+class TestMessage(unittest.TestCase):
+    """测试Message类"""
+    
+    def test_message_creation(self):
+        """测试消息创建"""
+        timestamp = datetime.now()
+        message = Message(
+            sender_id="sender-001",
+            receiver_id="receiver-001", 
+            timestamp=timestamp,
+            content="Test message content"
+        )
+        
+        self.assertEqual(message.sender_id, "sender-001")
+        self.assertEqual(message.receiver_id, "receiver-001")
+        self.assertEqual(message.timestamp, timestamp)
+        self.assertEqual(message.content, "Test message content")
+    
+    def test_message_to_dict(self):
+        """测试消息转换为字典"""
+        timestamp = datetime.now()
+        message = Message(
+            sender_id="sender-001",
+            receiver_id="receiver-001",
+            timestamp=timestamp,
+            content="Test message"
+        )
+        
+        message_dict = message.to_dict()
+        
+        self.assertEqual(message_dict['sender_id'], "sender-001")
+        self.assertEqual(message_dict['receiver_id'], "receiver-001")
+        self.assertEqual(message_dict['timestamp'], timestamp.isoformat())
+        self.assertEqual(message_dict['content'], "Test message")
+    
+    def test_message_from_dict(self):
+        """测试从字典创建消息"""
+        timestamp = datetime.now()
+        message_dict = {
+            'sender_id': 'sender-001',
+            'receiver_id': 'receiver-001',
+            'timestamp': timestamp.isoformat(),
+            'content': 'Test message'
+        }
+        
+        message = Message.from_dict(message_dict)
+        
+        self.assertEqual(message.sender_id, "sender-001")
+        self.assertEqual(message.receiver_id, "receiver-001")
+        self.assertEqual(message.content, "Test message")
+        # 比较datetime对象可能会有毫秒级差异，所以转换为字符串比较
+        self.assertEqual(message.timestamp.isoformat(), timestamp.isoformat())
+
+
+if __name__ == "__main__":
+    unittest.main()