zlei6/ragflow
1
0
Fork 0
mirror of https://github.com/infiniflow/ragflow.git synced 2026-08-03 06:17:29 +08:00
Code Issues Packages Projects Releases Wiki Activity

Feat: chat channels — connect assistants to external messaging bots (#15850)

Browse Source 
### What problem does this PR solve?

#15844

Adds a **Chat channels** capability so a RAGFlow assistant (Dialog) can
be exposed as a bot on external messaging platforms (Feishu/Lark,
Discord, Telegram, Slack, WeCom, LINE, etc.). An admin configures a bot
in the UI, connects it to an assistant, and inbound messages are
answered from that assistant's knowledge base — replies are delivered
back on the channel.

**Feishu/Lark is implemented and tested end-to-end.** Discord, Telegram,
LINE, and WeCom are scaffolded against the same interface; the remaining
listed channels are tracked as follow-ups.

### Design

**Backend**
- New `chat_channel` table (`tenant_id`, `name`, `channel`, `config`
JSON holding `{credential: {...}}`, `dialog_id`, `status`) +
`ChatChannelService` and RESTful CRUD under `/api/v1/chat_channels`.
- Channel framework under `api/channels/`: a `core` registry +
per-channel packages that self-register a builder and implement a common
`Channel` interface (`start`/`stop`/`send` + inbound normalization) over
`IncomingMessage`/`OutgoingMessage`.
- Embedded **reconcile loop** in `ragflow_server`
(`api/channels/bootstrap.py`): loads enabled bots, and
starts/stops/restarts them as rows change (no server restart needed).
Inbound messages run the connected dialog via the non-streaming
completion path, keeping per-end-user conversation history.
- Missing optional channel SDKs degrade gracefully (channel skipped with
a warning; others unaffected). Channel-level errors are logged, not
crashed.
- Feishu's WebSocket client runs in a dedicated thread with its own
event loop to avoid cross-loop/contextvars conflicts with the channel
runtime.

**Frontend**
- **Settings → Chat channels** panel: available-channels grid +
configured-bots list with add/edit/delete and a **Connect assistant**
popup that binds a bot to a dialog.
- Brand icons via simple-icons / reused shared data-source assets, with
colored fallbacks for brands not available.
- Route, sidebar entry, i18n (en/zh), and a top-nav segment-boundary fix
so the settings page no longer highlights the Chat tab.

### Type of change

- [x] New Feature (non-breaking change which adds functionality)

### Notes
- DB: new `chat_channel` table is auto-created; `chat_channel.dialog_id`
is also covered by a `migrate_db` `alter_db_add_column` for existing
installs.
- Channel SDKs (`lark-oapi`, `discord.py`, `python-telegram-bot`,
`line-bot-sdk`, `wechatpy`, `aiohttp`) added to dependencies.
- Screenshots / per-channel credential docs to follow.

<img width="1338" height="1290" alt="Image"
src="https://github.com/user-attachments/assets/042cb2f9-0dad-4e6a-bcf7-43ced4bbd704"
/>

<img width="1344" height="738" alt="Image"
src="https://github.com/user-attachments/assets/373cd08e-ec40-4c67-9c51-4d948b1ba617"
/>

<img width="672" height="887" alt="Image"
src="https://github.com/user-attachments/assets/5a34953f-a9a3-4c1e-869e-5eff0dc64c84"
/>

---------
This commit is contained in:
Kevin Hu
2026-06-12 18:21:30 +08:00
committed by GitHub
parent 5a7d7771a3
commit b5a426e6e0
68 changed files with 8232 additions and 5138 deletions
Download Patch File Download Diff File Expand all files Collapse all files

118
api/apps/restful_apis/chat_channel_api.py Normal file
View File

@@ -0,0 +1,118 @@
#
# Copyright 2024 The InfiniFlow Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import logging
from api.apps import current_user, login_required
from api.db.services.chat_channel_service import ChatChannelService
from api.db.services.dialog_service import DialogService
from api.utils.api_utils import get_data_error_result, get_json_result, get_request_json, validate_request
from common.constants import RetCode
from common.misc_utils import get_uuid
LOGGER = logging.getLogger(__name__)
def _chat_channel_auth_error(channel_id: str, user_id: str):
"""Return the chat channel authorization failure response and log the denial."""
LOGGER.warning("chat channel access denied: channel_id=%s user_id=%s", channel_id, user_id)
return get_json_result(data=False, message="No authorization.", code=RetCode.AUTHENTICATION_ERROR)
@manager.route("/chat_channels", methods=["POST"]) # noqa: F821
@login_required
@validate_request("name", "channel", "config")
async def create_chat_channel():
"""Create a chat channel bot owned by the current tenant."""
req = await get_request_json()
channel = {
"id": get_uuid(),
"tenant_id": current_user.id,
"name": req["name"],
"channel": req["channel"],
"config": req["config"],
"dialog_id": req.get("dialog_id") or None,
"status": "1",
}
ChatChannelService.insert(**channel)
e, conn = ChatChannelService.get_by_id(channel["id"])
if not e:
return get_data_error_result(message="Failed to create chat channel!")
return get_json_result(data=conn.to_dict())
@manager.route("/chat_channels", methods=["GET"]) # noqa: F821
@login_required
def list_chat_channel():
"""List chat channel bots owned by the current tenant."""
return get_json_result(data=ChatChannelService.list(current_user.id))
@manager.route("/chat_channels/<channel_id>", methods=["GET"]) # noqa: F821
@login_required
def get_chat_channel(channel_id):
"""Return a chat channel bot's details when the current user can access it."""
if not ChatChannelService.accessible(channel_id, current_user.id):
return _chat_channel_auth_error(channel_id, current_user.id)
e, conn = ChatChannelService.get_by_id(channel_id)
if not e:
return get_data_error_result(message="Can't find this chat channel!")
return get_json_result(data=conn.to_dict())
@manager.route("/chat_channels/<channel_id>", methods=["PATCH"]) # noqa: F821
@login_required
async def update_chat_channel(channel_id):
"""Update an accessible chat channel bot's name/config/status."""
if not ChatChannelService.accessible(channel_id, current_user.id):
return _chat_channel_auth_error(channel_id, current_user.id)
e, conn = ChatChannelService.get_by_id(channel_id)
if not e:
return get_data_error_result(message="Can't find this chat channel!")
req = await get_request_json()
if isinstance(req, dict) and isinstance(req.get("data"), dict):
req = req["data"]
# Validate the connected dialog (if provided) belongs to the channel's tenant.
if req.get("dialog_id"):
e, dia = DialogService.get_by_id(req["dialog_id"])
if not e:
return get_data_error_result(message="Can't find this chat assistant!")
if dia.tenant_id != conn.tenant_id:
return _chat_channel_auth_error(channel_id, current_user.id)
update_fields = {fld: req[fld] for fld in ["name", "config", "dialog_id", "status"] if fld in req}
if update_fields:
ChatChannelService.update_by_id(channel_id, update_fields)
e, conn = ChatChannelService.get_by_id(channel_id)
if not e:
return get_data_error_result(message="Can't find this chat channel!")
return get_json_result(data=conn.to_dict())
@manager.route("/chat_channels/<channel_id>", methods=["DELETE"]) # noqa: F821
@login_required
def rm_chat_channel(channel_id):
"""Delete an accessible chat channel bot."""
if not ChatChannelService.accessible(channel_id, current_user.id):
return _chat_channel_auth_error(channel_id, current_user.id)
ChatChannelService.delete_by_id(channel_id)
return get_json_result(data=True)

0
api/channels/__init__.py Normal file
View File

267
api/channels/bootstrap.py Normal file
View File

@@ -0,0 +1,267 @@
#
# Copyright 2024 The InfiniFlow Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
"""Chat channel runtime, embedded in the RAGFlow API server.
Continuously reconciles the running channel bots against the ``chat_channel``
table: newly added bots are started, deleted ones are stopped, and edited ones
(credential/type change) are restarted — without restarting the server. Inbound
messages are answered with a RAG completion routed through the conversation
wired to that bot. Replaces the standalone ``server.py`` entrypoint.
"""
from __future__ import annotations
import asyncio
import hashlib
import importlib
import json
import logging
import threading
LOGGER = logging.getLogger(__name__)
# Channel packages bundled under api/channels that self-register on import.
_BUNDLED_CHANNELS = ("feishu", "discord", "telegram", "line", "wecom")
# How often (seconds) to reconcile running channels against the database.
_RECONCILE_INTERVAL_SECS = 10
def _register_channels() -> None:
"""Import each bundled channel package so it self-registers a builder.
Each channel is imported independently: a missing optional dependency only
disables that one channel instead of taking down the whole channel server.
"""
for name in _BUNDLED_CHANNELS:
try:
importlib.import_module(f"api.channels.{name}")
except Exception as ex:
LOGGER.warning("chat channel '%s' unavailable: %s", name, ex)
def _fingerprint(channel: str, credential: dict) -> str:
"""Stable hash of the parts that require a channel restart when changed."""
payload = json.dumps(
{"channel": channel, "credential": credential},
sort_keys=True,
default=str,
)
return hashlib.md5(payload.encode("utf-8")).hexdigest()
def _desired_channels() -> dict:
"""Return {chat_channel.id: (channel_type, credential, fingerprint)} for enabled bots."""
from api.db.services.chat_channel_service import ChatChannelService
desired: dict = {}
for row in ChatChannelService.list_active():
credential = (row.config or {}).get("credential", {}) or {}
desired[row.id] = (row.channel, credential, _fingerprint(row.channel, credential))
return desired
def _build_one(account_id: str, channel: str, credential: dict):
"""Build a single Channel instance, or None if the type has no builder."""
from api.channels.core.registry import build_channels
# account_id == chat_channel.id.
instances = build_channels(
{"channels": {channel: {"accounts": {account_id: credential}}}}
)
return instances[0] if instances else None
def _make_chat_handler(ch):
"""Build the inbound-message handler bound to a single channel.
Mirrors the non-streaming path of ``session_completion``: the message is
appended to a per-end-user conversation under the dialog connected to the
bot, a RAG completion is run against that dialog, and the answer is sent
back. The connected dialog is resolved per message, so connection changes
take effect immediately without restarting the channel. Channels with no
connected dialog ignore inbound messages.
"""
from api.channels.core.base import IncomingMessage, OutgoingMessage
from api.db.services.chat_channel_service import ChatChannelService
from api.db.services.conversation_service import ConversationService, structure_answer
from api.db.services.dialog_service import DialogService, async_chat
from common.misc_utils import get_uuid
async def handle(msg: IncomingMessage) -> None:
if not (msg.text or "").strip():
return
# account_id == chat_channel.id; re-read so a re-connected dialog applies live.
e, cc = ChatChannelService.get_by_id(ch.account_id)
if not e or not cc.dialog_id:
LOGGER.info(
"[%s:%s] no dialog connected; ignoring message",
ch.channel_id,
ch.account_id,
)
return
e, dia = DialogService.get_by_id(cc.dialog_id)
if not e:
LOGGER.warning("[%s:%s] connected dialog not found: %s", ch.channel_id, ch.account_id, cc.dialog_id)
return
conv = ConversationService.get_or_create_for_channel(cc.dialog_id, ch.account_id, msg.chat_id)
if conv is None:
LOGGER.warning("[%s:%s] failed to get conversation for chat %s", ch.channel_id, ch.account_id, msg.chat_id)
return
message_id = get_uuid()
if not conv.message:
conv.message = []
conv.message.append({"role": "user", "content": msg.text, "id": message_id})
if not conv.reference:
conv.reference = []
conv.reference = [r for r in conv.reference if r]
conv.reference.append({"chunks": [], "doc_aggs": []})
history = []
for m in conv.message:
if m["role"] == "system":
continue
if m["role"] == "assistant" and not history:
continue
history.append(m)
answer_text = ""
try:
async for ans in async_chat(dia, history, False, quote=False):
structure_answer(conv, ans, message_id, conv.id)
answer_text = (ans or {}).get("answer", "") or ""
ConversationService.update_by_id(conv.id, conv.to_dict())
break
except Exception as ex:
LOGGER.exception("[%s:%s] completion failed: %s", ch.channel_id, ch.account_id, ex)
answer_text = f"**ERROR**: {ex}"
if answer_text:
await ch.send(
OutgoingMessage(
chat_id=msg.chat_id,
text=answer_text,
reply_to_message_id=msg.message_id or None,
)
)
return handle
async def _stop_channel(running: dict, account_id: str) -> None:
entry = running.pop(account_id, None)
if not entry:
return
ch = entry["ch"]
try:
await ch.stop()
LOGGER.info("stopped chat channel %s:%s", ch.channel_id, account_id)
except Exception as ex:
LOGGER.error("failed to stop chat channel %s: %s", account_id, ex)
async def _start_channel(running: dict, account_id: str, channel: str, credential: dict, fp: str) -> bool:
"""Build, wire and start one channel. Returns True on success.
Any failure (e.g. invalid credentials) is contained here so a single bad bot
config never aborts the reconcile pass for the other channels.
"""
try:
ch = _build_one(account_id, channel, credential)
except Exception as ex:
LOGGER.error(
"failed to build chat channel %s (%s); check its credentials: %s",
account_id,
channel,
ex,
)
return False
if ch is None:
return False
ch.set_message_handler(_make_chat_handler(ch))
try:
await ch.start()
except Exception as ex:
LOGGER.error("failed to start chat channel %s (%s): %s", account_id, channel, ex)
return False
running[account_id] = {"ch": ch, "fp": fp}
LOGGER.info("started chat channel %s:%s", ch.channel_id, account_id)
return True
async def _reconcile(running: dict, failed: dict) -> None:
"""Diff desired (DB) vs running channels and apply start/stop/restart.
``failed`` remembers configs that could not be started so they are not
retried (and re-logged) every tick until their credentials change.
"""
desired = await asyncio.to_thread(_desired_channels)
# Stop channels that were removed or whose credentials/type changed.
for account_id in list(running.keys()):
changed = account_id in desired and desired[account_id][2] != running[account_id]["fp"]
if account_id not in desired or changed:
await _stop_channel(running, account_id)
# Drop remembered failures that are gone or whose config changed, so an
# edited (hopefully fixed) bot is retried.
for account_id in list(failed.keys()):
if account_id not in desired or desired[account_id][2] != failed[account_id]:
failed.pop(account_id, None)
# Start channels that are new (skip ones already known to fail with this config).
for account_id, (channel, credential, fp) in desired.items():
if account_id in running or failed.get(account_id) == fp:
continue
if not await _start_channel(running, account_id, channel, credential, fp):
failed[account_id] = fp
async def run_channels(stop_event: threading.Event) -> None:
"""Reconcile and run channels until ``stop_event`` is set."""
_register_channels()
running: dict = {}
failed: dict = {}
try:
while not stop_event.is_set():
try:
await _reconcile(running, failed)
except Exception as ex:
LOGGER.error("chat channel reconcile failed: %s", ex)
for _ in range(_RECONCILE_INTERVAL_SECS):
if stop_event.is_set():
break
await asyncio.sleep(1)
finally:
LOGGER.info("Stopping chat channels...")
for account_id in list(running.keys()):
await _stop_channel(running, account_id)
def start_channel_server(stop_event: threading.Event) -> None:
"""Thread entrypoint: run the channel event loop, isolating any failure."""
try:
asyncio.run(run_channels(stop_event))
except Exception as ex:
LOGGER.exception("Chat channel server crashed: %s", ex)

0
api/channels/core/__init__.py Normal file
View File

60
api/channels/core/base.py Normal file
View File

@@ -0,0 +1,60 @@
from __future__ import annotations
import logging
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any, Awaitable, Callable, ClassVar, Optional
LOGGER = logging.getLogger(__name__)
@dataclass
class IncomingMessage:
channel: str
account_id: str
chat_id: str
chat_type: str
message_id: str
sender_id: str
text: str
raw: Any = None
@dataclass
class OutgoingMessage:
chat_id: str
text: str
reply_to_message_id: Optional[str] = None
MessageHandler = Callable[[IncomingMessage], Awaitable[None]]
class Channel(ABC):
"""One configured bot identity on one messaging platform."""
channel_id: ClassVar[str]
account_id: str
def __init__(self) -> None:
self._handler: Optional[MessageHandler] = None
def set_message_handler(self, handler: MessageHandler) -> None:
self._handler = handler
async def _dispatch(self, message: IncomingMessage) -> None:
if self._handler is None:
return
try:
await self._handler(message)
except Exception: # framework boundary — keep one bad msg from killing the channel
LOGGER.error("[%s:%s] handler error", self.channel_id, self.account_id, exc_info=True)
@abstractmethod
async def start(self) -> None: ...
@abstractmethod
async def stop(self) -> None: ...
@abstractmethod
async def send(self, message: OutgoingMessage) -> None: ...

47
api/channels/core/registry.py Normal file
View File

@@ -0,0 +1,47 @@
from __future__ import annotations
import logging
from typing import Callable, Dict, List
from .base import Channel
LOGGER = logging.getLogger(__name__)
ChannelBuilder = Callable[[str, dict], Channel]
_BUILDERS: Dict[str, ChannelBuilder] = {}
def register_channel(name: str, builder: ChannelBuilder) -> None:
_BUILDERS[name] = builder
def registered_channel_ids() -> List[str]:
return sorted(_BUILDERS)
def build_channels(config: dict) -> List[Channel]:
"""Walk config.channels.<name>.accounts.<id> and construct one Channel per account."""
instances: List[Channel] = []
channels_cfg = config.get("channels") or {}
for name, raw in channels_cfg.items():
if not isinstance(raw, dict) or raw.get("enabled") is False:
continue
builder = _BUILDERS.get(name)
if builder is None:
LOGGER.warning("no builder registered for channel '%s'; skipping", name)
continue
accounts = raw.get("accounts") or {}
if not accounts:
# Allow a flat single-account config without an `accounts:` block.
accounts = {"default": {k: v for k, v in raw.items() if k != "accounts"}}
shared = {k: v for k, v in raw.items() if k not in ("accounts", "default_account")}
for account_id, account_cfg in accounts.items():
if not isinstance(account_cfg, dict):
continue
if account_cfg.get("enabled") is False:
continue
merged = {**shared, **account_cfg}
instances.append(builder(str(account_id), merged))
return instances

1
api/channels/discord/__init__.py Normal file
View File

@@ -0,0 +1 @@
from . import channel # noqa: F401

135
api/channels/discord/channel.py Normal file
View File

@@ -0,0 +1,135 @@
from __future__ import annotations
import asyncio
import logging
from dataclasses import dataclass
from typing import Optional
import discord
from ..core.base import Channel, IncomingMessage, OutgoingMessage
from ..core.registry import register_channel
LOGGER = logging.getLogger(__name__)
@dataclass
class DiscordAccount:
account_id: str
token: str
def _chat_type(channel: discord.abc.Messageable) -> str:
if isinstance(channel, discord.DMChannel):
return "p2p"
if isinstance(channel, discord.Thread):
return "thread"
if isinstance(channel, (discord.TextChannel, discord.VoiceChannel, discord.StageChannel)):
return "group"
return type(channel).__name__
class DiscordChannel(Channel):
channel_id = "discord"
def __init__(self, account: DiscordAccount) -> None:
super().__init__()
self.account = account
self.account_id = account.account_id
intents = discord.Intents.default()
# Message Content is a privileged intent; must also be enabled in the
# Developer Portal under the application's Bot page.
intents.message_content = True
self._client = discord.Client(intents=intents)
self._run_task: Optional[asyncio.Task] = None
self._register_handlers()
def _register_handlers(self) -> None:
@self._client.event
async def on_ready() -> None:
try:
user = self._client.user
LOGGER.info(
"[discord:%s] connected as %s (id=%s)",
self.account_id,
user,
user.id if user else "unknown",
)
except Exception:
LOGGER.error("[discord:%s] on_ready error", self.account_id, exc_info=True)
@self._client.event
async def on_message(message: discord.Message) -> None:
try:
if message.author.bot:
return
me = self._client.user
if me is not None and message.author.id == me.id:
return
incoming = IncomingMessage(
channel=self.channel_id,
account_id=self.account_id,
chat_id=str(message.channel.id),
chat_type=_chat_type(message.channel),
message_id=str(message.id),
sender_id=str(message.author.id),
text=message.content or "",
raw=message,
)
await self._dispatch(incoming)
except Exception:
LOGGER.error("[discord:%s] inbound message handling error", self.account_id, exc_info=True)
async def start(self) -> None:
LOGGER.info("[discord:%s] starting gateway client", self.account_id)
self._run_task = asyncio.create_task(self._client.start(self.account.token))
async def stop(self) -> None:
if not self._client.is_closed():
await self._client.close()
if self._run_task and not self._run_task.done():
try:
await self._run_task
except (asyncio.CancelledError, Exception):
pass
async def send(self, message: OutgoingMessage) -> None:
try:
channel_id = int(message.chat_id)
except (TypeError, ValueError):
LOGGER.error("[discord:%s] invalid chat_id: %r", self.account_id, message.chat_id)
return
target = self._client.get_channel(channel_id)
if target is None:
try:
target = await self._client.fetch_channel(channel_id)
except discord.HTTPException as err:
LOGGER.error("[discord:%s] fetch_channel failed: %s", self.account_id, err)
return
reference = None
if message.reply_to_message_id:
try:
reference = discord.MessageReference(
message_id=int(message.reply_to_message_id),
channel_id=channel_id,
fail_if_not_exists=False,
)
except (TypeError, ValueError):
reference = None
try:
await target.send(message.text, reference=reference)
except discord.HTTPException as err:
LOGGER.error("[discord:%s] send failed: %s", self.account_id, err)
def _build(account_id: str, cfg: dict) -> Channel:
token = cfg.get("token")
if not token:
raise ValueError(f"discord account '{account_id}' is missing token")
return DiscordChannel(DiscordAccount(account_id=account_id, token=str(token)))
register_channel("discord", _build)

1
api/channels/feishu/__init__.py Normal file
View File

@@ -0,0 +1 @@
from . import channel # noqa: F401

210
api/channels/feishu/channel.py Normal file
View File

@@ -0,0 +1,210 @@
from __future__ import annotations
import asyncio
import json
import logging
import threading
from dataclasses import dataclass
from typing import Optional
import lark_oapi as lark
from lark_oapi.api.im.v1 import (
CreateMessageRequest,
CreateMessageRequestBody,
P2ImMessageReceiveV1,
ReplyMessageRequest,
ReplyMessageRequestBody,
)
from ..core.base import Channel, IncomingMessage, OutgoingMessage
from ..core.registry import register_channel
LOGGER = logging.getLogger(__name__)
@dataclass
class FeishuAccount:
account_id: str
app_id: str
app_secret: str
domain: str = "feishu" # "feishu" or "lark"
def _lark_domain(domain: str) -> str:
return lark.FEISHU_DOMAIN if domain != "lark" else lark.LARK_DOMAIN
class FeishuChannel(Channel):
channel_id = "feishu"
def __init__(self, account: FeishuAccount) -> None:
super().__init__()
self.account = account
self.account_id = account.account_id
self._loop: Optional[asyncio.AbstractEventLoop] = None
self._ws_client = None
self._ws_thread: Optional[threading.Thread] = None
self._rest = (
lark.Client.builder()
.app_id(account.app_id)
.app_secret(account.app_secret)
.domain(_lark_domain(account.domain))
.log_level(lark.LogLevel.DEBUG)
.build()
)
async def start(self) -> None:
# The channel loop is the cross-thread dispatch target for inbound events.
self._loop = asyncio.get_running_loop()
LOGGER.info("[feishu:%s] starting WebSocket client", self.account_id)
self._ws_thread = threading.Thread(
target=self._run_ws,
name=f"feishu-ws-{self.account_id}",
daemon=True,
)
self._ws_thread.start()
def _run_ws(self) -> None:
# Everything lark touches must be created and run on THIS thread with its
# own event loop. lark captures the running loop when the handler/client
# are built and when start() runs; building them on the channel daemon
# loop made lark schedule its WebSocket onto that loop, colliding with
# run_channels() ("Leaving task ... does not match" / "cannot enter
# context: already entered"). A dedicated isolated loop avoids that.
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
handler = (
lark.EventDispatcherHandler.builder("", "")
.register_p2_im_message_receive_v1(self._on_message_receive)
.build()
)
self._ws_client = lark.ws.Client(
self.account.app_id,
self.account.app_secret,
domain=_lark_domain(self.account.domain),
event_handler=handler,
log_level=lark.LogLevel.DEBUG,
)
# Blocks, running lark's own connect/reconnect loop on this thread.
self._ws_client.start()
except Exception:
LOGGER.error("[feishu:%s] WebSocket client crashed", self.account_id, exc_info=True)
finally:
try:
loop.close()
except Exception:
pass
async def stop(self) -> None:
# lark's ws client exposes no clean public stop; disconnect best-effort.
client = self._ws_client
if client is not None:
for attr in ("stop", "_disconnect", "disconnect"):
fn = getattr(client, attr, None)
if callable(fn):
try:
fn()
except Exception:
LOGGER.error("[feishu:%s] ws stop error", self.account_id, exc_info=True)
break
self._ws_client = None
self._ws_thread = None
async def send(self, message: OutgoingMessage) -> None:
content = json.dumps({"text": message.text}, ensure_ascii=False)
if message.reply_to_message_id:
req = (
ReplyMessageRequest.builder()
.message_id(message.reply_to_message_id)
.request_body(
ReplyMessageRequestBody.builder()
.content(content)
.msg_type("text")
.build()
)
.build()
)
resp = await asyncio.to_thread(self._rest.im.v1.message.reply, req)
else:
req = (
CreateMessageRequest.builder()
.receive_id_type("chat_id")
.request_body(
CreateMessageRequestBody.builder()
.receive_id(message.chat_id)
.content(content)
.msg_type("text")
.build()
)
.build()
)
resp = await asyncio.to_thread(self._rest.im.v1.message.create, req)
if not resp.success():
LOGGER.error(
"[feishu:%s] send failed: code=%s msg=%s",
self.account_id,
resp.code,
resp.msg,
)
def _on_message_receive(self, data: P2ImMessageReceiveV1) -> None:
# Runs on the lark-oapi WS thread; bounce into asyncio for downstream handlers.
try:
incoming = self._normalize(data)
if self._loop and not self._loop.is_closed():
future = asyncio.run_coroutine_threadsafe(self._dispatch(incoming), self._loop)
future.add_done_callback(self._log_dispatch_result)
except Exception:
LOGGER.error("[feishu:%s] inbound message handling error", self.account_id, exc_info=True)
def _log_dispatch_result(self, future) -> None:
try:
future.result()
except Exception:
LOGGER.error("[feishu:%s] dispatch error", self.account_id, exc_info=True)
def _normalize(self, data: P2ImMessageReceiveV1) -> IncomingMessage:
event = data.event
msg = event.message
sender = event.sender
text = ""
if msg.content:
try:
payload = json.loads(msg.content)
text = payload.get("text", "") if isinstance(payload, dict) else ""
except (json.JSONDecodeError, TypeError):
text = msg.content
sender_id = ""
if sender and getattr(sender, "sender_id", None):
sender_id = getattr(sender.sender_id, "open_id", "") or ""
return IncomingMessage(
channel=self.channel_id,
account_id=self.account_id,
chat_id=msg.chat_id or "",
chat_type=msg.chat_type or "",
message_id=msg.message_id or "",
sender_id=sender_id,
text=text,
raw=data,
)
def _build(account_id: str, cfg: dict) -> Channel:
app_id = cfg.get("app_id")
app_secret = cfg.get("app_secret")
if not app_id or not app_secret:
raise ValueError(
f"feishu account '{account_id}' is missing app_id or app_secret"
)
return FeishuChannel(
FeishuAccount(
account_id=account_id,
app_id=str(app_id),
app_secret=str(app_secret),
domain=str(cfg.get("domain", "feishu")),
)
)
register_channel("feishu", _build)

1
api/channels/line/__init__.py Normal file
View File

@@ -0,0 +1 @@
from . import channel # noqa: F401

230
api/channels/line/channel.py Normal file
View File

@@ -0,0 +1,230 @@
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import Dict, Optional, Tuple
from aiohttp import web
from linebot.v3 import WebhookParser
from linebot.v3.exceptions import InvalidSignatureError
from linebot.v3.messaging import (
AsyncApiClient,
AsyncMessagingApi,
Configuration,
PushMessageRequest,
ReplyMessageRequest,
TextMessage,
)
from linebot.v3.webhooks import (
GroupSource,
MessageEvent,
RoomSource,
TextMessageContent,
UserSource,
)
from ..core.base import Channel, IncomingMessage, OutgoingMessage
from ..core.registry import register_channel
LOGGER = logging.getLogger(__name__)
@dataclass
class LineAccount:
account_id: str
channel_secret: str
channel_access_token: str
webhook_host: str = "0.0.0.0"
webhook_port: int = 3001
class _SharedWebhookServer:
def __init__(self, host: str, port: int) -> None:
self.host = host
self.port = port
self.app = web.Application()
self.app.router.add_post("/line/{account_id}/webhook", self._handle_request)
self.runner: Optional[web.AppRunner] = None
self.site: Optional[web.TCPSite] = None
self.channels: Dict[str, "LineChannel"] = {}
async def start(self) -> None:
if self.runner is not None:
return
self.runner = web.AppRunner(self.app)
await self.runner.setup()
self.site = web.TCPSite(self.runner, self.host, self.port)
await self.site.start()
LOGGER.info(
"[line] webhook listening on http://%s:%s/line/<account_id>/webhook",
self.host,
self.port,
)
async def stop(self) -> None:
if self.site is not None:
await self.site.stop()
if self.runner is not None:
await self.runner.cleanup()
self.runner = None
self.site = None
async def _handle_request(self, request: web.Request) -> web.Response:
account_id = request.match_info.get("account_id", "")
try:
body = await request.text()
signature = request.headers.get("x-line-signature", "")
channel = self.channels.get(account_id)
if channel is None:
return web.Response(status=404, text="unknown account")
try:
events = channel.parser.parse(body, signature)
except InvalidSignatureError:
return web.Response(status=403, text="bad signature")
for event in events:
try:
await channel.handle_event(event)
except Exception:
LOGGER.error("[line:%s] event handling error", account_id, exc_info=True)
except Exception:
LOGGER.error("[line:%s] inbound request handling error", account_id, exc_info=True)
return web.Response(status=200, text="ok")
_servers: Dict[Tuple[str, int], _SharedWebhookServer] = {}
_active_per_server: Dict[Tuple[str, int], int] = {}
async def _acquire_server(host: str, port: int) -> _SharedWebhookServer:
key = (host, port)
server = _servers.get(key)
if server is None:
server = _SharedWebhookServer(host, port)
_servers[key] = server
await server.start()
_active_per_server[key] = _active_per_server.get(key, 0) + 1
return server
async def _release_server(host: str, port: int) -> None:
key = (host, port)
remaining = _active_per_server.get(key, 0) - 1
_active_per_server[key] = remaining
if remaining <= 0:
server = _servers.pop(key, None)
_active_per_server.pop(key, None)
if server is not None:
await server.stop()
def _chat_type_and_id(source) -> Tuple[str, str]:
if isinstance(source, GroupSource):
return ("group", source.group_id or "")
if isinstance(source, RoomSource):
return ("group", source.room_id or "")
if isinstance(source, UserSource):
return ("p2p", source.user_id or "")
return (type(source).__name__, getattr(source, "user_id", "") or "")
class LineChannel(Channel):
channel_id = "line"
def __init__(self, account: LineAccount) -> None:
super().__init__()
self.account = account
self.account_id = account.account_id
self.parser = WebhookParser(account.channel_secret)
self._config = Configuration(access_token=account.channel_access_token)
self._server: Optional[_SharedWebhookServer] = None
# LINE reply tokens are single-use and expire ~30s after the event.
self._reply_tokens: Dict[str, str] = {}
async def start(self) -> None:
self._server = await _acquire_server(self.account.webhook_host, self.account.webhook_port)
self._server.channels[self.account_id] = self
LOGGER.info(
"[line:%s] registered at path /line/%s/webhook",
self.account_id,
self.account_id,
)
async def stop(self) -> None:
if self._server is not None:
self._server.channels.pop(self.account_id, None)
await _release_server(self.account.webhook_host, self.account.webhook_port)
self._server = None
async def handle_event(self, event) -> None:
try:
if not isinstance(event, MessageEvent):
return
content = event.message
if not isinstance(content, TextMessageContent):
return
chat_type, chat_id = _chat_type_and_id(event.source)
sender_id = getattr(event.source, "user_id", "") or ""
if event.reply_token:
self._reply_tokens[content.id] = event.reply_token
incoming = IncomingMessage(
channel=self.channel_id,
account_id=self.account_id,
chat_id=chat_id,
chat_type=chat_type,
message_id=content.id,
sender_id=sender_id,
text=content.text or "",
raw=event,
)
await self._dispatch(incoming)
except Exception:
LOGGER.error("[line:%s] inbound message handling error", self.account_id, exc_info=True)
async def send(self, message: OutgoingMessage) -> None:
reply_token: Optional[str] = None
if message.reply_to_message_id:
reply_token = self._reply_tokens.pop(message.reply_to_message_id, None)
try:
async with AsyncApiClient(self._config) as api_client:
api = AsyncMessagingApi(api_client)
if reply_token:
await api.reply_message(
ReplyMessageRequest(
reply_token=reply_token,
messages=[TextMessage(text=message.text)],
)
)
else:
if not message.chat_id:
LOGGER.error("[line:%s] no chat_id for push send", self.account_id)
return
await api.push_message(
PushMessageRequest(
to=message.chat_id,
messages=[TextMessage(text=message.text)],
)
)
except Exception:
LOGGER.error("[line:%s] send failed", self.account_id, exc_info=True)
def _build(account_id: str, cfg: dict) -> Channel:
channel_secret = cfg.get("channel_secret")
channel_access_token = cfg.get("channel_access_token")
if not channel_secret or not channel_access_token:
raise ValueError(
f"line account '{account_id}' missing channel_secret or channel_access_token"
)
return LineChannel(
LineAccount(
account_id=account_id,
channel_secret=str(channel_secret),
channel_access_token=str(channel_access_token),
webhook_host=str(cfg.get("webhook_host", "0.0.0.0")),
webhook_port=int(cfg.get("webhook_port", 3001)),
)
)
register_channel("line", _build)

1
api/channels/telegram/__init__.py Normal file
View File

@@ -0,0 +1 @@
from . import channel # noqa: F401

118
api/channels/telegram/channel.py Normal file
View File

@@ -0,0 +1,118 @@
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import Optional
from telegram import ReplyParameters, Update
from telegram.ext import Application, ContextTypes, MessageHandler, filters
from ..core.base import Channel, IncomingMessage, OutgoingMessage
from ..core.registry import register_channel
LOGGER = logging.getLogger(__name__)
@dataclass
class TelegramAccount:
account_id: str
token: str
def _chat_type(chat) -> str:
t = getattr(chat, "type", "")
if t == "private":
return "p2p"
if t in ("group", "supergroup"):
return "group"
if t == "channel":
return "channel"
return str(t) or "unknown"
class TelegramChannel(Channel):
channel_id = "telegram"
def __init__(self, account: TelegramAccount) -> None:
super().__init__()
self.account = account
self.account_id = account.account_id
self._app: Optional[Application] = None
async def start(self) -> None:
self._app = Application.builder().token(self.account.token).build()
self._app.add_handler(MessageHandler(filters.ALL, self._on_update))
LOGGER.info("[telegram:%s] starting long-poll", self.account_id)
await self._app.initialize()
await self._app.start()
await self._app.updater.start_polling(drop_pending_updates=True)
async def stop(self) -> None:
if self._app is None:
return
try:
if self._app.updater and self._app.updater.running:
await self._app.updater.stop()
await self._app.stop()
await self._app.shutdown()
except Exception:
LOGGER.error("[telegram:%s] stop error", self.account_id, exc_info=True)
finally:
self._app = None
async def send(self, message: OutgoingMessage) -> None:
if self._app is None:
return
try:
chat_id = int(message.chat_id)
except (TypeError, ValueError):
LOGGER.error("[telegram:%s] invalid chat_id: %r", self.account_id, message.chat_id)
return
reply_parameters = None
if message.reply_to_message_id:
try:
reply_parameters = ReplyParameters(
message_id=int(message.reply_to_message_id),
allow_sending_without_reply=True,
)
except (TypeError, ValueError):
reply_parameters = None
try:
await self._app.bot.send_message(
chat_id=chat_id,
text=message.text,
reply_parameters=reply_parameters,
)
except Exception:
LOGGER.error("[telegram:%s] send failed", self.account_id, exc_info=True)
async def _on_update(self, update: Update, _ctx: ContextTypes.DEFAULT_TYPE) -> None:
try:
msg = update.effective_message
if msg is None or msg.from_user is None or msg.from_user.is_bot:
return
text = msg.text or msg.caption or ""
incoming = IncomingMessage(
channel=self.channel_id,
account_id=self.account_id,
chat_id=str(msg.chat.id),
chat_type=_chat_type(msg.chat),
message_id=str(msg.message_id),
sender_id=str(msg.from_user.id),
text=text,
raw=update,
)
await self._dispatch(incoming)
except Exception:
LOGGER.error("[telegram:%s] inbound message handling error", self.account_id, exc_info=True)
def _build(account_id: str, cfg: dict) -> Channel:
token = cfg.get("token")
if not token:
raise ValueError(f"telegram account '{account_id}' is missing token")
return TelegramChannel(TelegramAccount(account_id=account_id, token=str(token)))
register_channel("telegram", _build)

1
api/channels/wecom/__init__.py Normal file
View File

@@ -0,0 +1 @@
from . import channel # noqa: F401

290
api/channels/wecom/channel.py Normal file
View File

@@ -0,0 +1,290 @@
from __future__ import annotations
import asyncio
import logging
import time
from dataclasses import dataclass
from typing import Dict, Optional, Tuple
import aiohttp
from aiohttp import web
from wechatpy.enterprise import parse_message
from wechatpy.enterprise.crypto import WeChatCrypto
from wechatpy.exceptions import InvalidSignatureException
from ..core.base import Channel, IncomingMessage, OutgoingMessage
from ..core.registry import register_channel
LOGGER = logging.getLogger(__name__)
WECOM_API_BASE = "https://qyapi.weixin.qq.com/cgi-bin"
@dataclass
class WeComAccount:
account_id: str
corp_id: str
agent_id: int
secret: str
token: str
aes_key: str
webhook_host: str = "0.0.0.0"
webhook_port: int = 3002
class _SharedWebhookServer:
"""Single aiohttp server shared by all WeComChannel instances."""
def __init__(self, host: str, port: int) -> None:
self.host = host
self.port = port
self.app = web.Application()
self.app.router.add_get("/wecom/{account_id}/callback", self._handle_request)
self.app.router.add_post("/wecom/{account_id}/callback", self._handle_request)
self.runner: Optional[web.AppRunner] = None
self.site: Optional[web.TCPSite] = None
self.channels: Dict[str, "WeComChannel"] = {}
async def start(self) -> None:
if self.runner is not None:
return
self.runner = web.AppRunner(self.app)
await self.runner.setup()
self.site = web.TCPSite(self.runner, self.host, self.port)
await self.site.start()
LOGGER.info(
"[wecom] webhook listening on http://%s:%s/wecom/<account_id>/callback",
self.host,
self.port,
)
async def stop(self) -> None:
if self.site is not None:
await self.site.stop()
if self.runner is not None:
await self.runner.cleanup()
self.runner = None
self.site = None
async def _handle_request(self, request: web.Request) -> web.Response:
account_id = request.match_info.get("account_id", "")
try:
channel = self.channels.get(account_id)
if channel is None:
return web.Response(status=404, text="unknown account")
signature = request.query.get("msg_signature", "")
timestamp = request.query.get("timestamp", "")
nonce = request.query.get("nonce", "")
# GET = URL verification on first save in the WeCom admin console.
if request.method == "GET":
echo_str = request.query.get("echostr", "")
try:
decrypted = channel.crypto.check_signature(
signature, timestamp, nonce, echo_str
)
return web.Response(text=decrypted)
except InvalidSignatureException:
return web.Response(status=403, text="bad signature")
# POST = encrypted inbound event.
body = await request.text()
try:
xml = channel.crypto.decrypt_message(body, signature, timestamp, nonce)
except InvalidSignatureException:
return web.Response(status=403, text="bad signature")
try:
msg = parse_message(xml)
except Exception:
LOGGER.error("[wecom:%s] parse error", account_id, exc_info=True)
return web.Response(text="")
try:
await channel.handle_decrypted_message(msg)
except Exception:
LOGGER.error("[wecom:%s] handler error", account_id, exc_info=True)
except Exception:
LOGGER.error("[wecom:%s] inbound request handling error", account_id, exc_info=True)
# Empty 200 OK tells WeCom we accepted the event.
return web.Response(text="")
_servers: Dict[Tuple[str, int], _SharedWebhookServer] = {}
_active_per_server: Dict[Tuple[str, int], int] = {}
async def _acquire_server(host: str, port: int) -> _SharedWebhookServer:
key = (host, port)
server = _servers.get(key)
if server is None:
server = _SharedWebhookServer(host, port)
_servers[key] = server
await server.start()
_active_per_server[key] = _active_per_server.get(key, 0) + 1
return server
async def _release_server(host: str, port: int) -> None:
key = (host, port)
remaining = _active_per_server.get(key, 0) - 1
_active_per_server[key] = remaining
if remaining <= 0:
server = _servers.pop(key, None)
_active_per_server.pop(key, None)
if server is not None:
await server.stop()
class WeComChannel(Channel):
channel_id = "wecom"
def __init__(self, account: WeComAccount) -> None:
super().__init__()
self.account = account
self.account_id = account.account_id
self.crypto = WeChatCrypto(
account.token, account.aes_key, account.corp_id
)
self._server: Optional[_SharedWebhookServer] = None
self._access_token: Optional[str] = None
self._access_token_expires_at: float = 0.0
self._access_token_lock = asyncio.Lock()
async def start(self) -> None:
self._server = await _acquire_server(
self.account.webhook_host, self.account.webhook_port
)
self._server.channels[self.account_id] = self
LOGGER.info(
"[wecom:%s] registered at path /wecom/%s/callback (agent_id=%s)",
self.account_id,
self.account_id,
self.account.agent_id,
)
async def stop(self) -> None:
if self._server is not None:
self._server.channels.pop(self.account_id, None)
await _release_server(
self.account.webhook_host, self.account.webhook_port
)
self._server = None
async def handle_decrypted_message(self, msg) -> None:
try:
# Only handle plain text events; ignore image/voice/event etc.
if getattr(msg, "type", "") != "text":
return
user_id = str(getattr(msg, "source", "") or "")
if not user_id:
return
incoming = IncomingMessage(
channel=self.channel_id,
account_id=self.account_id,
chat_id=user_id,
chat_type="p2p",
message_id=str(getattr(msg, "id", "") or ""),
sender_id=user_id,
text=getattr(msg, "content", "") or "",
raw=msg,
)
await self._dispatch(incoming)
except Exception:
LOGGER.error("[wecom:%s] inbound message handling error", self.account_id, exc_info=True)
async def _get_access_token(self) -> str:
async with self._access_token_lock:
now = time.time()
if self._access_token and now < self._access_token_expires_at:
return self._access_token
params = {
"corpid": self.account.corp_id,
"corpsecret": self.account.secret,
}
async with aiohttp.ClientSession() as session:
async with session.get(
f"{WECOM_API_BASE}/gettoken", params=params
) as resp:
data = await resp.json(content_type=None)
if data.get("errcode", 0) != 0 or "access_token" not in data:
raise RuntimeError(f"wecom gettoken failed: {data}")
self._access_token = data["access_token"]
# 60s safety margin against clock skew / in-flight calls.
self._access_token_expires_at = (
now + int(data.get("expires_in", 7200)) - 60
)
return self._access_token
async def send(self, message: OutgoingMessage) -> None:
if not message.chat_id:
LOGGER.error("[wecom:%s] missing chat_id; cannot send", self.account_id)
return
try:
token = await self._get_access_token()
except Exception:
LOGGER.error("[wecom:%s] access_token error", self.account_id, exc_info=True)
return
payload = {
"touser": message.chat_id,
"msgtype": "text",
"agentid": int(self.account.agent_id),
"text": {"content": message.text},
"safe": 0,
}
try:
async with aiohttp.ClientSession() as session:
async with session.post(
f"{WECOM_API_BASE}/message/send",
params={"access_token": token},
json=payload,
) as resp:
data = await resp.json(content_type=None)
except Exception:
LOGGER.error("[wecom:%s] send transport error", self.account_id, exc_info=True)
return
if data.get("errcode", 0) != 0:
# 40014 / 42001 = access_token expired or invalid; drop cache.
if data.get("errcode") in (40014, 42001):
self._access_token = None
self._access_token_expires_at = 0.0
LOGGER.error("[wecom:%s] send failed: %s", self.account_id, data)
def _build(account_id: str, cfg: dict) -> Channel:
required = ("corp_id", "agent_id", "secret", "token", "aes_key")
missing = [k for k in required if not cfg.get(k)]
if missing:
raise ValueError(
f"wecom account '{account_id}' missing required fields: {missing}"
)
try:
agent_id = int(cfg["agent_id"])
except (TypeError, ValueError) as err:
raise ValueError(
f"wecom account '{account_id}' agent_id must be int: {err}"
) from err
# WeCom EncodingAESKey is always 43 characters; reject placeholders early so
# the failure is a clear message instead of a base64 "Incorrect padding" error.
aes_key = str(cfg["aes_key"])
if len(aes_key) != 43:
raise ValueError(
f"wecom account '{account_id}' aes_key (EncodingAESKey) must be 43 characters, got {len(aes_key)}"
)
return WeComChannel(
WeComAccount(
account_id=account_id,
corp_id=str(cfg["corp_id"]),
agent_id=agent_id,
secret=str(cfg["secret"]),
token=str(cfg["token"]),
aes_key=str(cfg["aes_key"]),
webhook_host=str(cfg.get("webhook_host", "0.0.0.0")),
webhook_port=int(cfg.get("webhook_port", 3002)),
)
)
register_channel("wecom", _build)

17
api/db/db_models.py
View File

@@ -1215,6 +1215,22 @@ class Connector2Kb(DataBaseModel):
db_table = "connector2kb"
class ChatChannel(DataBaseModel):
id = CharField(max_length=32, primary_key=True)
tenant_id = CharField(max_length=32, null=False, index=True)
name = CharField(max_length=128, null=False, help_text="Bot name", index=False)
channel = CharField(max_length=128, null=False, help_text="Chat channel type", index=True)
config = JSONField(null=False, default={}, help_text="Channel credential & settings")
dialog_id = CharField(max_length=32, null=True, default=None, help_text="connected dialog id", index=True)
status = CharField(max_length=16, null=True, help_text="1: valid, 0: invalid", default="1", index=True)
def __str__(self):
return self.name
class Meta:
db_table = "chat_channel"
class DateTimeTzField(CharField):
field_type = 'VARCHAR'
@@ -1693,6 +1709,7 @@ def migrate_db():
alter_db_add_column(migrator, "canvas_template", "canvas_category", CharField(max_length=32, null=False, default="agent_canvas", help_text="agent_canvas|dataflow_canvas", index=True))
alter_db_add_column(migrator, "canvas_template", "canvas_types", ListField(null=True, default=list, help_text="Canvas types"))
alter_db_add_column(migrator, "knowledgebase", "pipeline_id", CharField(max_length=32, null=True, help_text="Pipeline ID", index=True))
alter_db_add_column(migrator, "chat_channel", "dialog_id", CharField(max_length=32, null=True, help_text="connected dialog id", index=True))
alter_db_add_column(migrator, "document", "pipeline_id", CharField(max_length=32, null=True, help_text="Pipeline ID", index=True))
alter_db_add_column(migrator, "knowledgebase", "graphrag_task_id", CharField(max_length=32, null=True, help_text="Gragh RAG task ID", index=True))
alter_db_add_column(migrator, "knowledgebase", "raptor_task_id", CharField(max_length=32, null=True, help_text="RAPTOR task ID", index=True))

82
api/db/services/chat_channel_service.py Normal file
View File

@@ -0,0 +1,82 @@
#
# Copyright 2024 The InfiniFlow Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import logging
from peewee import JOIN
from api.db.db_models import DB, ChatChannel, Dialog
from api.db.services.common_service import CommonService
LOGGER = logging.getLogger(__name__)
class ChatChannelService(CommonService):
model = ChatChannel
@classmethod
@DB.connection_context()
def list(cls, tenant_id):
"""List a tenant's chat channel bots with their connected dialog (no credentials)."""
fields = [
cls.model.id,
cls.model.name,
cls.model.channel,
cls.model.dialog_id,
cls.model.status,
Dialog.name.alias("dialog_name"),
]
return list(
cls.model.select(*fields)
.join(
Dialog,
join_type=JOIN.LEFT_OUTER,
on=(Dialog.id == cls.model.dialog_id),
)
.where(cls.model.tenant_id == tenant_id)
.order_by(cls.model.create_time.desc())
.dicts()
)
@classmethod
@DB.connection_context()
def list_active(cls):
"""Return all enabled chat channel bots across tenants (with credentials)."""
return list(cls.model.select().where(cls.model.status == "1"))
@classmethod
@DB.connection_context()
def accessible(cls, channel_id: str, user_id: str) -> bool:
"""Return whether the user can access the chat channel's tenant."""
e, channel = cls.get_by_id(channel_id)
if not e:
LOGGER.warning("chat channel access denied: not found channel_id=%s user_id=%s", channel_id, user_id)
return False
if channel.tenant_id == user_id:
return True
from api.db.services.user_service import TenantService
joined_tenants = TenantService.get_joined_tenants_by_user_id(user_id)
has_access = any(tenant["tenant_id"] == channel.tenant_id for tenant in joined_tenants)
if not has_access:
LOGGER.warning(
"chat channel access denied: tenant mismatch channel_id=%s user_id=%s tenant_id=%s",
channel_id,
user_id,
channel.tenant_id,
)
return has_access

24
api/db/services/conversation_service.py
View File

@@ -13,6 +13,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
import hashlib
import time
import logging
from uuid import uuid4
@@ -53,6 +54,29 @@ class ConversationService(CommonService):
return list(sessions.dicts())
@classmethod
@DB.connection_context()
def get_or_create_for_channel(cls, dialog_id, channel_id, chat_id, name=None):
"""Find or create the conversation backing one channel end-user chat.
A chat_channel is bound to a dialog; each end-user chat on that channel
keeps its own conversation history. The conversation is identified by a
deterministic id derived from (channel_id, chat_id) so history persists
across restarts without a back-reference column on the conversation.
"""
conv_id = hashlib.md5(f"{channel_id}:{chat_id}".encode("utf-8")).hexdigest()[:32]
conv = cls.model.get_or_none(cls.model.id == conv_id)
if conv is not None:
return conv
cls.save(
id=conv_id,
dialog_id=dialog_id,
name=name or f"channel:{channel_id}:{chat_id}",
message=[],
reference=[],
)
return cls.model.get_or_none(cls.model.id == conv_id)
@classmethod
@DB.connection_context()
def get_all_conversation_by_dialog_ids(cls, dialog_ids):

16
api/ragflow_server.py
View File

@@ -141,11 +141,27 @@ if __name__ == '__main__':
t = threading.Thread(target=update_progress, daemon=True)
t.start()
def start_chat_channels():
try:
from api.channels.bootstrap import start_channel_server
logging.info("Starting chat channel server thread")
t = threading.Thread(
target=start_channel_server,
args=(stop_event,),
daemon=True,
name="chat-channels",
)
t.start()
except Exception:
logging.exception("Failed to start chat channel server")
if RuntimeConfig.DEBUG:
if os.environ.get("WERKZEUG_RUN_MAIN") == "true":
threading.Timer(1.0, delayed_start_update_progress).start()
start_chat_channels()
else:
threading.Timer(1.0, delayed_start_update_progress).start()
start_chat_channels()
# start http server
try: