opensource/WeKnora
3
0
Fork 0
mirror of https://github.com/Tencent/WeKnora.git synced 2026-09-19 10:28:49 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: implement IM channel sessions and update message handling

Browse Source
This commit is contained in:
wizardchen
2026-03-16 02:40:42 +08:00
parent 6159e8e3f7
commit 8b1febbe99
8 changed files with 67 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/im/service.go
+28 -8
View File
@@ -2,7 +2,7 @@ package im
import (
"context"
"fmt"
"fmt"
"strings"
"sync"
"time"
@@ -65,6 +65,24 @@ type Service struct {
stopCh chan struct{}
}
func buildIMQARequest(
session *types.Session,
query string,
assistantMessageID string,
userMessageID string,
customAgent *types.CustomAgent,
kbIDs []string,
) *types.QARequest {
return &types.QARequest{
Session: session,
Query: query,
AssistantMessageID: assistantMessageID,
CustomAgent: customAgent,
KnowledgeBaseIDs: kbIDs,
UserMessageID: userMessageID,
}
}
// NewService creates a new IM service.
func NewService(
db *gorm.DB,
@@ -479,10 +497,11 @@ func (s *Service) handleMessageStream(ctx context.Context, msg *IncomingMessage,
requestID := uuid.New().String()
// Create user message
if _, err := s.messageService.CreateMessage(qaCtx, &types.Message{
userMsg, err := s.messageService.CreateMessage(qaCtx, &types.Message{
SessionID: session.ID, Role: "user", Content: msg.Content,
RequestID: requestID, CreatedAt: time.Now(), IsCompleted: true,
}); err != nil {
})
if err != nil {
return fmt.Errorf("create user message: %w", err)
}
@@ -498,10 +517,11 @@ func (s *Service) handleMessageStream(ctx context.Context, msg *IncomingMessage,
// Run QA async
go func() {
var err error
req := buildIMQARequest(session, msg.Content, assistantMsg.ID, userMsg.ID, customAgent, kbIDs)
if useAgent {
err = s.sessionService.AgentQA(qaCtx, session, msg.Content, assistantMsg.ID, "", eventBus, customAgent, kbIDs, nil)
err = s.sessionService.AgentQA(qaCtx, req, eventBus)
} else {
err = s.sessionService.KnowledgeQA(qaCtx, session, msg.Content, kbIDs, nil, assistantMsg.ID, "", false, eventBus, customAgent, false)
err = s.sessionService.KnowledgeQA(qaCtx, req, eventBus)
}
if err != nil {
logger.Errorf(ctx, "[IM] QA stream execution error: %v", err)
@@ -644,7 +664,6 @@ func (s *Service) runQA(ctx context.Context, session *types.Session, query strin
if err != nil {
return "", fmt.Errorf("create user message: %w", err)
}
_ = userMsg
// Create a placeholder assistant message
assistantMsg, err := s.messageService.CreateMessage(ctx, &types.Message{
@@ -661,10 +680,11 @@ func (s *Service) runQA(ctx context.Context, session *types.Session, query strin
// Run QA async
go func() {
var err error
req := buildIMQARequest(session, query, assistantMsg.ID, userMsg.ID, customAgent, kbIDs)
if useAgent {
err = s.sessionService.AgentQA(ctx, session, query, assistantMsg.ID, "", eventBus, customAgent, kbIDs, nil)
err = s.sessionService.AgentQA(ctx, req, eventBus)
} else {
err = s.sessionService.KnowledgeQA(ctx, session, query, kbIDs, nil, assistantMsg.ID, "", false, eventBus, customAgent, false)
err = s.sessionService.KnowledgeQA(ctx, req, eventBus)
}
if err != nil {
logger.Errorf(ctx, "[IM] QA execution error: %v", err)
migrations/versioned/000021_im_channel.down.sql
+3
View File
@@ -0,0 +1,3 @@
ALTER TABLE im_channel_sessions DROP COLUMN IF EXISTS im_channel_id;
DROP TABLE IF EXISTS im_channels;
DROP TABLE IF EXISTS im_channel_sessions;
migrations/versioned/000021_im_channel_sessions.up.sql → migrations/versioned/000021_im_channel.up.sql
+36 -3
View File
@@ -1,6 +1,6 @@
-- Migration: 000021_im_channel_sessions
-- Description: Create IM channel-to-session mapping table
DO $$ BEGIN RAISE NOTICE '[Migration 000021] Creating table: im_channel_sessions'; END $$;
-- Description: Create IM channel session mapping and IM channel configuration tables
DO $$ BEGIN RAISE NOTICE '[Migration 000021] Creating IM channel integration tables'; END $$;
CREATE TABLE IF NOT EXISTS im_channel_sessions (
id VARCHAR(36) PRIMARY KEY DEFAULT uuid_generate_v4(),
@@ -41,4 +41,37 @@ COMMENT ON COLUMN im_channel_sessions.agent_id IS 'Custom agent ID used for this
COMMENT ON COLUMN im_channel_sessions.status IS 'Channel status: active, paused, expired';
COMMENT ON COLUMN im_channel_sessions.metadata IS 'Platform-specific extra data (JSON)';
DO $$ BEGIN RAISE NOTICE '[Migration 000021] im_channel_sessions setup completed successfully!'; END $$;
DO $$ BEGIN RAISE NOTICE '[Migration 000021] Creating table: im_channels'; END $$;
CREATE TABLE IF NOT EXISTS im_channels (
id VARCHAR(36) PRIMARY KEY DEFAULT uuid_generate_v4(),
tenant_id BIGINT NOT NULL,
agent_id VARCHAR(36) NOT NULL,
platform VARCHAR(20) NOT NULL,
name VARCHAR(255) NOT NULL DEFAULT '',
enabled BOOLEAN NOT NULL DEFAULT true,
mode VARCHAR(20) NOT NULL DEFAULT 'websocket',
output_mode VARCHAR(20) NOT NULL DEFAULT 'stream',
credentials JSONB NOT NULL DEFAULT '{}',
created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
deleted_at TIMESTAMP WITH TIME ZONE
);
CREATE INDEX IF NOT EXISTS idx_im_channels_tenant ON im_channels (tenant_id);
CREATE INDEX IF NOT EXISTS idx_im_channels_agent ON im_channels (agent_id);
CREATE INDEX IF NOT EXISTS idx_im_channels_deleted ON im_channels (deleted_at) WHERE deleted_at IS NOT NULL;
COMMENT ON TABLE im_channels IS 'IM platform channel configurations bound to agents';
COMMENT ON COLUMN im_channels.agent_id IS 'Agent ID this channel is bound to';
COMMENT ON COLUMN im_channels.platform IS 'IM platform: wecom, feishu';
COMMENT ON COLUMN im_channels.name IS 'User-defined channel name for identification';
COMMENT ON COLUMN im_channels.mode IS 'Connection mode: webhook or websocket';
COMMENT ON COLUMN im_channels.output_mode IS 'Output mode: stream (real-time) or full (wait for complete answer)';
COMMENT ON COLUMN im_channels.credentials IS 'Platform credentials (JSONB): WeCom webhook={corp_id,agent_secret,token,encoding_aes_key,corp_agent_id}, WeCom ws={bot_id,bot_secret}, Feishu={app_id,app_secret,verification_token,encrypt_key}';
-- Add im_channel_id column to im_channel_sessions for linking
ALTER TABLE im_channel_sessions ADD COLUMN IF NOT EXISTS im_channel_id VARCHAR(36) DEFAULT '';
CREATE INDEX IF NOT EXISTS idx_im_channel_sessions_channel ON im_channel_sessions (im_channel_id) WHERE im_channel_id != '';
DO $$ BEGIN RAISE NOTICE '[Migration 000021] IM channel integration setup completed successfully!'; END $$;
migrations/versioned/000021_im_channel_sessions.down.sql
-1
View File
@@ -1 +0,0 @@
DROP TABLE IF EXISTS im_channel_sessions;
migrations/versioned/000022_im_channels.down.sql
-7
View File
@@ -1,7 +0,0 @@
-- Rollback: 000022_im_channels
DO $$ BEGIN RAISE NOTICE '[Migration 000022] Rolling back: im_channels'; END $$;
ALTER TABLE im_channel_sessions DROP COLUMN IF EXISTS im_channel_id;
DROP TABLE IF EXISTS im_channels;
DO $$ BEGIN RAISE NOTICE '[Migration 000022] Rollback completed'; END $$;
migrations/versioned/000022_im_channels.up.sql
-36
View File
@@ -1,36 +0,0 @@
-- Migration: 000022_im_channels
-- Description: Create IM channels table for database-driven IM integration (replaces config.yaml IM settings)
DO $$ BEGIN RAISE NOTICE '[Migration 000022] Creating table: im_channels'; END $$;
CREATE TABLE IF NOT EXISTS im_channels (
id VARCHAR(36) PRIMARY KEY DEFAULT uuid_generate_v4(),
tenant_id BIGINT NOT NULL,
agent_id VARCHAR(36) NOT NULL,
platform VARCHAR(20) NOT NULL,
name VARCHAR(255) NOT NULL DEFAULT '',
enabled BOOLEAN NOT NULL DEFAULT true,
mode VARCHAR(20) NOT NULL DEFAULT 'websocket',
output_mode VARCHAR(20) NOT NULL DEFAULT 'stream',
credentials JSONB NOT NULL DEFAULT '{}',
created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
deleted_at TIMESTAMP WITH TIME ZONE
);
CREATE INDEX IF NOT EXISTS idx_im_channels_tenant ON im_channels (tenant_id);
CREATE INDEX IF NOT EXISTS idx_im_channels_agent ON im_channels (agent_id);
CREATE INDEX IF NOT EXISTS idx_im_channels_deleted ON im_channels (deleted_at) WHERE deleted_at IS NOT NULL;
COMMENT ON TABLE im_channels IS 'IM platform channel configurations bound to agents';
COMMENT ON COLUMN im_channels.agent_id IS 'Agent ID this channel is bound to';
COMMENT ON COLUMN im_channels.platform IS 'IM platform: wecom, feishu';
COMMENT ON COLUMN im_channels.name IS 'User-defined channel name for identification';
COMMENT ON COLUMN im_channels.mode IS 'Connection mode: webhook or websocket';
COMMENT ON COLUMN im_channels.output_mode IS 'Output mode: stream (real-time) or full (wait for complete answer)';
COMMENT ON COLUMN im_channels.credentials IS 'Platform credentials (JSONB): WeCom webhook={corp_id,agent_secret,token,encoding_aes_key,corp_agent_id}, WeCom ws={bot_id,bot_secret}, Feishu={app_id,app_secret,verification_token,encrypt_key}';
-- Add im_channel_id column to im_channel_sessions for linking
ALTER TABLE im_channel_sessions ADD COLUMN IF NOT EXISTS im_channel_id VARCHAR(36) DEFAULT '';
CREATE INDEX IF NOT EXISTS idx_im_channel_sessions_channel ON im_channel_sessions (im_channel_id) WHERE im_channel_id != '';
DO $$ BEGIN RAISE NOTICE '[Migration 000022] im_channels setup completed successfully!'; END $$;
migrations/versioned/000021_message_images.down.sql → migrations/versioned/000022_message_images.down.sql
View File
migrations/versioned/000021_message_images.up.sql → migrations/versioned/000022_message_images.up.sql
View File