Files

T

stefanfeng 76f0730b57 docs: add avatar takeover chat implementation plan

2026-08-07 16:22:58 +08:00

31 KiB

Raw Blame History

数字分身接管聊天功能 — 实现计划

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 实现数字分身接管主人在会会平台上的聊天，支持立即接管和延迟接管两种模式，优先使用知识库生成回复。

Architecture: 在数字分身应用后端新增 takeover 模块，包含消息监听器、接管决策器、回复执行器。通过轮询会会平台推送消息接口获取未读消息，调用现有分身聊天接口生成回复，通过盒子 IM 接口以主人身份发送。

Tech Stack: FastAPI (Python), SQLite, Redis, HTTPX, APScheduler, Vue 3 (frontend)

全局约束

数据库：SQLite (avatar.db)，使用 _try_add_columns 模式添加新列
响应格式：统一使用 ok() / fail() from responses.py
认证：Authorization: Bearer <app_token> header
盒子 IM 接口前缀：/api/im/box/*（通过会会平台网关）
签名机制：复用 news_service.py 中的 _build_form / _make_sign 模式
YAGNI：本期只做单聊接管，群聊接管标记为后续扩展
TDD：每个功能模块先写测试再实现

文件清单

新建文件

文件	职责
`digital-avatar-app/backend/routers/takeover.py`	接管配置 API + 内部处理接口
`digital-avatar-app/backend/services/takeover_service.py`	接管核心逻辑（监听、决策、回复）
`digital-avatar-app/backend/services/boxim_client.py`	盒子 IM 客户端（获取凭证、发送消息）
`digital-avatar-app/backend/requirements.txt`	新增 redis、apscheduler 依赖
`digital-avatar-app/src/api/index.ts`	新增 takeover 相关 API 函数
`digital-avatar-app/src/views/AuthorizationManage.vue`	添加接管开关 UI

修改文件

文件	修改内容
`digital-avatar-app/backend/models.py`	Authorization 表新增 takeover 字段
`digital-avatar-app/backend/database.py`	`_try_add_columns` 添加 takeover 列
`digital-avatar-app/backend/main.py`	注册 takeover router，启动定时任务
`digital-avatar-app/backend/requirements.txt`	添加 redis, apscheduler

任务分解

Task 1: 扩展 Authorization 数据模型

Files:

Modify: digital-avatar-app/backend/models.py
Modify: digital-avatar-app/backend/database.py
Test: digital-avatar-app/backend/tests/test_takeover_model.py

Interfaces:

新增字段：takeover_enabled, takeover_mode, takeover_delay_seconds
Authorization.to_dict() 返回新增字段
Step 1: Write the failing test

# digital-avatar-app/backend/tests/test_takeover_model.py
from database import SessionLocal
from models import Authorization

def test_authorization_takeover_fields():
    db = SessionLocal()
    try:
        auth = db.query(Authorization).first()
        assert auth is not None
        # 检查新字段存在且有默认值
        assert hasattr(auth, 'takeover_enabled')
        assert hasattr(auth, 'takeover_mode')
        assert hasattr(auth, 'takeover_delay_seconds')
        assert auth.takeover_enabled == False
        assert auth.takeover_mode == 'immediate'
        assert auth.takeover_delay_seconds == 30
    finally:
        db.close()

def test_authorization_to_dict_includes_takeover():
    db = SessionLocal()
    try:
        auth = db.query(Authorization).first()
        d = auth.to_dict()
        assert 'takeover_enabled' in d
        assert 'takeover_mode' in d
        assert 'takeover_delay_seconds' in d
    finally:
        db.close()

Step 2: Run test to verify it fails

cd digital-avatar-app/backend
python -m pytest tests/test_takeover_model.py -v

Expected: FAIL with "no such column: takeover_enabled"

Step 3: Update Authorization model

# digital-avatar-app/backend/models.py
# 在 Authorization 类中新增字段（在 created_at 之前）
class Authorization(Base):
    # ... 现有字段 ...
    takeover_enabled = Column(Boolean, default=False)  # 是否开启分身接管
    takeover_mode = Column(String, default="immediate")  # immediate | delayed
    takeover_delay_seconds = Column(Integer, default=30)  # 延迟秒数
    created_at = Column(DateTime, server_default=func.now())

    def to_dict(self):
        return {
            # ... 现有字段 ...
            "takeoverEnabled": self.takeover_enabled,
            "takeoverMode": self.takeover_mode,
            "takeoverDelaySeconds": self.takeover_delay_seconds,
            "createdAt": _iso(self.created_at),
        }

Step 4: Add column migration

# digital-avatar-app/backend/database.py
# 在 _try_add_columns 调用中新增：
_try_add_columns(
    # ... 现有列 ...
    ("authorizations", "takeover_enabled", "BOOLEAN DEFAULT 0"),
    ("authorizations", "takeover_mode", "VARCHAR DEFAULT 'immediate'"),
    ("authorizations", "takeover_delay_seconds", "INTEGER DEFAULT 30"),
)

Step 5: Run test to verify it passes

cd digital-avatar-app/backend
python -m pytest tests/test_takeover_model.py -v

Expected: PASS

Step 6: Commit

cd /Users/yqq/Works/AiProject/huihuiSquare
git add digital-avatar-app/backend/models.py digital-avatar-app/backend/database.py digital-avatar-app/backend/tests/test_takeover_model.py
git commit -m "feat: add takeover fields to Authorization model"

Task 2: 盒子 IM 客户端

Files:

Create: digital-avatar-app/backend/services/boxim_client.py
Test: digital-avatar-app/backend/tests/test_boxim_client.py

Interfaces:

BoxIMClient.get_credentials(user_id: str) -> dict — 获取网易云信凭证
BoxIMClient.send_p2p_message(from_accid: str, to_accid: str, content: str) -> bool — 发送单聊消息

Dependencies:

会会平台签名机制（复用 news_service 中的模式）
HTTPX for async HTTP calls
平台配置从环境变量读取
Step 1: Write the failing test

# digital-avatar-app/backend/tests/test_boxim_client.py
import pytest
from unittest.mock import AsyncMock, patch
from services.boxim_client import BoxIMClient

@pytest.fixture
def mock_config():
    return {
        "HUIHUI_IM_BASE_URL": "http://192.168.1.200:60040",
        "HUIHUI_APP_ID": "test_app",
        "HUIHUI_ACCESS_ID": "test_access",
        "HUIHUI_ACCESS_SECRET": "test_secret",
    }

@patch("services.boxim_client.httpx.AsyncClient")
@pytest.mark.asyncio
async def test_get_credentials(mock_client_class, mock_config):
    mock_response = AsyncMock()
    mock_response.json.return_value = {"code": 200, "data": {"accid": "user123", "token": "tok_xyz"}}
    mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response

    client = BoxIMClient(mock_config)
    result = await client.get_credentials("user123")

    assert result["accid"] == "user123"
    assert result["token"] == "tok_xyz"

@patch("services.boxim_client.httpx.AsyncClient")
@pytest.mark.asyncio
async def test_send_p2p_message_success(mock_client_class, mock_config):
    mock_response = AsyncMock()
    mock_response.json.return_value = {"code": 200}
    mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response

    client = BoxIMClient(mock_config)
    result = await client.send_p2p_message("owner_acc", "target_acc", "Hello")

    assert result is True

Step 2: Run test to verify it fails

cd digital-avatar-app/backend
python -m pytest tests/test_boxim_client.py -v

Expected: FAIL with "ModuleNotFoundError: No module named 'services.boxim_client'"

Step 3: Write minimal implementation

# digital-avatar-app/backend/services/boxim_client.py
"""盒子 IM 客户端 — 封装网易云信 IM 接口调用"""
import hashlib
import random
import string
from datetime import datetime
from typing import Optional
import httpx


class BoxIMClient:
    """盒子 IM 客户端，通过会会平台网关调用网易云信 IM"""

    def __init__(self, config: dict):
        self.base_url = config.get("HUIHUI_IM_BASE_URL", "http://192.168.1.200:60040")
        self.app_id = config.get("HUIHUI_APP_ID", "")
        self.access_id = config.get("HUIHUI_ACCESS_ID", "")
        self.access_secret = config.get("HUIHUI_ACCESS_SECRET", "")

    def _build_sign_params(self, extra: dict) -> dict:
        """构建带签名的请求参数"""
        nonce = "".join(random.choices(string.ascii_lowercase + string.digits, k=12))
        timestamp = datetime.now().strftime("%Y%m%d%H%M%S")  # 24小时制
        params = {
            "appId": self.app_id,
            "accessId": self.access_id,
            "nonce": nonce,
            "timestamp": timestamp,
            **extra,
        }
        # 计算签名
        keys = sorted(params.keys())
        sign_parts = []
        for k in keys:
            v = params.get(k)
            if v and v != "" and v != []:
                sign_parts.append(f"{k}={v}")
        sign_str = "&".join(sign_parts) + f"&accessSecret={self.access_secret}"
        signature = hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
        params["signature"] = signature
        params["signType"] = "MD5"
        params["signVersion"] = "1.0"
        return params

    async def get_credentials(self, user_id: str) -> Optional[dict]:
        """获取用户的网易云信 IM 凭证 (accid, token)"""
        params = self._build_sign_params({"userId": user_id})
        async with httpx.AsyncClient(timeout=10) as client:
            r = await client.post(
                f"{self.base_url}/box/netease",
                params=params,
            )
        data = r.json()
        if data.get("code") in (0, 200):
            return data.get("data", {})
        return None

    async def send_p2p_message(
        self, from_accid: str, to_accid: str, content: str
    ) -> bool:
        """发送单聊消息（文本）"""
        params = self._build_sign_params({
            "from": from_accid,
            "to": to_accid,
            "msgType": "text",
            "content": content,
        })
        async with httpx.AsyncClient(timeout=10) as client:
            r = await client.post(
                f"{self.base_url}/box/message/send/p2p",
                params=params,
            )
        data = r.json()
        return data.get("code") in (0, 200)

Step 4: Add test dependencies

cd digital-avatar-app/backend
pip install pytest pytest-asyncio httpx

Step 5: Run test to verify it passes

cd digital-avatar-app/backend
python -m pytest tests/test_boxim_client.py -v

Expected: PASS

Step 6: Commit

git add digital-avatar-app/backend/services/boxim_client.py digital-avatar-app/backend/tests/test_boxim_client.py
git commit -m "feat: add Box IM client for Netease Yunxin integration"

Task 3: 接管服务核心逻辑

Files:

Create: digital-avatar-app/backend/services/takeover_service.py
Test: digital-avatar-app/backend/tests/test_takeover_service.py

Interfaces:

TakeoverService.check_takeover_enabled(owner_huihui_id: str, from_user_id: str) -> Authorization | None
TakeoverService.generate_reply(avatar_id: str, message: str) -> str
TakeoverService.execute_takeover(auth: Authorization, message: dict) -> bool
TakeoverService.process_delayed_queue() — 处理延迟队列中到期的消息

Dependencies:

BoxIMClient from Task 2
Existing chat API: POST /api/avatar/{id}/chat
Redis for delayed queue
Authorization model from Task 1
Step 1: Write the failing test

# digital-avatar-app/backend/tests/test_takeover_service.py
import pytest
from unittest.mock import AsyncMock, patch, MagicMock
from services.takeover_service import TakeoverService

@pytest.fixture
def mock_db():
    db = MagicMock()
    return db

@pytest.fixture
def mock_boxim():
    client = AsyncMock()
    client.get_credentials.return_value = {"accid": "owner_acc", "token": "tok"}
    client.send_p2p_message.return_value = True
    return client

@pytest.fixture
def mock_auth():
    auth = MagicMock()
    auth.takeover_enabled = True
    auth.takeover_mode = "immediate"
    auth.takeover_delay_seconds = 30
    auth.avatar_id = "avatar_123"
    auth.target_id = "target_user_123"
    return auth

def test_check_takeover_enabled_returns_auth_when_enabled(mock_db, mock_auth):
    mock_db.query.return_value.filter.return_value.filter.return_value.first.return_value = mock_auth
    service = TakeoverService(mock_db, None)
    result = service.check_takeover_enabled("owner_123", "target_123")
    assert result == mock_auth

def test_check_takeover_enabled_returns_none_when_disabled(mock_db):
    disabled_auth = MagicMock()
    disabled_auth.takeover_enabled = False
    mock_db.query.return_value.filter.return_value.filter.return_value.first.return_value = disabled_auth
    service = TakeoverService(mock_db, None)
    result = service.check_takeover_enabled("owner_123", "target_123")
    assert result is None

Step 2: Run test to verify it fails

cd digital-avatar-app/backend
python -m pytest tests/test_takeover_service.py -v

Expected: FAIL

Step 3: Write minimal implementation

# digital-avatar-app/backend/services/takeover_service.py
"""分身接管聊天服务 — 消息监听、决策、回复执行"""
import json
import logging
from typing import Optional
from sqlalchemy.orm import Session
import httpx

from models import Authorization, Avatar
from services.boxim_client import BoxIMClient

logger = logging.getLogger(__name__)


class TakeoverService:
    """分身接管服务"""

    def __init__(self, db: Session, boxim_client: BoxIMClient, redis_client=None):
        self.db = db
        self.boxim = boxim_client
        self.redis = redis_client
        self._chat_api_base = "http://localhost:8000/api"  # 分身应用自己的 API

    def check_takeover_enabled(
        self, owner_huihui_id: str, from_user_id: str
    ) -> Optional[Authorization]:
        """检查是否有接管授权"""
        auth = (
            self.db.query(Authorization)
            .filter(Authorization.target_id == from_user_id)
            .filter(Authorization.takeover_enabled == True)
            .first()
        )
        if auth and auth.takeover_enabled:
            return auth
        return None

    async def generate_reply(self, avatar_id: str, message: str) -> str:
        """调用分身聊天接口生成回复"""
        try:
            async with httpx.AsyncClient(timeout=30) as client:
                r = await client.post(
                    f"{self._chat_api_base}/avatar/{avatar_id}/chat",
                    json={"message": message, "history": []},
                )
            data = r.json()
            if data.get("code") in (0, 200):
                return data.get("data", {}).get("answer", "")
            logger.warning(f"分身聊天接口返回异常: {data}")
            return ""
        except Exception as e:
            logger.error(f"调用分身聊天接口失败: {e}")
            return ""

    async def execute_takeover(self, auth: Authorization, message: dict) -> bool:
        """执行接管：生成回复并以主人身份发送"""
        try:
            # 1. 获取主人的 IM 凭证
            owner_huihui_id = auth.owner_id if hasattr(auth, 'owner_id') else ""
            credentials = await self.boxim.get_credentials(owner_huihui_id)
            if not credentials:
                logger.warning(f"无法获取主人 IM 凭证: {owner_huihui_id}")
                return False

            # 2. 生成回复
            reply = await self.generate_reply(auth.avatar_id, message.get("content", ""))
            if not reply:
                logger.warning("分身未生成回复")
                return False

            # 3. 发送消息
            success = await self.boxim.send_p2p_message(
                from_accid=credentials["accid"],
                to_accid=message.get("from_accid", ""),
                content=reply,
            )
            if success:
                logger.info(f"分身接管回复成功: {reply[:50]}...")
            return success
        except Exception as e:
            logger.error(f"执行接管失败: {e}")
            return False

    def enqueue_delayed_message(self, auth: Authorization, message: dict):
        """将消息写入 Redis 延迟队列"""
        if not self.redis:
            logger.warning("Redis 未配置，降级为立即接管")
            return
        key = f"takeover:delayed:{auth.target_id}:{message.get('msg_id', '')}"
        value = json.dumps({
            "avatar_id": auth.avatar_id,
            "from_accid": message.get("from_accid", ""),
            "content": message.get("content", ""),
            "owner_huihui_id": auth.owner_id if hasattr(auth, 'owner_id') else "",
        })
        self.redis.setex(key, auth.takeover_delay_seconds + 10, value)
        logger.info(f"消息写入延迟队列: {key}")

    async def process_delayed_queue(self):
        """处理延迟队列中到期的消息"""
        if not self.redis:
            return
        # Redis 会自动过期，这里不需要主动处理
        # 实际使用时可以用 Redis keyspace notifications 或定期扫描
        pass

Step 4: Run test to verify it passes

cd digital-avatar-app/backend
python -m pytest tests/test_takeover_service.py -v

Expected: PASS

Step 5: Commit

git add digital-avatar-app/backend/services/takeover_service.py digital-avatar-app/backend/tests/test_takeover_service.py
git commit -m "feat: add takeover service core logic"

Task 4: 接管配置 API

Files:

Create: digital-avatar-app/backend/routers/takeover.py
Modify: digital-avatar-app/backend/main.py

Interfaces:

PUT /api/avatar/{avatar_id}/authorizations/takeover — 更新接管配置
请求体：{ authorization_id, takeover_enabled, takeover_mode, takeover_delay_seconds }

Dependencies:

Authorization model from Task 1
ok() / fail() from responses.py
Step 1: Write the failing test

# digital-avatar-app/backend/tests/test_takeover_api.py
from fastapi.testclient import TestClient
from main import app
from database import SessionLocal, Base, engine
from models import Authorization, Avatar

def setup_test_db():
    Base.metadata.create_all(bind=engine)
    db = SessionLocal()
    avatar = Avatar(name="test", status="active", config={})
    db.add(avatar)
    db.commit()
    auth = Authorization(avatar_id=avatar.id, target_id="user1", target_name="测试用户")
    db.add(auth)
    db.commit()
    return db, auth.id

def test_update_takeover_config():
    db, auth_id = setup_test_db()
    client = TestClient(app)
    response = client.put(
        f"/api/avatar/test_auth/avatar_id/authorizations/takeover",
        json={
            "authorization_id": auth_id,
            "takeover_enabled": True,
            "takeover_mode": "delayed",
            "takeover_delay_seconds": 60,
        },
    )
    assert response.status_code == 200
    data = response.json()
    assert data["code"] == 200
    # 验证数据库已更新
    auth = db.query(Authorization).filter(Authorization.id == auth_id).first()
    assert auth.takeover_enabled == True
    assert auth.takeover_mode == "delayed"
    assert auth.takeover_delay_seconds == 60

Step 2: Run test to verify it fails

cd digital-avatar-app/backend
python -m pytest tests/test_takeover_api.py -v

Expected: FAIL with 404 (endpoint not found)

Step 3: Write minimal implementation

# digital-avatar-app/backend/routers/takeover.py
"""分身接管配置 API"""
from fastapi import APIRouter, Depends, Body
from sqlalchemy.orm import Session

from database import get_db
from models import Authorization
from responses import ok, fail

router = APIRouter(tags=["分身接管"])


@router.put("/avatar/{avatar_id}/authorizations/takeover")
def update_takeover_config(
    avatar_id: str,
    payload: dict = Body(...),
    db: Session = Depends(get_db),
):
    """更新分身接管配置"""
    auth_id = payload.get("authorization_id")
    if not auth_id:
        return fail("缺少 authorization_id", 400)

    auth = db.query(Authorization).filter(Authorization.id == auth_id).first()
    if not auth:
        return fail("授权不存在", 404)

    # 更新接管配置
    if "takeover_enabled" in payload:
        auth.takeover_enabled = payload["takeover_enabled"]
    if "takeover_mode" in payload:
        mode = payload["takeover_mode"]
        if mode not in ("immediate", "delayed"):
            return fail("takeover_mode 必须是 immediate 或 delayed", 400)
        auth.takeover_mode = mode
    if "takeover_delay_seconds" in payload:
        delay = payload["takeover_delay_seconds"]
        if not isinstance(delay, int) or delay < 5:
            return fail("takeover_delay_seconds 必须 >= 5", 400)
        auth.takeover_delay_seconds = delay

    db.commit()
    return ok(auth.to_dict())

Step 4: Register router in main.py

# digital-avatar-app/backend/main.py
# 在 import 部分添加：
import routers.takeover

# 在 router 注册部分添加：
app.include_router(routers.takeover.router, prefix="/api")

Step 5: Run test to verify it passes

cd digital-avatar-app/backend
python -m pytest tests/test_takeover_api.py -v

Expected: PASS

Step 6: Commit

git add digital-avatar-app/backend/routers/takeover.py digital-avatar-app/backend/main.py digital-avatar-app/backend/tests/test_takeover_api.py
git commit -m "feat: add takeover config API endpoint"

Task 5: 定时消息监听任务

Files:

Modify: digital-avatar-app/backend/main.py
Modify: digital-avatar-app/backend/requirements.txt

Interfaces:

APScheduler 定时任务：每 10 秒轮询一次未读消息
调用 TakeoverService.check_takeover_enabled 判断是否接管
根据模式调用立即执行或延迟队列

Dependencies:

TakeoverService from Task 3
BoxIMClient from Task 2
APScheduler for scheduling
redis for delayed queue
Step 1: Add dependencies

# digital-avatar-app/backend/requirements.txt
# 新增：
redis>=5.0
apscheduler>=3.10

Step 2: Add scheduler setup in main.py

# digital-avatar-app/backend/main.py
# 在 import 部分添加：
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.interval import IntervalTrigger
import redis as redis_lib
import os

# 在 on_startup 函数中添加：
@app.on_event("startup")
def on_startup():
    init_db()
    seed()

    # 初始化 Redis（可选）
    redis_client = None
    redis_url = os.getenv("REDIS_URL", "")
    if redis_url:
        try:
            redis_client = redis_lib.from_url(redis_url)
            redis_client.ping()
        except Exception:
            logger.warning("Redis 连接失败，延迟接管将降级为立即接管")

    # 初始化 Box IM 客户端
    boxim_config = {
        "HUIHUI_IM_BASE_URL": os.getenv("HUIHUI_IM_BASE_URL", "http://192.168.1.200:60040"),
        "HUIHUI_APP_ID": os.getenv("HUIHUI_APP_ID", ""),
        "HUIHUI_ACCESS_ID": os.getenv("HUIHUI_ACCESS_ID", ""),
        "HUIHUI_ACCESS_SECRET": os.getenv("HUIHUI_ACCESS_SECRET", ""),
    }
    boxim_client = BoxIMClient(boxim_config)

    # 初始化接管服务
    from services.takeover_service import TakeoverService
    takeover_service = TakeoverService(SessionLocal(), boxim_client, redis_client)

    # 启动定时任务
    scheduler = BackgroundScheduler()
    scheduler.add_job(
        takeover_service.poll_and_process_messages,
        trigger=IntervalTrigger(seconds=10),
        id="takeover_message_poll",
    )
    scheduler.start()

Step 3: Add poll method to TakeoverService

# digital-avatar-app/backend/services/takeover_service.py
# 在 TakeoverService 类中新增：
    async def poll_and_process_messages(self):
        """定时轮询未读消息并处理"""
        # 这里简化处理，实际需要调用会会平台的消息推送接口
        # 伪代码：
        # messages = await self.fetch_unread_messages()
        # for msg in messages:
        #     await self.process_message(msg)
        pass

    async def process_message(self, message: dict):
        """处理单条消息"""
        owner_id = message.get("owner_huihui_id", "")
        from_id = message.get("from_accid", "")

        auth = self.check_takeover_enabled(owner_id, from_id)
        if not auth:
            return

        if auth.takeover_mode == "immediate":
            await self.execute_takeover(auth, message)
        else:
            self.enqueue_delayed_message(auth, message)

Step 4: Commit

git add digital-avatar-app/backend/main.py digital-avatar-app/backend/requirements.txt digital-avatar-app/backend/services/takeover_service.py
git commit -m "feat: add scheduled message polling for takeover"

Task 6: 前端授权管理 UI 增强

Files:

Modify: digital-avatar-app/src/views/AuthorizationManage.vue
Modify: digital-avatar-app/src/api/index.ts

Interfaces:

授权卡片新增接管开关和配置
开关开启时显示模式选择和延迟秒数输入
调用 PUT /api/avatar/{id}/authorizations/takeover 保存配置

Dependencies:

Existing authorization API functions
Takeover config API from Task 4
Step 1: Add API function

// digital-avatar-app/src/api/index.ts
// 新增：
export const updateTakeoverConfig = (avatarId: string, data: {
  authorizationId: string
  takeoverEnabled: boolean
  takeoverMode?: string
  takeoverDelaySeconds?: number
}) => {
  return api.put(`/api/avatar/${avatarId}/authorizations/takeover`, data)
}

Step 2: Update AuthorizationManage.vue

<!-- digital-avatar-app/src/views/AuthorizationManage.vue -->
<!-- 在授权卡片的权限标签后添加接管配置 -->
<div v-if="auth.permissions?.includes('takeover') || auth.takeoverEnabled" class="takeover-config">
  <div class="takeover-header">
    <span class="takeover-label">分身接管聊天</span>
    <button class="toggle-btn" :class="{ active: auth.takeoverEnabled }" @click="toggleTakeover(auth)">
      <span class="toggle-dot"></span>
    </button>
  </div>

  <div v-if="auth.takeoverEnabled" class="takeover-options">
    <div class="option-row">
      <label>接管模式</label>
      <select v-model="auth.takeoverMode" @change="saveTakeoverConfig(auth)">
        <option value="immediate">立即接管</option>
        <option value="delayed">延迟接管</option>
      </select>
    </div>

    <div v-if="auth.takeoverMode === 'delayed'" class="option-row">
      <label>延迟秒数</label>
      <input type="number" v-model.number="auth.takeoverDelaySeconds"
             min="5" max="300" @change="saveTakeoverConfig(auth)" />
    </div>
  </div>
</div>

Step 3: Add methods to AuthorizationManage.vue

<script setup lang="ts">
// 新增 import
import { updateTakeoverConfig } from '@/api'

// 新增方法
const toggleTakeover = async (auth: any) => {
  auth.takeoverEnabled = !auth.takeoverEnabled
  await saveTakeoverConfig(auth)
}

const saveTakeoverConfig = async (auth: any) => {
  try {
    await updateTakeoverConfig(currentAvatarId.value, {
      authorizationId: auth.id,
      takeoverEnabled: auth.takeoverEnabled,
      takeoverMode: auth.takeoverMode || 'immediate',
      takeoverDelaySeconds: auth.takeoverDelaySeconds || 30,
    })
  } catch (e) {
    console.error('保存接管配置失败', e)
  }
}
</script>

Step 4: Add styles

<style scoped>
.takeover-config {
  margin-top: 12px;
  padding: 10px;
  background: rgba(88, 166, 255, 0.05);
  border-radius: 8px;
}
.takeover-header {
  display: flex;
  justify-content: space-between;
  align-items: center;
  margin-bottom: 8px;
}
.takeover-label {
  font-size: 13px;
  font-weight: 600;
}
.takeover-options {
  display: flex;
  flex-direction: column;
  gap: 8px;
}
.option-row {
  display: flex;
  align-items: center;
  gap: 8px;
  font-size: 12px;
}
.option-row label {
  width: 70px;
  flex-shrink: 0;
  color: var(--color-text-muted);
}
.option-row select, .option-row input {
  flex: 1;
  padding: 4px 8px;
  border: 1px solid var(--color-border);
  border-radius: 4px;
  background: var(--color-bg);
  color: var(--color-text);
}
</style>

Step 5: Commit

git add digital-avatar-app/src/views/AuthorizationManage.vue digital-avatar-app/src/api/index.ts
git commit -m "feat: add takeover config UI to authorization management"

Task 7: 集成测试和部署

Files:

Modify: docker-compose.yml (digital-avatar-app)
Create: digital-avatar-app/.env.example

Interfaces:

添加 Redis 容器（可选，用于延迟队列）
添加环境变量配置
Step 1: Add Redis to docker-compose

# digital-avatar-app/docker-compose.yml
services:
  avatar-backend:
    # ... 现有配置 ...
    environment:
      - REDIS_URL=redis://avatar-redis:6379
      - HUIHUI_IM_BASE_URL=http://192.168.1.200:60040
      - HUIHUI_APP_ID=${HUIHUI_APP_ID}
      - HUIHUI_ACCESS_ID=${HUIHUI_ACCESS_ID}
      - HUIHUI_ACCESS_SECRET=${HUIHUI_ACCESS_SECRET}
    depends_on:
      - avatar-redis

  avatar-redis:
    image: redis:7-alpine
    container_name: avatar-redis
    restart: unless-stopped
    volumes:
      - redis_data:/data
    networks:
      - avatar-net

Step 2: Create .env.example

# digital-avatar-app/.env.example
HUIHUI_IM_BASE_URL=http://192.168.1.200:60040
HUIHUI_APP_ID=your_app_id
HUIHUI_ACCESS_ID=your_access_id
HUIHUI_ACCESS_SECRET=your_access_secret
REDIS_URL=redis://avatar-redis:6379

Step 3: Manual testing checklist

## 测试清单

### 功能测试
- [ ] 授权管理页面显示接管开关
- [ ] 开关开启后显示模式选择
- [ ] 延迟模式显示延迟秒数输入
- [ ] 保存配置后刷新页面配置保留
- [ ] 立即接管：模拟消息到达后分身立即回复
- [ ] 延迟接管：N 秒内主人回复则不分身接管
- [ ] 延迟接管：超时后分身自动回复

### 边界测试
- [ ] 主人有多个分身时使用第一个 active 分身
- [ ] 知识库为空时使用 Qwen 兜底
- [ ] IM 接口异常时记录日志不崩溃
- [ ] Redis 不可用时降级为立即接管

### 性能测试
- [ ] 轮询间隔 10 秒时 CPU 占用正常
- [ ] 延迟队列消息过期自动清理

Step 4: Commit

git add digital-avatar-app/docker-compose.yml digital-avatar-app/.env.example
git commit -m "feat: add Redis for takeover delayed queue and env config"

执行顺序

Task 1 → Task 2 → Task 3 → Task 4 → Task 5 → Task 6 → Task 7
每个 Task 完成后运行测试验证
Task 5 和 Task 6 可以并行（后端定时任务和前端 UI 无依赖）

风险与缓解

风险	缓解措施
会会平台 IM 接口文档不完整	先做接口探索，用 Postman 手动验证
盒子 IM 替换云信的 API 差异	保留云信客户端作为 fallback
Redis 延迟队列复杂性	第一期可简化为内存队列，Redis 作为优化
分身聊天接口响应慢	设置超时，失败时不阻塞轮询

31 KiB Raw Blame History

数字分身接管聊天功能 — 实现计划

全局约束

文件清单

新建文件

修改文件

任务分解

Task 1: 扩展 Authorization 数据模型

Task 2: 盒子 IM 客户端

Task 3: 接管服务核心逻辑

Task 4: 接管配置 API

Task 5: 定时消息监听任务

Task 6: 前端授权管理 UI 增强

Task 7: 集成测试和部署

执行顺序

风险与缓解

31 KiB

Raw Blame History