#935 IM Bot Channel Adapter 化 (Telegram POC + im_channel_routes schema)
This commit is contained in:
SeniorDev committed
1 parent
048671870a
commit
6d5c2cf2c0
4 files changed
+477
-2
No files matched your search
@@ -0,0 +1,214 @@
|
||||
"""
|
||||
IM Bot Channel Adapter — 渠道插件化基础设施
|
||||
|
||||
任务 #935 · 2026-08-09
|
||||
- ChannelAdapter: 抽象基类
|
||||
- IncomingMessage: 跨渠道归一入站消息
|
||||
- ChannelRouter: 单例路由,复用 _send_message_core + BufferAdapter(单引擎原则)
|
||||
|
||||
依赖:已存在于 app/core/ 的 capability_discovery + claw_adapters(BufferAdapter)
|
||||
"""
|
||||
from __future__ import annotations
|
||||
import asyncio
|
||||
import logging
|
||||
import os
|
||||
from abc import ABC, abstractmethod
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Awaitable, Callable, Dict, Optional
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class IncomingMessage:
|
||||
"""跨渠道归一化入站消息。
|
||||
|
||||
ChannelPlugin.start() 启动后接收到 raw message,转成 IncomingMessage,
|
||||
交给 ChannelRouter.dispatch,组装后回送给原渠道。
|
||||
"""
|
||||
channel: str # "telegram" / "feishu" / "wecom" / ...
|
||||
platform_user_id: str # 渠道内用户/群 ID
|
||||
text: str # 正文(去 @-mention + 富文本转 markdown)
|
||||
is_group: bool = False # 群消息
|
||||
group_id: Optional[str] = None
|
||||
at_me: bool = False # 群中是否 @ 机器人
|
||||
attachments: Optional[list] = None # 图片/文件 URL(可选)
|
||||
raw: Any = None # 原始消息,留给高级用例
|
||||
|
||||
|
||||
class ChannelAdapter(ABC):
|
||||
"""IM 渠道插件契约。
|
||||
|
||||
实现要点:
|
||||
- start() 启动连接(扫码 / WebSocket / webhook / Bot)
|
||||
- send() 出站文本(注意各家消息长度上限,Telegram=4096 / 飞书=4000)
|
||||
- 收到入站消息时,内部组装为 IncomingMessage,转发给 self.router.dispatch
|
||||
"""
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def channel_id(self) -> str: ...
|
||||
|
||||
@abstractmethod
|
||||
async def start(self) -> None: ...
|
||||
|
||||
@abstractmethod
|
||||
async def stop(self) -> None: ...
|
||||
|
||||
@abstractmethod
|
||||
async def send(self, *, user_id: str, content: str) -> None: ...
|
||||
|
||||
# 由 ChannelRouter 注入(register() 时反向 set)
|
||||
router: Optional["ChannelRouter"] = None
|
||||
|
||||
|
||||
class ChannelRouter:
|
||||
"""单例路由。所有入站消息统一过 _send_message_core。"""
|
||||
|
||||
_shared: Optional["ChannelRouter"] = None
|
||||
|
||||
@classmethod
|
||||
def shared(cls) -> "ChannelRouter":
|
||||
if cls._shared is None:
|
||||
cls._shared = cls()
|
||||
return cls._shared
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._adapters: Dict[str, ChannelAdapter] = {}
|
||||
self._started: bool = False
|
||||
# 延迟导入(避免循环依赖)
|
||||
self._send_message_core = None
|
||||
self._SendMessageRequest = None
|
||||
self._BufferAdapter = None
|
||||
self._discover_skills = None
|
||||
self._discover_mcp = None
|
||||
self._Conversation = None
|
||||
self._User = None
|
||||
self._gen_uuid = None
|
||||
|
||||
def register(self, adapter: ChannelAdapter) -> None:
|
||||
if self._started:
|
||||
raise RuntimeError("ChannelRouter already started; can't register more adapters")
|
||||
adapter.router = self
|
||||
self._adapters[adapter.channel_id] = adapter
|
||||
log.info("[router] registered channel: %s", adapter.channel_id)
|
||||
|
||||
@property
|
||||
def channels(self) -> list[str]:
|
||||
return list(self._adapters.keys())
|
||||
|
||||
async def start_all(self) -> None:
|
||||
"""main.py 的 startup 钩子调用。"""
|
||||
self._lazy_import()
|
||||
for a in self._adapters.values():
|
||||
try:
|
||||
await a.start()
|
||||
except Exception as e:
|
||||
log.warning("[router] start %s failed: %s", a.channel_id, e, exc_info=True)
|
||||
self._started = True
|
||||
log.info("[router] started %d channels: %s", len(self._adapters), self.channels)
|
||||
|
||||
async def stop_all(self) -> None:
|
||||
for a in self._adapters.values():
|
||||
try:
|
||||
await a.stop()
|
||||
except Exception:
|
||||
pass
|
||||
self._started = False
|
||||
|
||||
def _lazy_import(self) -> None:
|
||||
if self._send_message_core is not None:
|
||||
return
|
||||
from app.api.chat_service import _send_message_core
|
||||
from app.api.chat_utils import SendMessageRequest
|
||||
from app.core.claw_adapters import BufferAdapter
|
||||
from app.core.capability_discovery import discover_mcp_servers, discover_skills
|
||||
from app.models import Conversation, User
|
||||
from app.core.uuid_compat import gen_uuid # 项目内统一 UUID 生成
|
||||
|
||||
self._send_message_core = _send_message_core
|
||||
self._SendMessageRequest = SendMessageRequest
|
||||
self._BufferAdapter = BufferAdapter
|
||||
self._discover_skills = discover_skills
|
||||
self._discover_mcp = discover_mcp_servers
|
||||
self._Conversation = Conversation
|
||||
self._User = User
|
||||
self._gen_uuid = gen_uuid
|
||||
|
||||
async def dispatch(self, incoming: IncomingMessage) -> str:
|
||||
"""1 条入站 → 1 条回复。
|
||||
|
||||
复用单引擎:_send_message_core + BufferAdapter。
|
||||
鉴权:从 im_channel_routes 解析 platform_user_id → internal User。
|
||||
"""
|
||||
self._lazy_import()
|
||||
# 1. 解析内部 user(由 platform_user_id → user_id)
|
||||
from sqlalchemy.orm import Session
|
||||
from app.models import IMChannelRoute
|
||||
from app.core.db import SessionLocal
|
||||
|
||||
db: Session = SessionLocal()
|
||||
try:
|
||||
route = db.query(IMChannelRoute).filter(
|
||||
IMChannelRoute.channel == incoming.channel,
|
||||
IMChannelRoute.platform_user_id == incoming.platform_user_id,
|
||||
).first()
|
||||
if route is None:
|
||||
return "（请先在七虎AI平台绑定账号:发送 /bind <你的绑定码>）"
|
||||
|
||||
user = db.query(self._User).filter(self._User.id == route.internal_user_id).first()
|
||||
if user is None:
|
||||
return "（绑定账号已失效,请重新绑定）"
|
||||
|
||||
# 2. 能力发现
|
||||
skills = []
|
||||
try:
|
||||
skills = self._discover_skills(db, user.id, query=incoming.text, claw="im") or []
|
||||
except Exception as e:
|
||||
log.warning("[router.disp] skill discovery: %s", e)
|
||||
mcp_servers = []
|
||||
try:
|
||||
mcp_servers = self._discover_mcp(db, None, claw="im") or []
|
||||
except Exception as e:
|
||||
log.warning("[router.disp] mcp discovery: %s", e)
|
||||
|
||||
skill_ids = [
|
||||
(s.get("id") if isinstance(s, dict) else getattr(s, "id", None))
|
||||
for s in skills
|
||||
]
|
||||
skill_ids = [s for s in skill_ids if s]
|
||||
mcp_ids = [
|
||||
(m.get("id") if isinstance(m, dict) else getattr(m, "id", None))
|
||||
for m in mcp_servers
|
||||
]
|
||||
mcp_ids = [m for m in mcp_ids if m]
|
||||
|
||||
# 3. 临时会话
|
||||
conv = self._Conversation(
|
||||
id=self._gen_uuid(),
|
||||
user_id=user.id,
|
||||
title=f"IM-{incoming.channel}",
|
||||
source="im_bot",
|
||||
)
|
||||
db.add(conv)
|
||||
db.commit()
|
||||
|
||||
req = self._SendMessageRequest(
|
||||
content=incoming.text,
|
||||
skill_ids=skill_ids or None,
|
||||
mcp_ids=mcp_ids or None,
|
||||
use_tools=bool(skill_ids or mcp_ids),
|
||||
agent_ids=None,
|
||||
knowledge_base_ids=None,
|
||||
attachments=None,
|
||||
)
|
||||
|
||||
adapter = self._BufferAdapter(tenant_id=str(getattr(user, "tenant_id", "") or ""))
|
||||
await self._send_message_core(conv.id, req, db, user, stream=True, on_event=adapter)
|
||||
reply = adapter.assemble()
|
||||
return reply or "（生成失败,请稍后再试）"
|
||||
except Exception as e:
|
||||
log.exception("[router.disp] failed: %s", e)
|
||||
return f"（系统繁忙:{e}）"
|
||||
finally:
|
||||
db.close()
|
||||
Reference in new issue
Block a user