feat(im): add file support and knowledge base integration
This commit is contained in:
wizardchen
+2
-6
@@ -141,12 +141,8 @@ conversation:
|
||||
- Do not copy or reference any content from examples; ensure the summary is entirely based on the user's new article
|
||||
- Make every effort to extract key points and summarize for any text, regardless of length or complexity
|
||||
|
||||
## CRITICAL: Language Rule
|
||||
- Write the summary in the SAME LANGUAGE as the original document
|
||||
- If the document is in Korean, write the summary in Korean
|
||||
- If the document is in English, write the summary in English
|
||||
- If the document is in Chinese, write the summary in Chinese
|
||||
- NEVER translate the content to a different language
|
||||
## Requirements:
|
||||
- Use the Chinese language for all outputs
|
||||
|
||||
## The following is the article information provided by the user:
|
||||
|
||||
|
||||
@@ -187,6 +187,7 @@ export interface IMChannel {
|
||||
enabled: boolean;
|
||||
mode: 'webhook' | 'websocket';
|
||||
output_mode: 'stream' | 'full';
|
||||
knowledge_base_id?: string;
|
||||
credentials: Record<string, any>;
|
||||
created_at?: string;
|
||||
updated_at?: string;
|
||||
|
||||
@@ -116,6 +116,20 @@
|
||||
</t-radio-group>
|
||||
</div>
|
||||
|
||||
<!-- Knowledge base for file messages -->
|
||||
<div class="form-item">
|
||||
<label class="form-label">{{ $t('agentEditor.im.fileKnowledgeBase') }}</label>
|
||||
<t-select
|
||||
v-model="formData.knowledge_base_id"
|
||||
:placeholder="$t('agentEditor.im.fileKnowledgeBasePlaceholder')"
|
||||
clearable
|
||||
filterable
|
||||
>
|
||||
<t-option v-for="kb in knowledgeBases" :key="kb.id" :value="kb.id" :label="kb.name" />
|
||||
</t-select>
|
||||
<p class="form-hint">{{ $t('agentEditor.im.fileKnowledgeBaseHint') }}</p>
|
||||
</div>
|
||||
|
||||
<!-- Credentials divider -->
|
||||
<div class="form-divider"></div>
|
||||
|
||||
@@ -200,6 +214,7 @@ import { ref, onMounted } from 'vue';
|
||||
import { useI18n } from 'vue-i18n';
|
||||
import { MessagePlugin } from 'tdesign-vue-next';
|
||||
import { listIMChannels, createIMChannel, updateIMChannel, deleteIMChannel, toggleIMChannel } from '@/api/agent';
|
||||
import { listKnowledgeBases } from '@/api/knowledge-base';
|
||||
import type { IMChannel } from '@/api/agent';
|
||||
|
||||
const { t } = useI18n();
|
||||
@@ -213,6 +228,9 @@ const loading = ref(false);
|
||||
const showCreateDialog = ref(false);
|
||||
const editingChannel = ref<IMChannel | null>(null);
|
||||
|
||||
// Knowledge base options for file-to-KB feature
|
||||
const knowledgeBases = ref<{ id: string; name: string }[]>([]);
|
||||
|
||||
const defaultCredentials = (): Record<string, any> => ({});
|
||||
|
||||
const formData = ref({
|
||||
@@ -220,14 +238,19 @@ const formData = ref({
|
||||
name: '',
|
||||
mode: 'websocket' as 'webhook' | 'websocket',
|
||||
output_mode: 'stream' as 'stream' | 'full',
|
||||
knowledge_base_id: '',
|
||||
credentials: defaultCredentials(),
|
||||
});
|
||||
|
||||
async function loadChannels() {
|
||||
loading.value = true;
|
||||
try {
|
||||
const res = await listIMChannels(props.agentId);
|
||||
channels.value = res.data || [];
|
||||
const [channelRes, kbRes] = await Promise.all([
|
||||
listIMChannels(props.agentId),
|
||||
listKnowledgeBases(),
|
||||
]);
|
||||
channels.value = channelRes.data || [];
|
||||
knowledgeBases.value = (kbRes.data || []).map((kb: any) => ({ id: kb.id, name: kb.name }));
|
||||
} catch {
|
||||
channels.value = [];
|
||||
} finally {
|
||||
@@ -256,6 +279,7 @@ function editChannel(channel: IMChannel) {
|
||||
name: channel.name,
|
||||
mode: channel.mode,
|
||||
output_mode: channel.output_mode,
|
||||
knowledge_base_id: channel.knowledge_base_id || '',
|
||||
credentials: { ...channel.credentials },
|
||||
};
|
||||
showCreateDialog.value = true;
|
||||
@@ -268,6 +292,7 @@ function resetForm() {
|
||||
name: '',
|
||||
mode: 'websocket',
|
||||
output_mode: 'stream',
|
||||
knowledge_base_id: '',
|
||||
credentials: defaultCredentials(),
|
||||
};
|
||||
}
|
||||
@@ -279,6 +304,7 @@ async function handleSave() {
|
||||
name: formData.value.name,
|
||||
mode: formData.value.mode,
|
||||
output_mode: formData.value.output_mode,
|
||||
knowledge_base_id: formData.value.knowledge_base_id,
|
||||
credentials: formData.value.credentials,
|
||||
});
|
||||
MessagePlugin.success(t('common.updateSuccess'));
|
||||
@@ -288,6 +314,7 @@ async function handleSave() {
|
||||
name: formData.value.name,
|
||||
mode: formData.value.mode,
|
||||
output_mode: formData.value.output_mode,
|
||||
knowledge_base_id: formData.value.knowledge_base_id,
|
||||
credentials: formData.value.credentials,
|
||||
});
|
||||
MessagePlugin.success(t('common.createSuccess'));
|
||||
@@ -296,7 +323,8 @@ async function handleSave() {
|
||||
resetForm();
|
||||
await loadChannels();
|
||||
} catch (e: any) {
|
||||
MessagePlugin.error(e?.message || t('common.operationFailed'));
|
||||
const msg = e?.message || (typeof e?.error === 'string' ? e.error : null) || t('common.operationFailed');
|
||||
MessagePlugin.error(msg);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -3275,6 +3275,9 @@ export default {
|
||||
feishuConsole: 'Feishu Open Platform',
|
||||
modeHint: 'WebSocket is recommended for easier setup',
|
||||
consoleTip: 'to get credentials',
|
||||
fileKnowledgeBase: 'File Storage Knowledge Base',
|
||||
fileKnowledgeBasePlaceholder: 'Select a knowledge base (optional)',
|
||||
fileKnowledgeBaseHint: 'When configured, files sent by users will be automatically saved to this knowledge base',
|
||||
},
|
||||
mcp: {
|
||||
label: 'MCP Services',
|
||||
|
||||
@@ -3305,6 +3305,9 @@ export default {
|
||||
feishuConsole: "Feishu 개발 플랫폼",
|
||||
modeHint: "WebSocket 방식이 설정이 더 간편하여 권장됩니다",
|
||||
consoleTip: "자격 증명 정보를 가져오세요",
|
||||
fileKnowledgeBase: "파일 저장 지식 베이스",
|
||||
fileKnowledgeBasePlaceholder: "지식 베이스 선택 (선택 사항)",
|
||||
fileKnowledgeBaseHint: "설정 시 사용자가 보낸 파일이 자동으로 해당 지식 베이스에 저장됩니다",
|
||||
},
|
||||
tools: {
|
||||
thinking: '사고',
|
||||
|
||||
@@ -2858,6 +2858,9 @@ export default {
|
||||
feishuConsole: 'Платформа Feishu',
|
||||
modeHint: 'Рекомендуется WebSocket — проще настроить',
|
||||
consoleTip: 'для получения учётных данных',
|
||||
fileKnowledgeBase: 'База знаний для файлов',
|
||||
fileKnowledgeBasePlaceholder: 'Выберите базу знаний (необязательно)',
|
||||
fileKnowledgeBaseHint: 'При настройке файлы, отправленные пользователями, автоматически сохраняются в эту базу знаний',
|
||||
},
|
||||
mcp: {
|
||||
label: 'MCP-сервисы',
|
||||
|
||||
@@ -3250,6 +3250,9 @@ export default {
|
||||
feishuConsole: "飞书开放平台",
|
||||
modeHint: "推荐使用 WebSocket 方式接入，配置更简单",
|
||||
consoleTip: "前往获取凭证信息",
|
||||
fileKnowledgeBase: "文件保存知识库",
|
||||
fileKnowledgeBasePlaceholder: "选择知识库（可选）",
|
||||
fileKnowledgeBaseHint: "配置后，用户发送的文件将自动保存到该知识库中",
|
||||
},
|
||||
tools: {
|
||||
thinking: "思考",
|
||||
|
||||
@@ -18,6 +18,12 @@ import (
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
const (
|
||||
// llmPerCallTimeout is the maximum time allowed for a single LLM call (stream initiation + full response).
|
||||
// This prevents a single slow call from consuming the entire pipeline deadline.
|
||||
llmPerCallTimeout = 120 * time.Second
|
||||
)
|
||||
|
||||
// generateEventID generates a unique event ID with type suffix for better traceability
|
||||
func generateEventID(suffix string) string {
|
||||
return fmt.Sprintf("%s-%s", uuid.New().String()[:8], suffix)
|
||||
@@ -719,7 +725,14 @@ func (e *AgentEngine) streamLLMToEventBus(
|
||||
) (string, []types.LLMToolCall, error) {
|
||||
logger.Debugf(ctx, "[Agent][Stream] Starting LLM stream with %d messages", len(messages))
|
||||
|
||||
stream, err := e.chatModel.ChatStream(ctx, messages, opts)
|
||||
// Create a per-call timeout context so that a single LLM call gets a
|
||||
// guaranteed time window even when the parent context's deadline is almost
|
||||
// exhausted after previous iterations. The shorter of the two deadlines wins,
|
||||
// so the parent pipeline timeout is still respected.
|
||||
llmCtx, llmCancel := context.WithTimeout(ctx, llmPerCallTimeout)
|
||||
defer llmCancel()
|
||||
|
||||
stream, err := e.chatModel.ChatStream(llmCtx, messages, opts)
|
||||
if err != nil {
|
||||
logger.Errorf(ctx, "[Agent][Stream] Failed to start LLM stream: %v", err)
|
||||
return "", nil, err
|
||||
|
||||
@@ -29,7 +29,7 @@ const (
|
||||
"4. Organize content in the original reading order\n" +
|
||||
"5. Only output extracted text content, do not add any HTML tags\n" +
|
||||
"If there is no recognizable text content in the image, reply: No text content."
|
||||
vlmCaptionPrompt = "Provide a brief and concise description of the main content of the image"
|
||||
vlmCaptionPrompt = "Provide a brief and concise description of the main content of the image in Chinese"
|
||||
)
|
||||
|
||||
// ImageMultimodalService handles image:multimodal asynq tasks.
|
||||
|
||||
+62
-23
@@ -3,6 +3,7 @@ package handler
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
"github.com/Tencent/WeKnora/internal/im"
|
||||
"github.com/Tencent/WeKnora/internal/logger"
|
||||
@@ -39,12 +40,13 @@ func (h *IMHandler) CreateIMChannel(c *gin.Context) {
|
||||
}
|
||||
|
||||
var req struct {
|
||||
Platform string `json:"platform" binding:"required"`
|
||||
Name string `json:"name"`
|
||||
Mode string `json:"mode"`
|
||||
OutputMode string `json:"output_mode"`
|
||||
Credentials types.JSON `json:"credentials"`
|
||||
Enabled *bool `json:"enabled"`
|
||||
Platform string `json:"platform" binding:"required"`
|
||||
Name string `json:"name"`
|
||||
Mode string `json:"mode"`
|
||||
OutputMode string `json:"output_mode"`
|
||||
KnowledgeBaseID string `json:"knowledge_base_id"`
|
||||
Credentials types.JSON `json:"credentials"`
|
||||
Enabled *bool `json:"enabled"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&req); err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||||
@@ -57,14 +59,15 @@ func (h *IMHandler) CreateIMChannel(c *gin.Context) {
|
||||
}
|
||||
|
||||
channel := &im.IMChannel{
|
||||
TenantID: tenantID,
|
||||
AgentID: agentID,
|
||||
Platform: req.Platform,
|
||||
Name: req.Name,
|
||||
Mode: req.Mode,
|
||||
OutputMode: req.OutputMode,
|
||||
Credentials: req.Credentials,
|
||||
Enabled: true,
|
||||
TenantID: tenantID,
|
||||
AgentID: agentID,
|
||||
Platform: req.Platform,
|
||||
Name: req.Name,
|
||||
Mode: req.Mode,
|
||||
OutputMode: req.OutputMode,
|
||||
KnowledgeBaseID: req.KnowledgeBaseID,
|
||||
Credentials: req.Credentials,
|
||||
Enabled: true,
|
||||
}
|
||||
if req.Enabled != nil {
|
||||
channel.Enabled = *req.Enabled
|
||||
@@ -81,6 +84,10 @@ func (h *IMHandler) CreateIMChannel(c *gin.Context) {
|
||||
|
||||
if err := h.imService.CreateChannel(channel); err != nil {
|
||||
logger.Errorf(c.Request.Context(), "[IM] Create channel failed: %v", err)
|
||||
if strings.HasPrefix(err.Error(), "duplicate_bot:") {
|
||||
c.JSON(http.StatusConflict, gin.H{"error": strings.TrimPrefix(err.Error(), "duplicate_bot: ")})
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create channel"})
|
||||
return
|
||||
}
|
||||
@@ -96,7 +103,13 @@ func (h *IMHandler) ListIMChannels(c *gin.Context) {
|
||||
return
|
||||
}
|
||||
|
||||
channels, err := h.imService.ListChannelsByAgent(agentID)
|
||||
tenantID, ok := c.Request.Context().Value(types.TenantIDContextKey).(uint64)
|
||||
if !ok {
|
||||
c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized"})
|
||||
return
|
||||
}
|
||||
|
||||
channels, err := h.imService.ListChannelsByAgent(agentID, tenantID)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list channels"})
|
||||
return
|
||||
@@ -113,18 +126,25 @@ func (h *IMHandler) UpdateIMChannel(c *gin.Context) {
|
||||
return
|
||||
}
|
||||
|
||||
channel, err := h.imService.GetChannelByID(channelID)
|
||||
tenantID, ok := c.Request.Context().Value(types.TenantIDContextKey).(uint64)
|
||||
if !ok {
|
||||
c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized"})
|
||||
return
|
||||
}
|
||||
|
||||
channel, err := h.imService.GetChannelByIDAndTenant(channelID, tenantID)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": "channel not found"})
|
||||
return
|
||||
}
|
||||
|
||||
var req struct {
|
||||
Name *string `json:"name"`
|
||||
Mode *string `json:"mode"`
|
||||
OutputMode *string `json:"output_mode"`
|
||||
Credentials types.JSON `json:"credentials"`
|
||||
Enabled *bool `json:"enabled"`
|
||||
Name *string `json:"name"`
|
||||
Mode *string `json:"mode"`
|
||||
OutputMode *string `json:"output_mode"`
|
||||
KnowledgeBaseID *string `json:"knowledge_base_id"`
|
||||
Credentials types.JSON `json:"credentials"`
|
||||
Enabled *bool `json:"enabled"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&req); err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||||
@@ -140,6 +160,9 @@ func (h *IMHandler) UpdateIMChannel(c *gin.Context) {
|
||||
if req.OutputMode != nil {
|
||||
channel.OutputMode = *req.OutputMode
|
||||
}
|
||||
if req.KnowledgeBaseID != nil {
|
||||
channel.KnowledgeBaseID = *req.KnowledgeBaseID
|
||||
}
|
||||
if req.Credentials != nil {
|
||||
channel.Credentials = req.Credentials
|
||||
}
|
||||
@@ -149,6 +172,10 @@ func (h *IMHandler) UpdateIMChannel(c *gin.Context) {
|
||||
|
||||
if err := h.imService.UpdateChannel(channel); err != nil {
|
||||
logger.Errorf(c.Request.Context(), "[IM] Update channel failed: %v", err)
|
||||
if strings.HasPrefix(err.Error(), "duplicate_bot:") {
|
||||
c.JSON(http.StatusConflict, gin.H{"error": strings.TrimPrefix(err.Error(), "duplicate_bot: ")})
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to update channel"})
|
||||
return
|
||||
}
|
||||
@@ -164,7 +191,13 @@ func (h *IMHandler) DeleteIMChannel(c *gin.Context) {
|
||||
return
|
||||
}
|
||||
|
||||
if err := h.imService.DeleteChannel(channelID); err != nil {
|
||||
tenantID, ok := c.Request.Context().Value(types.TenantIDContextKey).(uint64)
|
||||
if !ok {
|
||||
c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized"})
|
||||
return
|
||||
}
|
||||
|
||||
if err := h.imService.DeleteChannel(channelID, tenantID); err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete channel"})
|
||||
return
|
||||
}
|
||||
@@ -180,7 +213,13 @@ func (h *IMHandler) ToggleIMChannel(c *gin.Context) {
|
||||
return
|
||||
}
|
||||
|
||||
channel, err := h.imService.ToggleChannel(channelID)
|
||||
tenantID, ok := c.Request.Context().Value(types.TenantIDContextKey).(uint64)
|
||||
if !ok {
|
||||
c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized"})
|
||||
return
|
||||
}
|
||||
|
||||
channel, err := h.imService.ToggleChannel(channelID, tenantID)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to toggle channel"})
|
||||
return
|
||||
|
||||
+29
-1
@@ -2,6 +2,7 @@ package im
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
@@ -14,10 +15,21 @@ const (
|
||||
PlatformFeishu Platform = "feishu"
|
||||
)
|
||||
|
||||
// MessageType identifies the kind of IM message.
|
||||
type MessageType string
|
||||
|
||||
const (
|
||||
MessageTypeText MessageType = "text"
|
||||
MessageTypeFile MessageType = "file"
|
||||
MessageTypeImage MessageType = "image"
|
||||
)
|
||||
|
||||
// IncomingMessage is the unified message parsed from an IM callback.
|
||||
type IncomingMessage struct {
|
||||
// Platform identifies which IM platform the message comes from.
|
||||
Platform Platform
|
||||
// MessageType is "text" (default) or "file".
|
||||
MessageType MessageType
|
||||
// UserID is the IM-platform user identifier.
|
||||
UserID string
|
||||
// UserName is the display name of the user (optional).
|
||||
@@ -26,10 +38,16 @@ type IncomingMessage struct {
|
||||
ChatID string
|
||||
// ChatType distinguishes direct message from group chat.
|
||||
ChatType ChatType
|
||||
// Content is the text content of the message.
|
||||
// Content is the text content of the message (empty for file messages).
|
||||
Content string
|
||||
// MessageID is the IM-platform message identifier (for dedup).
|
||||
MessageID string
|
||||
// FileKey is the platform file identifier (for file messages).
|
||||
FileKey string
|
||||
// FileName is the original file name (for file messages).
|
||||
FileName string
|
||||
// FileSize is the file size in bytes (for file messages, optional).
|
||||
FileSize int64
|
||||
// Extra holds platform-specific fields (e.g., WeCom stream ID).
|
||||
Extra map[string]string
|
||||
}
|
||||
@@ -89,3 +107,13 @@ type StreamSender interface {
|
||||
// EndStream finalizes a streaming reply.
|
||||
EndStream(ctx context.Context, incoming *IncomingMessage, streamID string) error
|
||||
}
|
||||
|
||||
// FileDownloader is an optional interface that adapters can implement to support
|
||||
// downloading file attachments from the IM platform. When the adapter implements
|
||||
// this interface and the IM channel has a knowledge_base_id configured, file
|
||||
// messages will be downloaded and saved to the specified knowledge base.
|
||||
type FileDownloader interface {
|
||||
// DownloadFile downloads a file resource from the IM platform.
|
||||
// Returns the file content reader, the resolved file name, and any error.
|
||||
DownloadFile(ctx context.Context, msg *IncomingMessage) (io.ReadCloser, string, error)
|
||||
}
|
||||
|
||||
+196
-33
@@ -29,8 +29,9 @@ import (
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
// Compile-time check that Adapter implements im.StreamSender.
|
||||
// Compile-time check that Adapter implements im.StreamSender and im.FileDownloader.
|
||||
var _ im.StreamSender = (*Adapter)(nil)
|
||||
var _ im.FileDownloader = (*Adapter)(nil)
|
||||
|
||||
var httpClient = &http.Client{Timeout: 10 * time.Second}
|
||||
|
||||
@@ -223,19 +224,6 @@ func (a *Adapter) ParseCallback(c *gin.Context) (*im.IncomingMessage, error) {
|
||||
}
|
||||
msg := eventBody.Event.Message
|
||||
|
||||
if msg.MessageType != "text" {
|
||||
logger.Infof(c.Request.Context(), "[Feishu] Ignoring non-text message type: %s", msg.MessageType)
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// Parse text content
|
||||
var textContent struct {
|
||||
Text string `json:"text"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(msg.Content), &textContent); err != nil {
|
||||
return nil, fmt.Errorf("unmarshal text content: %w", err)
|
||||
}
|
||||
|
||||
// Determine chat type
|
||||
chatType := im.ChatTypeDirect
|
||||
chatID := ""
|
||||
@@ -250,28 +238,146 @@ func (a *Adapter) ParseCallback(c *gin.Context) (*im.IncomingMessage, error) {
|
||||
openID = eventBody.Event.Sender.SenderID.OpenID
|
||||
}
|
||||
|
||||
// Strip @bot mention from group messages
|
||||
content := textContent.Text
|
||||
if chatType == im.ChatTypeGroup {
|
||||
// Feishu @mentions are in the format @_user_xxx
|
||||
for strings.HasPrefix(content, "@_user_") {
|
||||
idx := strings.Index(content, " ")
|
||||
if idx >= 0 {
|
||||
content = content[idx+1:]
|
||||
} else {
|
||||
break
|
||||
switch msg.MessageType {
|
||||
case "text":
|
||||
// Parse text content
|
||||
var textContent struct {
|
||||
Text string `json:"text"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(msg.Content), &textContent); err != nil {
|
||||
return nil, fmt.Errorf("unmarshal text content: %w", err)
|
||||
}
|
||||
|
||||
// Strip @bot mention from group messages
|
||||
content := textContent.Text
|
||||
if chatType == im.ChatTypeGroup {
|
||||
for strings.HasPrefix(content, "@_user_") {
|
||||
idx := strings.Index(content, " ")
|
||||
if idx >= 0 {
|
||||
content = content[idx+1:]
|
||||
} else {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return &im.IncomingMessage{
|
||||
Platform: im.PlatformFeishu,
|
||||
UserID: openID,
|
||||
ChatID: chatID,
|
||||
ChatType: chatType,
|
||||
Content: strings.TrimSpace(content),
|
||||
MessageID: msg.MessageID,
|
||||
}, nil
|
||||
return &im.IncomingMessage{
|
||||
Platform: im.PlatformFeishu,
|
||||
MessageType: im.MessageTypeText,
|
||||
UserID: openID,
|
||||
ChatID: chatID,
|
||||
ChatType: chatType,
|
||||
Content: strings.TrimSpace(content),
|
||||
MessageID: msg.MessageID,
|
||||
}, nil
|
||||
|
||||
case "file":
|
||||
var fileContent struct {
|
||||
FileKey string `json:"file_key"`
|
||||
FileName string `json:"file_name"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(msg.Content), &fileContent); err != nil {
|
||||
return nil, fmt.Errorf("unmarshal file content: %w", err)
|
||||
}
|
||||
if fileContent.FileKey == "" {
|
||||
return nil, nil
|
||||
}
|
||||
return &im.IncomingMessage{
|
||||
Platform: im.PlatformFeishu,
|
||||
MessageType: im.MessageTypeFile,
|
||||
UserID: openID,
|
||||
ChatID: chatID,
|
||||
ChatType: chatType,
|
||||
MessageID: msg.MessageID,
|
||||
FileKey: fileContent.FileKey,
|
||||
FileName: fileContent.FileName,
|
||||
}, nil
|
||||
|
||||
case "image":
|
||||
var imageContent struct {
|
||||
ImageKey string `json:"image_key"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(msg.Content), &imageContent); err != nil {
|
||||
return nil, fmt.Errorf("unmarshal image content: %w", err)
|
||||
}
|
||||
if imageContent.ImageKey == "" {
|
||||
return nil, nil
|
||||
}
|
||||
return &im.IncomingMessage{
|
||||
Platform: im.PlatformFeishu,
|
||||
MessageType: im.MessageTypeImage,
|
||||
UserID: openID,
|
||||
ChatID: chatID,
|
||||
ChatType: chatType,
|
||||
MessageID: msg.MessageID,
|
||||
FileKey: imageContent.ImageKey,
|
||||
FileName: imageContent.ImageKey + ".png",
|
||||
}, nil
|
||||
|
||||
case "post":
|
||||
// Rich text: extract plain text for QA
|
||||
var postContent struct {
|
||||
Title string `json:"title"`
|
||||
Content [][]json.RawMessage `json:"content"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(msg.Content), &postContent); err != nil {
|
||||
return nil, fmt.Errorf("unmarshal post content: %w", err)
|
||||
}
|
||||
|
||||
var textParts []string
|
||||
if postContent.Title != "" {
|
||||
textParts = append(textParts, postContent.Title)
|
||||
}
|
||||
for _, line := range postContent.Content {
|
||||
var lineText strings.Builder
|
||||
for _, elem := range line {
|
||||
var tag struct {
|
||||
Tag string `json:"tag"`
|
||||
Text string `json:"text"`
|
||||
}
|
||||
if err := json.Unmarshal(elem, &tag); err != nil {
|
||||
continue
|
||||
}
|
||||
switch tag.Tag {
|
||||
case "text", "a":
|
||||
lineText.WriteString(tag.Text)
|
||||
}
|
||||
}
|
||||
if t := strings.TrimSpace(lineText.String()); t != "" {
|
||||
textParts = append(textParts, t)
|
||||
}
|
||||
}
|
||||
|
||||
content := strings.Join(textParts, "\n")
|
||||
if chatType == im.ChatTypeGroup {
|
||||
for strings.HasPrefix(content, "@_user_") {
|
||||
idx := strings.Index(content, " ")
|
||||
if idx >= 0 {
|
||||
content = content[idx+1:]
|
||||
} else {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
content = strings.TrimSpace(content)
|
||||
if content == "" {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
return &im.IncomingMessage{
|
||||
Platform: im.PlatformFeishu,
|
||||
MessageType: im.MessageTypeText,
|
||||
UserID: openID,
|
||||
ChatID: chatID,
|
||||
ChatType: chatType,
|
||||
Content: content,
|
||||
MessageID: msg.MessageID,
|
||||
}, nil
|
||||
|
||||
default:
|
||||
logger.Infof(c.Request.Context(), "[Feishu] Ignoring unsupported message type: %s", msg.MessageType)
|
||||
return nil, nil
|
||||
}
|
||||
}
|
||||
|
||||
// SendReply sends a reply message via Feishu API.
|
||||
@@ -327,6 +433,63 @@ func (a *Adapter) SendReply(ctx context.Context, incoming *im.IncomingMessage, r
|
||||
return nil
|
||||
}
|
||||
|
||||
// ──────────────────────────────────────────────────────────────────────
|
||||
// File download support via Feishu GetMessageResource API
|
||||
// ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
// DownloadFile downloads a file or image attachment from a Feishu message.
|
||||
// Uses the GetMessageResource API: GET /open-apis/im/v1/messages/:message_id/resources/:file_key?type={file|image}
|
||||
func (a *Adapter) DownloadFile(ctx context.Context, msg *im.IncomingMessage) (io.ReadCloser, string, error) {
|
||||
if msg.FileKey == "" || msg.MessageID == "" {
|
||||
return nil, "", fmt.Errorf("file_key and message_id are required")
|
||||
}
|
||||
|
||||
accessToken, err := a.getTenantAccessToken(ctx)
|
||||
if err != nil {
|
||||
return nil, "", fmt.Errorf("get access token: %w", err)
|
||||
}
|
||||
|
||||
// Determine resource type based on message type
|
||||
resourceType := "file"
|
||||
if msg.MessageType == im.MessageTypeImage {
|
||||
resourceType = "image"
|
||||
}
|
||||
|
||||
apiURL := fmt.Sprintf("https://open.feishu.cn/open-apis/im/v1/messages/%s/resources/%s?type=%s",
|
||||
msg.MessageID, msg.FileKey, resourceType)
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, apiURL, nil)
|
||||
if err != nil {
|
||||
return nil, "", fmt.Errorf("create request: %w", err)
|
||||
}
|
||||
req.Header.Set("Authorization", "Bearer "+accessToken)
|
||||
|
||||
resp, err := httpClient.Do(req)
|
||||
if err != nil {
|
||||
return nil, "", fmt.Errorf("download file: %w", err)
|
||||
}
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
resp.Body.Close()
|
||||
return nil, "", fmt.Errorf("download file failed: status=%d", resp.StatusCode)
|
||||
}
|
||||
|
||||
// Use the original file name from the message, or extract from Content-Disposition
|
||||
fileName := msg.FileName
|
||||
if fileName == "" {
|
||||
if cd := resp.Header.Get("Content-Disposition"); cd != "" {
|
||||
if idx := strings.Index(cd, "filename="); idx >= 0 {
|
||||
fileName = strings.Trim(cd[idx+len("filename="):], "\" ")
|
||||
}
|
||||
}
|
||||
}
|
||||
if fileName == "" {
|
||||
fileName = msg.FileKey
|
||||
}
|
||||
|
||||
return resp.Body, fileName, nil
|
||||
}
|
||||
|
||||
// ──────────────────────────────────────────────────────────────────────
|
||||
// Feishu CardKit v1 streaming implementation (official best practice)
|
||||
//
|
||||
|
||||
+163
-19
@@ -81,27 +81,18 @@ func (l *feishuLoggerAdapter) Error(ctx context.Context, args ...interface{}) {
|
||||
}
|
||||
|
||||
// convertEvent converts a Feishu SDK event to a unified IncomingMessage.
|
||||
// Returns nil for non-text messages.
|
||||
// Supports text and file messages. Returns nil for unsupported types.
|
||||
func convertEvent(event *larkim.P2MessageReceiveV1) *im.IncomingMessage {
|
||||
if event == nil || event.Event == nil || event.Event.Message == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
msg := event.Event.Message
|
||||
if msg.MessageType == nil || *msg.MessageType != "text" {
|
||||
if msg.MessageType == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Parse text content from JSON: {"text": "..."}
|
||||
var textContent struct {
|
||||
Text string `json:"text"`
|
||||
}
|
||||
if msg.Content == nil {
|
||||
return nil
|
||||
}
|
||||
if err := json.Unmarshal([]byte(*msg.Content), &textContent); err != nil {
|
||||
return nil
|
||||
}
|
||||
msgType := *msg.MessageType
|
||||
|
||||
// Sender info
|
||||
openID := ""
|
||||
@@ -125,7 +116,32 @@ func convertEvent(event *larkim.P2MessageReceiveV1) *im.IncomingMessage {
|
||||
messageID = *msg.MessageId
|
||||
}
|
||||
|
||||
// Strip @bot mention from group messages
|
||||
switch msgType {
|
||||
case "text":
|
||||
return convertTextEvent(msg, openID, chatID, chatType, messageID)
|
||||
case "file":
|
||||
return convertFileEvent(msg, openID, chatID, chatType, messageID)
|
||||
case "image":
|
||||
return convertImageEvent(msg, openID, chatID, chatType, messageID)
|
||||
case "post":
|
||||
return convertPostEvent(msg, openID, chatID, chatType, messageID)
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// convertTextEvent handles text message type.
|
||||
func convertTextEvent(msg *larkim.EventMessage, openID, chatID string, chatType im.ChatType, messageID string) *im.IncomingMessage {
|
||||
var textContent struct {
|
||||
Text string `json:"text"`
|
||||
}
|
||||
if msg.Content == nil {
|
||||
return nil
|
||||
}
|
||||
if err := json.Unmarshal([]byte(*msg.Content), &textContent); err != nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
content := textContent.Text
|
||||
if chatType == im.ChatTypeGroup {
|
||||
for strings.HasPrefix(content, "@_user_") {
|
||||
@@ -139,11 +155,139 @@ func convertEvent(event *larkim.P2MessageReceiveV1) *im.IncomingMessage {
|
||||
}
|
||||
|
||||
return &im.IncomingMessage{
|
||||
Platform: im.PlatformFeishu,
|
||||
UserID: openID,
|
||||
ChatID: chatID,
|
||||
ChatType: chatType,
|
||||
Content: strings.TrimSpace(content),
|
||||
MessageID: messageID,
|
||||
Platform: im.PlatformFeishu,
|
||||
MessageType: im.MessageTypeText,
|
||||
UserID: openID,
|
||||
ChatID: chatID,
|
||||
ChatType: chatType,
|
||||
Content: strings.TrimSpace(content),
|
||||
MessageID: messageID,
|
||||
}
|
||||
}
|
||||
|
||||
// convertFileEvent handles file message type.
|
||||
func convertFileEvent(msg *larkim.EventMessage, openID, chatID string, chatType im.ChatType, messageID string) *im.IncomingMessage {
|
||||
if msg.Content == nil {
|
||||
return nil
|
||||
}
|
||||
var fileContent struct {
|
||||
FileKey string `json:"file_key"`
|
||||
FileName string `json:"file_name"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(*msg.Content), &fileContent); err != nil {
|
||||
return nil
|
||||
}
|
||||
if fileContent.FileKey == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
return &im.IncomingMessage{
|
||||
Platform: im.PlatformFeishu,
|
||||
MessageType: im.MessageTypeFile,
|
||||
UserID: openID,
|
||||
ChatID: chatID,
|
||||
ChatType: chatType,
|
||||
MessageID: messageID,
|
||||
FileKey: fileContent.FileKey,
|
||||
FileName: fileContent.FileName,
|
||||
}
|
||||
}
|
||||
|
||||
// convertImageEvent handles image message type.
|
||||
// Downloads via GetMessageResource API with type=image.
|
||||
func convertImageEvent(msg *larkim.EventMessage, openID, chatID string, chatType im.ChatType, messageID string) *im.IncomingMessage {
|
||||
if msg.Content == nil {
|
||||
return nil
|
||||
}
|
||||
var imageContent struct {
|
||||
ImageKey string `json:"image_key"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(*msg.Content), &imageContent); err != nil {
|
||||
return nil
|
||||
}
|
||||
if imageContent.ImageKey == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
return &im.IncomingMessage{
|
||||
Platform: im.PlatformFeishu,
|
||||
MessageType: im.MessageTypeImage,
|
||||
UserID: openID,
|
||||
ChatID: chatID,
|
||||
ChatType: chatType,
|
||||
MessageID: messageID,
|
||||
FileKey: imageContent.ImageKey,
|
||||
FileName: imageContent.ImageKey + ".png",
|
||||
}
|
||||
}
|
||||
|
||||
// convertPostEvent handles rich-text (post) message type.
|
||||
// Extracts all plain text content and treats it as a text query for QA.
|
||||
func convertPostEvent(msg *larkim.EventMessage, openID, chatID string, chatType im.ChatType, messageID string) *im.IncomingMessage {
|
||||
if msg.Content == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Post content structure: {"title":"...", "content":[[{"tag":"text","text":"..."},{"tag":"a","href":"...","text":"..."}]]}
|
||||
var postContent struct {
|
||||
Title string `json:"title"`
|
||||
Content [][]json.RawMessage `json:"content"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(*msg.Content), &postContent); err != nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
var textParts []string
|
||||
if postContent.Title != "" {
|
||||
textParts = append(textParts, postContent.Title)
|
||||
}
|
||||
|
||||
for _, line := range postContent.Content {
|
||||
var lineText strings.Builder
|
||||
for _, elem := range line {
|
||||
var tag struct {
|
||||
Tag string `json:"tag"`
|
||||
Text string `json:"text"`
|
||||
}
|
||||
if err := json.Unmarshal(elem, &tag); err != nil {
|
||||
continue
|
||||
}
|
||||
switch tag.Tag {
|
||||
case "text", "a":
|
||||
lineText.WriteString(tag.Text)
|
||||
case "at":
|
||||
// Skip @mentions
|
||||
}
|
||||
}
|
||||
if t := strings.TrimSpace(lineText.String()); t != "" {
|
||||
textParts = append(textParts, t)
|
||||
}
|
||||
}
|
||||
|
||||
content := strings.Join(textParts, "\n")
|
||||
if chatType == im.ChatTypeGroup {
|
||||
for strings.HasPrefix(content, "@_user_") {
|
||||
idx := strings.Index(content, " ")
|
||||
if idx >= 0 {
|
||||
content = content[idx+1:]
|
||||
} else {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
content = strings.TrimSpace(content)
|
||||
if content == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
return &im.IncomingMessage{
|
||||
Platform: im.PlatformFeishu,
|
||||
MessageType: im.MessageTypeText,
|
||||
UserID: openID,
|
||||
ChatID: chatID,
|
||||
ChatType: chatType,
|
||||
Content: content,
|
||||
MessageID: messageID,
|
||||
}
|
||||
}
|
||||
|
||||
+689
-11
@@ -1,14 +1,19 @@
|
||||
package im
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"mime/multipart"
|
||||
"net/textproto"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/Tencent/WeKnora/internal/event"
|
||||
"github.com/Tencent/WeKnora/internal/logger"
|
||||
"github.com/Tencent/WeKnora/internal/models/chat"
|
||||
"github.com/Tencent/WeKnora/internal/types"
|
||||
"github.com/Tencent/WeKnora/internal/types/interfaces"
|
||||
"github.com/google/uuid"
|
||||
@@ -52,6 +57,12 @@ type Service struct {
|
||||
tenantService interfaces.TenantService
|
||||
agentService interfaces.CustomAgentService
|
||||
|
||||
// knowledgeService is used for saving IM file messages to knowledge bases.
|
||||
knowledgeService interfaces.KnowledgeService
|
||||
|
||||
// modelService is used to obtain the chat model for generating smart notification replies.
|
||||
modelService interfaces.ModelService
|
||||
|
||||
// channels maps channel ID -> running channel state
|
||||
channels map[string]*channelState
|
||||
mu sync.RWMutex
|
||||
@@ -90,6 +101,8 @@ func NewService(
|
||||
messageService interfaces.MessageService,
|
||||
tenantService interfaces.TenantService,
|
||||
agentService interfaces.CustomAgentService,
|
||||
knowledgeService interfaces.KnowledgeService,
|
||||
modelService interfaces.ModelService,
|
||||
) *Service {
|
||||
s := &Service{
|
||||
db: db,
|
||||
@@ -97,6 +110,8 @@ func NewService(
|
||||
messageService: messageService,
|
||||
tenantService: tenantService,
|
||||
agentService: agentService,
|
||||
knowledgeService: knowledgeService,
|
||||
modelService: modelService,
|
||||
channels: make(map[string]*channelState),
|
||||
adapterFactories: make(map[string]AdapterFactory),
|
||||
stopCh: make(chan struct{}),
|
||||
@@ -242,6 +257,15 @@ func (s *Service) GetChannelByID(channelID string) (*IMChannel, error) {
|
||||
return &ch, nil
|
||||
}
|
||||
|
||||
// GetChannelByIDAndTenant loads a channel from the database, scoped to a specific tenant.
|
||||
func (s *Service) GetChannelByIDAndTenant(channelID string, tenantID uint64) (*IMChannel, error) {
|
||||
var ch IMChannel
|
||||
if err := s.db.Where("id = ? AND tenant_id = ? AND deleted_at IS NULL", channelID, tenantID).First(&ch).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &ch, nil
|
||||
}
|
||||
|
||||
// HandleMessage processes an incoming IM message end-to-end using channel config.
|
||||
func (s *Service) HandleMessage(ctx context.Context, msg *IncomingMessage, channelID string) error {
|
||||
// Dedup: skip if this message was already processed (IM platforms may retry)
|
||||
@@ -280,8 +304,17 @@ func (s *Service) HandleMessage(ctx context.Context, msg *IncomingMessage, chann
|
||||
tenantID := channel.TenantID
|
||||
agentID := channel.AgentID
|
||||
|
||||
logger.Infof(ctx, "[IM] HandleMessage: channel=%s platform=%s user=%s chat=%s content_len=%d",
|
||||
channelID, msg.Platform, msg.UserID, msg.ChatID, len(msg.Content))
|
||||
logger.Infof(ctx, "[IM] HandleMessage: channel=%s platform=%s user=%s chat=%s msgtype=%s content_len=%d",
|
||||
channelID, msg.Platform, msg.UserID, msg.ChatID, msg.MessageType, len(msg.Content))
|
||||
logger.Debugf(ctx, "[IM] HandleMessage detail: msgid=%s filekey=%s filename=%s",
|
||||
msg.MessageID, msg.FileKey, msg.FileName)
|
||||
|
||||
// ── File/Image message shortcut ──
|
||||
// If the message is a file or image and the channel has a knowledge_base_id configured,
|
||||
// handle it separately without entering the QA pipeline.
|
||||
if (msg.MessageType == MessageTypeFile || msg.MessageType == MessageTypeImage) && channel.KnowledgeBaseID != "" {
|
||||
return s.handleFileMessage(ctx, msg, adapter, channel)
|
||||
}
|
||||
|
||||
// 1. Get tenant
|
||||
tenant, err := s.tenantService.GetTenantByID(ctx, tenantID)
|
||||
@@ -410,6 +443,71 @@ func (s *Service) resolveSession(ctx context.Context, msg *IncomingMessage, tena
|
||||
return &cs, nil
|
||||
}
|
||||
|
||||
// ── Agent tool call progress formatting ──────────────────────────────
|
||||
// These helpers format tool-call / tool-result events as Markdown text
|
||||
// that is injected into the streaming reply so IM users can see the
|
||||
// agent's reasoning process in real-time.
|
||||
// ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
// toolDisplayNames maps internal tool function names to user-friendly labels.
|
||||
var toolDisplayNames = map[string]string{
|
||||
"thinking": "深度思考",
|
||||
"todo_write": "制定计划",
|
||||
"knowledge_search": "知识库检索",
|
||||
"grep_chunks": "关键词搜索",
|
||||
"list_knowledge_chunks": "查看文档分块",
|
||||
"query_knowledge_graph": "查询知识图谱",
|
||||
"get_document_info": "获取文档信息",
|
||||
"database_query": "查询数据库",
|
||||
"data_analysis": "数据分析",
|
||||
"data_schema": "查看数据元信息",
|
||||
"web_search": "网络搜索",
|
||||
"web_fetch": "网页阅读",
|
||||
"read_skill": "读取技能",
|
||||
"execute_skill_script": "执行技能脚本",
|
||||
"final_answer": "生成回答",
|
||||
}
|
||||
|
||||
// internalToolNames lists tools whose execution should NOT be displayed in IM
|
||||
// messages because they are internal reasoning aids (thinking, planning) rather
|
||||
// than user-facing actions.
|
||||
var internalToolNames = map[string]bool{
|
||||
"thinking": true,
|
||||
"todo_write": true,
|
||||
}
|
||||
|
||||
// friendlyToolName returns a human-readable name for a tool.
|
||||
func friendlyToolName(toolName string) string {
|
||||
if display, ok := toolDisplayNames[toolName]; ok {
|
||||
return display
|
||||
}
|
||||
return toolName
|
||||
}
|
||||
|
||||
// isToolVisibleToUser returns true if the tool's execution progress should be
|
||||
// displayed to the IM user. Internal reasoning tools (thinking, planning) and
|
||||
// the final_answer pseudo-tool are hidden.
|
||||
func isToolVisibleToUser(toolName string) bool {
|
||||
if toolName == "final_answer" {
|
||||
return false
|
||||
}
|
||||
return !internalToolNames[toolName]
|
||||
}
|
||||
|
||||
// formatToolCallStart returns a Markdown snippet announcing that a tool is being invoked.
|
||||
func formatToolCallStart(toolName string) string {
|
||||
return fmt.Sprintf("\n\n> 🔧 正在调用工具: **%s** ...\n\n", friendlyToolName(toolName))
|
||||
}
|
||||
|
||||
// formatToolCallResult returns a Markdown snippet summarising a tool result.
|
||||
func formatToolCallResult(toolName string, success bool, durationMs int64) string {
|
||||
friendly := friendlyToolName(toolName)
|
||||
if success {
|
||||
return fmt.Sprintf("> ✅ **%s** 完成 (%.1fs)\n\n", friendly, float64(durationMs)/1000)
|
||||
}
|
||||
return fmt.Sprintf("> ⚠️ **%s** 失败 (%.1fs)\n\n", friendly, float64(durationMs)/1000)
|
||||
}
|
||||
|
||||
// handleMessageStream runs the QA pipeline and streams answer chunks to the IM platform
|
||||
// in real-time via the StreamSender interface. Chunks are batched at streamFlushInterval
|
||||
// to avoid API rate-limiting.
|
||||
@@ -435,6 +533,8 @@ func (s *Service) handleMessageStream(ctx context.Context, msg *IncomingMessage,
|
||||
done = make(chan struct{})
|
||||
closeOnce sync.Once
|
||||
inThinking bool // tracks whether we're inside a <think> block
|
||||
hasToolCalls bool // whether any tool call events have been received
|
||||
answerStarted bool // whether the final answer stream has begun
|
||||
)
|
||||
closeDone := func() { closeOnce.Do(func() { close(done) }) }
|
||||
|
||||
@@ -469,6 +569,12 @@ func (s *Service) handleMessageStream(ctx context.Context, msg *IncomingMessage,
|
||||
}
|
||||
|
||||
if content != "" {
|
||||
// Insert a separator before the first answer content when tool calls were displayed,
|
||||
// so the user can clearly distinguish the tool-call progress from the actual answer.
|
||||
if hasToolCalls && !answerStarted {
|
||||
answerStarted = true
|
||||
buf.WriteString("---\n\n")
|
||||
}
|
||||
buf.WriteString(content)
|
||||
}
|
||||
bufMu.Unlock()
|
||||
@@ -492,6 +598,42 @@ func (s *Service) handleMessageStream(ctx context.Context, msg *IncomingMessage,
|
||||
return nil
|
||||
})
|
||||
|
||||
// Subscribe to agent tool call events — show "calling tool X..." in real-time
|
||||
eventBus.On(event.EventAgentToolCall, func(_ context.Context, evt event.Event) error {
|
||||
data, ok := evt.Data.(event.AgentToolCallData)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
if !isToolVisibleToUser(data.ToolName) {
|
||||
return nil
|
||||
}
|
||||
snippet := formatToolCallStart(data.ToolName)
|
||||
bufMu.Lock()
|
||||
hasToolCalls = true
|
||||
buf.WriteString(snippet)
|
||||
bufMu.Unlock()
|
||||
logger.Debugf(ctx, "[IM] Tool call streamed to IM: tool=%s", data.ToolName)
|
||||
return nil
|
||||
})
|
||||
|
||||
// Subscribe to agent tool result events — show "tool X done" in real-time
|
||||
eventBus.On(event.EventAgentToolResult, func(_ context.Context, evt event.Event) error {
|
||||
data, ok := evt.Data.(event.AgentToolResultData)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
if !isToolVisibleToUser(data.ToolName) {
|
||||
return nil
|
||||
}
|
||||
snippet := formatToolCallResult(data.ToolName, data.Success, data.Duration)
|
||||
bufMu.Lock()
|
||||
buf.WriteString(snippet)
|
||||
bufMu.Unlock()
|
||||
logger.Debugf(ctx, "[IM] Tool result streamed to IM: tool=%s success=%v duration=%dms",
|
||||
data.ToolName, data.Success, data.Duration)
|
||||
return nil
|
||||
})
|
||||
|
||||
// Determine whether to use agent mode
|
||||
useAgent := customAgent != nil && customAgent.IsAgentMode()
|
||||
requestID := uuid.New().String()
|
||||
@@ -733,10 +875,10 @@ func (s *Service) runQA(ctx context.Context, session *types.Session, query strin
|
||||
|
||||
// ── CRUD operations for IM channels ──
|
||||
|
||||
// ListChannelsByAgent returns all channels for a given agent.
|
||||
func (s *Service) ListChannelsByAgent(agentID string) ([]IMChannel, error) {
|
||||
// ListChannelsByAgent returns all channels for a given agent within a tenant.
|
||||
func (s *Service) ListChannelsByAgent(agentID string, tenantID uint64) ([]IMChannel, error) {
|
||||
var channels []IMChannel
|
||||
if err := s.db.Where("agent_id = ? AND deleted_at IS NULL", agentID).
|
||||
if err := s.db.Where("agent_id = ? AND tenant_id = ? AND deleted_at IS NULL", agentID, tenantID).
|
||||
Order("created_at DESC").Find(&channels).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -744,7 +886,11 @@ func (s *Service) ListChannelsByAgent(agentID string) ([]IMChannel, error) {
|
||||
}
|
||||
|
||||
// CreateChannel creates a new IM channel and optionally starts it.
|
||||
// Returns a duplicate_bot error if the bot identity is already used by another channel.
|
||||
func (s *Service) CreateChannel(channel *IMChannel) error {
|
||||
if err := s.checkDuplicateBot(channel, ""); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := s.db.Create(channel).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -757,7 +903,11 @@ func (s *Service) CreateChannel(channel *IMChannel) error {
|
||||
}
|
||||
|
||||
// UpdateChannel updates a channel and restarts it if needed.
|
||||
// Returns a duplicate_bot error if the bot identity is already used by another channel.
|
||||
func (s *Service) UpdateChannel(channel *IMChannel) error {
|
||||
if err := s.checkDuplicateBot(channel, channel.ID); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := s.db.Save(channel).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -771,16 +921,23 @@ func (s *Service) UpdateChannel(channel *IMChannel) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// DeleteChannel soft-deletes a channel and stops it.
|
||||
func (s *Service) DeleteChannel(channelID string) error {
|
||||
// DeleteChannel soft-deletes a channel and stops it. Only deletes if the channel belongs to the given tenant.
|
||||
func (s *Service) DeleteChannel(channelID string, tenantID uint64) error {
|
||||
s.StopChannel(channelID)
|
||||
return s.db.Where("id = ?", channelID).Delete(&IMChannel{}).Error
|
||||
result := s.db.Where("id = ? AND tenant_id = ?", channelID, tenantID).Delete(&IMChannel{})
|
||||
if result.Error != nil {
|
||||
return result.Error
|
||||
}
|
||||
if result.RowsAffected == 0 {
|
||||
return fmt.Errorf("channel not found")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ToggleChannel enables or disables a channel.
|
||||
func (s *Service) ToggleChannel(channelID string) (*IMChannel, error) {
|
||||
// ToggleChannel enables or disables a channel. Only toggles if the channel belongs to the given tenant.
|
||||
func (s *Service) ToggleChannel(channelID string, tenantID uint64) (*IMChannel, error) {
|
||||
var ch IMChannel
|
||||
if err := s.db.Where("id = ? AND deleted_at IS NULL", channelID).First(&ch).Error; err != nil {
|
||||
if err := s.db.Where("id = ? AND tenant_id = ? AND deleted_at IS NULL", channelID, tenantID).First(&ch).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ch.Enabled = !ch.Enabled
|
||||
@@ -796,3 +953,524 @@ func (s *Service) ToggleChannel(channelID string) (*IMChannel, error) {
|
||||
}
|
||||
return &ch, nil
|
||||
}
|
||||
|
||||
// checkDuplicateBot queries the bot_identity index to see if another active channel
|
||||
// already uses the same bot. This is an O(1) index lookup, not a full table scan.
|
||||
// The DB unique index on bot_identity serves as an additional safety net.
|
||||
// excludeID is the channel's own ID (for updates); pass "" for new channels.
|
||||
func (s *Service) checkDuplicateBot(channel *IMChannel, excludeID string) error {
|
||||
// Compute bot_identity the same way the BeforeSave hook will
|
||||
botKey := channel.computeBotIdentity()
|
||||
if botKey == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
var existing IMChannel
|
||||
query := s.db.Where("bot_identity = ? AND deleted_at IS NULL", botKey)
|
||||
if excludeID != "" {
|
||||
query = query.Where("id != ?", excludeID)
|
||||
}
|
||||
if err := query.First(&existing).Error; err != nil {
|
||||
if err == gorm.ErrRecordNotFound {
|
||||
return nil // no conflict
|
||||
}
|
||||
return fmt.Errorf("check duplicate bot: %w", err)
|
||||
}
|
||||
return fmt.Errorf("duplicate_bot: this bot is already bound to channel %q (%s); each bot can only be connected to one channel", existing.Name, existing.ID)
|
||||
}
|
||||
|
||||
// ── File message handling ──────────────────────────────────────────────
|
||||
// These methods handle file messages received via IM platforms.
|
||||
// Files are downloaded from the IM platform, validated, and saved to the
|
||||
// configured knowledge base asynchronously. The user receives a notification
|
||||
// at the start and end of processing.
|
||||
// ────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// supportedKBFileExts is the set of file extensions that can be saved to a knowledge base.
|
||||
var supportedKBFileExts = map[string]bool{
|
||||
"pdf": true, "txt": true, "docx": true, "doc": true,
|
||||
"md": true, "markdown": true,
|
||||
"png": true, "jpg": true, "jpeg": true, "gif": true,
|
||||
"csv": true, "xlsx": true, "xls": true,
|
||||
"pptx": true, "ppt": true,
|
||||
}
|
||||
|
||||
// handleFileMessage processes a file message by downloading it from the IM platform
|
||||
// and saving it to the channel's configured knowledge base. Sends start/end
|
||||
// notifications to the user via the adapter.
|
||||
func (s *Service) handleFileMessage(ctx context.Context, msg *IncomingMessage, adapter Adapter, channel *IMChannel) error {
|
||||
// Check if the adapter supports file downloading
|
||||
downloader, ok := adapter.(FileDownloader)
|
||||
if !ok {
|
||||
logger.Infof(ctx, "[IM] Adapter for platform %s does not support file download, ignoring file message", msg.Platform)
|
||||
return s.sendSmartReply(ctx, adapter, msg, channel,
|
||||
"用户尝试发送文件，但当前平台暂不支持文件消息处理。",
|
||||
"❌ 当前平台暂不支持文件消息处理。")
|
||||
}
|
||||
|
||||
// For image messages, ensure a proper file extension is present.
|
||||
// IM platforms may only provide a hash/key as filename without extension.
|
||||
if msg.MessageType == MessageTypeImage && fileExtension(msg.FileName) == "" {
|
||||
msg.FileName = msg.FileName + ".png"
|
||||
}
|
||||
|
||||
// Validate file extension (pre-download).
|
||||
// Some platforms (e.g. WeCom aibot) do not provide original filenames in the
|
||||
// callback JSON — only a hash ID. For such cases we defer extension validation
|
||||
// to after the file is downloaded, where the real name may be obtained from
|
||||
// HTTP Content-Disposition or Content-Type headers.
|
||||
ext := fileExtension(msg.FileName)
|
||||
if ext != "" && !supportedKBFileExts[ext] {
|
||||
logger.Infof(ctx, "[IM] Unsupported file type: %s (file=%s)", ext, msg.FileName)
|
||||
return s.sendSmartReply(ctx, adapter, msg, channel,
|
||||
fmt.Sprintf("用户上传了一个不支持的文件类型「%s」。目前支持的类型包括：PDF、Word、TXT、Markdown、Excel、CSV、PPT、图片。", ext),
|
||||
fmt.Sprintf("❌ 不支持的文件类型「%s」。\n\n支持的类型：PDF、Word、TXT、Markdown、Excel、CSV、PPT、图片。", ext))
|
||||
}
|
||||
|
||||
displayName := msg.FileName
|
||||
if ext == "" {
|
||||
displayName = "文件"
|
||||
}
|
||||
|
||||
// Send "processing started" notification (streaming)
|
||||
if err := s.sendSmartReply(ctx, adapter, msg, channel,
|
||||
fmt.Sprintf("用户发送了一个文件「%s」，系统正在处理并保存到知识库中，需要告知用户请稍候。", displayName),
|
||||
fmt.Sprintf("📥 已收到%s，正在处理并保存到知识库，请稍候...", displayName)); err != nil {
|
||||
logger.Warnf(ctx, "[IM] Failed to send file processing start notification: %v", err)
|
||||
}
|
||||
|
||||
// Process asynchronously to avoid blocking the message handler
|
||||
go s.processFileToKnowledgeBase(context.WithoutCancel(ctx), msg, downloader, adapter, channel)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// processFileToKnowledgeBase is the async worker that downloads a file from the
|
||||
// IM platform and creates a knowledge entry in the configured knowledge base.
|
||||
func (s *Service) processFileToKnowledgeBase(ctx context.Context, msg *IncomingMessage, downloader FileDownloader, adapter Adapter, channel *IMChannel) {
|
||||
kbID := channel.KnowledgeBaseID
|
||||
tenantID := channel.TenantID
|
||||
|
||||
// Build context with tenant info for the knowledge service
|
||||
tenant, err := s.tenantService.GetTenantByID(ctx, tenantID)
|
||||
if err != nil {
|
||||
logger.Errorf(ctx, "[IM] Failed to get tenant %d for file processing: %v", tenantID, err)
|
||||
s.sendFileResult(ctx, adapter, msg, msg.FileName, false, "获取租户信息失败", channel)
|
||||
return
|
||||
}
|
||||
kbCtx := context.WithValue(ctx, types.TenantIDContextKey, tenantID)
|
||||
kbCtx = context.WithValue(kbCtx, types.TenantInfoContextKey, tenant)
|
||||
|
||||
// Download file from IM platform
|
||||
reader, fileName, err := downloader.DownloadFile(ctx, msg)
|
||||
if err != nil {
|
||||
logger.Errorf(ctx, "[IM] Failed to download file from %s: %v", msg.Platform, err)
|
||||
s.sendFileResult(ctx, adapter, msg, msg.FileName, false, "下载文件失败", channel)
|
||||
return
|
||||
}
|
||||
defer reader.Close()
|
||||
|
||||
logger.Debugf(ctx, "[IM] Downloaded file: original_name=%s resolved_name=%s", msg.FileName, fileName)
|
||||
|
||||
// Post-download extension validation: if the pre-download name had no extension
|
||||
// (e.g. WeCom file messages only provide a hash), check the resolved name now.
|
||||
ext := fileExtension(fileName)
|
||||
if !supportedKBFileExts[ext] {
|
||||
logger.Infof(ctx, "[IM] Unsupported file type after download: %s (file=%s)", ext, fileName)
|
||||
s.sendFileResult(ctx, adapter, msg, fileName, false,
|
||||
fmt.Sprintf("不支持的文件类型「%s」。支持：PDF、Word、TXT、Markdown、Excel、CSV、PPT、图片", ext), channel)
|
||||
return
|
||||
}
|
||||
|
||||
// Read file content into memory for multipart upload
|
||||
content, err := io.ReadAll(reader)
|
||||
if err != nil {
|
||||
logger.Errorf(ctx, "[IM] Failed to read file content: %v", err)
|
||||
s.sendFileResult(ctx, adapter, msg, fileName, false, "读取文件内容失败", channel)
|
||||
return
|
||||
}
|
||||
|
||||
// Create a multipart.FileHeader compatible wrapper
|
||||
fh := newInMemoryFileHeader(fileName, content)
|
||||
|
||||
// Create knowledge entry via the knowledge service
|
||||
knowledge, err := s.knowledgeService.CreateKnowledgeFromFile(kbCtx, kbID, fh, nil, nil, "", "")
|
||||
if err != nil {
|
||||
errMsg := err.Error()
|
||||
// Check for duplicate file
|
||||
if strings.Contains(errMsg, "duplicate") || strings.Contains(errMsg, "already exists") {
|
||||
logger.Infof(ctx, "[IM] File already exists in knowledge base: %s", fileName)
|
||||
s.sendFileResult(ctx, adapter, msg, fileName, false, "文件已存在于知识库中", channel)
|
||||
return
|
||||
}
|
||||
logger.Errorf(ctx, "[IM] Failed to create knowledge from file: %v", err)
|
||||
s.sendFileResult(ctx, adapter, msg, fileName, false, "保存到知识库失败", channel)
|
||||
return
|
||||
}
|
||||
|
||||
logger.Infof(ctx, "[IM] File saved to knowledge base: kb=%s knowledge=%s file=%s", kbID, knowledge.ID, fileName)
|
||||
s.sendFileResult(ctx, adapter, msg, fileName, true, "", channel)
|
||||
|
||||
// Start a background watcher to send the document summary once Asynq
|
||||
// finishes parsing + summary generation. This is intentionally decoupled
|
||||
// from the Asynq task pipeline to avoid modifying any existing logic.
|
||||
go s.watchAndSendSummary(ctx, kbCtx, adapter, msg, knowledge.ID, fileName, channel)
|
||||
}
|
||||
|
||||
// sendFileResult sends a notification about the file processing result.
|
||||
// It uses sendSmartReply to generate a friendly, streaming reply via the channel's LLM.
|
||||
// Falls back to a static template if the LLM is unavailable.
|
||||
func (s *Service) sendFileResult(ctx context.Context, adapter Adapter, msg *IncomingMessage, fileName string, success bool, errDetail string, channel *IMChannel) {
|
||||
var fallback string
|
||||
if success {
|
||||
fallback = fmt.Sprintf("✅ 文件「%s」已保存到知识库，正在解析中，完成后会通知你～", fileName)
|
||||
} else {
|
||||
fallback = fmt.Sprintf("❌ 文件「%s」处理失败：%s", fileName, errDetail)
|
||||
}
|
||||
|
||||
var situation string
|
||||
if success {
|
||||
situation = fmt.Sprintf("用户上传的文件「%s」已成功保存到知识库，但还需要后台解析文档内容（这需要一些时间）。请告知用户文件已收到，正在解析处理中，解析完成后会自动推送结果。", fileName)
|
||||
} else {
|
||||
situation = fmt.Sprintf("用户上传的文件「%s」处理失败，原因：%s。", fileName, errDetail)
|
||||
}
|
||||
|
||||
if err := s.sendSmartReply(ctx, adapter, msg, channel, situation, fallback); err != nil {
|
||||
logger.Warnf(ctx, "[IM] Failed to send file result notification: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// smartReplySystemPrompt is the system prompt used for generating smart notification replies.
|
||||
const smartReplySystemPrompt = "你是一个专业的 IM 机器人助手。请根据以下事件情况，生成一条简洁、清晰的通知消息。" +
|
||||
"要求：1) 可适当使用 emoji 但不要过多；2) 语气专业平等，像同事之间对话，不要谄媚讨好，不要用「啦」「哦」「呢」「哟」等撒娇语气词；" +
|
||||
"3) 直接输出消息内容，不要加任何额外解释；" +
|
||||
"4) 如果事件中包含摘要或详细内容，请用 Markdown 格式结构化展示（使用标题、列表、加粗等），完整呈现，不要删减或概括；如果是简单通知，则控制在 2-3 句话以内。"
|
||||
|
||||
// sendSmartReply generates a notification message using the channel's LLM and sends it
|
||||
// to the user. If the adapter supports streaming (StreamSender), it streams the reply
|
||||
// in real-time for a better user experience. Otherwise, it falls back to non-streaming.
|
||||
// If the LLM is unavailable or fails, it sends the provided fallback text.
|
||||
func (s *Service) sendSmartReply(ctx context.Context, adapter Adapter, msg *IncomingMessage, channel *IMChannel, situation string, fallback string) error {
|
||||
chatModel := s.getChatModelForChannel(ctx, channel)
|
||||
if chatModel == nil {
|
||||
return adapter.SendReply(ctx, msg, &ReplyMessage{Content: fallback, IsFinal: true})
|
||||
}
|
||||
|
||||
// If the adapter supports streaming, use stream mode
|
||||
if streamer, ok := adapter.(StreamSender); ok {
|
||||
if err := s.streamSmartReply(ctx, chatModel, streamer, msg, situation); err == nil {
|
||||
return nil
|
||||
}
|
||||
// Stream failed — fall through to non-streaming
|
||||
logger.Warnf(ctx, "[IM] Stream smart reply failed, falling back to non-streaming")
|
||||
}
|
||||
|
||||
// Non-streaming fallback
|
||||
content := s.generateSmartReply(ctx, chatModel, situation, fallback)
|
||||
return adapter.SendReply(ctx, msg, &ReplyMessage{Content: content, IsFinal: true})
|
||||
}
|
||||
|
||||
// streamSmartReply uses ChatStream to generate and stream a notification reply in real-time.
|
||||
func (s *Service) streamSmartReply(ctx context.Context, chatModel chat.Chat, streamer StreamSender, msg *IncomingMessage, situation string) error {
|
||||
messages := []chat.Message{
|
||||
{Role: "system", Content: smartReplySystemPrompt},
|
||||
{Role: "user", Content: situation},
|
||||
}
|
||||
|
||||
timeoutCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
|
||||
defer cancel()
|
||||
|
||||
streamCh, err := chatModel.ChatStream(timeoutCtx, messages, &chat.ChatOptions{
|
||||
Temperature: 0.7,
|
||||
MaxTokens: 800,
|
||||
})
|
||||
if err != nil {
|
||||
logger.Warnf(ctx, "[IM] ChatStream failed for smart reply: %v", err)
|
||||
return err
|
||||
}
|
||||
|
||||
// Start the stream on the IM platform
|
||||
streamID, err := streamer.StartStream(ctx, msg)
|
||||
if err != nil {
|
||||
logger.Warnf(ctx, "[IM] StartStream failed for smart reply: %v", err)
|
||||
return err
|
||||
}
|
||||
|
||||
// Flush loop with batching (same pattern as handleMessageStream)
|
||||
var (
|
||||
bufMu sync.Mutex
|
||||
buf strings.Builder
|
||||
done = make(chan struct{})
|
||||
)
|
||||
|
||||
go func() {
|
||||
defer close(done)
|
||||
for resp := range streamCh {
|
||||
if resp.Content != "" {
|
||||
bufMu.Lock()
|
||||
buf.WriteString(resp.Content)
|
||||
bufMu.Unlock()
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
ticker := time.NewTicker(streamFlushInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
flush := func() {
|
||||
bufMu.Lock()
|
||||
chunk := buf.String()
|
||||
buf.Reset()
|
||||
bufMu.Unlock()
|
||||
|
||||
if chunk != "" {
|
||||
if err := streamer.SendStreamChunk(ctx, msg, streamID, chunk); err != nil {
|
||||
logger.Warnf(ctx, "[IM] SendStreamChunk failed for smart reply: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
loop:
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
flush()
|
||||
case <-done:
|
||||
break loop
|
||||
case <-timeoutCtx.Done():
|
||||
break loop
|
||||
}
|
||||
}
|
||||
|
||||
// Final flush
|
||||
flush()
|
||||
|
||||
// End the stream
|
||||
if err := streamer.EndStream(ctx, msg, streamID); err != nil {
|
||||
logger.Warnf(ctx, "[IM] EndStream failed for smart reply: %v", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// generateSmartReply uses the channel's agent LLM to produce a natural-language
|
||||
// notification message for the given situation (non-streaming).
|
||||
// If the call fails, it returns the provided fallback text.
|
||||
func (s *Service) generateSmartReply(ctx context.Context, chatModel chat.Chat, situation string, fallback string) string {
|
||||
messages := []chat.Message{
|
||||
{Role: "system", Content: smartReplySystemPrompt},
|
||||
{Role: "user", Content: situation},
|
||||
}
|
||||
|
||||
timeoutCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
resp, err := chatModel.Chat(timeoutCtx, messages, &chat.ChatOptions{
|
||||
Temperature: 0.7,
|
||||
MaxTokens: 800,
|
||||
})
|
||||
if err != nil {
|
||||
logger.Warnf(ctx, "[IM] Smart reply generation failed, using fallback: %v", err)
|
||||
return fallback
|
||||
}
|
||||
|
||||
reply := strings.TrimSpace(resp.Content)
|
||||
if reply == "" {
|
||||
return fallback
|
||||
}
|
||||
return reply
|
||||
}
|
||||
|
||||
// getChatModelForChannel resolves the chat.Chat instance configured on the
|
||||
// channel's agent. Returns nil if the model cannot be resolved.
|
||||
func (s *Service) getChatModelForChannel(ctx context.Context, channel *IMChannel) chat.Chat {
|
||||
if channel == nil || channel.AgentID == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Ensure the context carries tenant ID — some call sites (e.g. handleFileMessage)
|
||||
// may invoke this before the tenant has been injected into ctx.
|
||||
if _, ok := types.TenantIDFromContext(ctx); !ok && channel.TenantID != 0 {
|
||||
ctx = context.WithValue(ctx, types.TenantIDContextKey, channel.TenantID)
|
||||
}
|
||||
|
||||
agent, err := s.agentService.GetAgentByID(ctx, channel.AgentID)
|
||||
if err != nil || agent == nil {
|
||||
logger.Debugf(ctx, "[IM] Cannot get agent %s for smart reply: %v", channel.AgentID, err)
|
||||
return nil
|
||||
}
|
||||
|
||||
modelID := agent.Config.ModelID
|
||||
if modelID == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
chatModel, err := s.modelService.GetChatModel(ctx, modelID)
|
||||
if err != nil {
|
||||
logger.Debugf(ctx, "[IM] Cannot get chat model %s for smart reply: %v", modelID, err)
|
||||
return nil
|
||||
}
|
||||
return chatModel
|
||||
}
|
||||
|
||||
// watchAndSendSummary polls the knowledge record until document parsing (and
|
||||
// optionally summary generation) completes, then sends the result back to the
|
||||
// IM user. This runs as a fire-and-forget goroutine, completely decoupled from
|
||||
// the Asynq worker pipeline.
|
||||
func (s *Service) watchAndSendSummary(
|
||||
ctx context.Context,
|
||||
kbCtx context.Context,
|
||||
adapter Adapter,
|
||||
msg *IncomingMessage,
|
||||
knowledgeID string,
|
||||
fileName string,
|
||||
channel *IMChannel,
|
||||
) {
|
||||
const (
|
||||
pollInterval = 5 * time.Second
|
||||
maxWait = 10 * time.Minute // give up after 10 minutes
|
||||
)
|
||||
|
||||
deadline := time.Now().Add(maxWait)
|
||||
ticker := time.NewTicker(pollInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
if time.Now().After(deadline) {
|
||||
logger.Infof(ctx, "[IM] Summary watcher timed out for knowledge %s", knowledgeID)
|
||||
return
|
||||
}
|
||||
|
||||
knowledge, err := s.knowledgeService.GetKnowledgeByID(kbCtx, knowledgeID)
|
||||
if err != nil {
|
||||
logger.Warnf(ctx, "[IM] Summary watcher: failed to get knowledge %s: %v", knowledgeID, err)
|
||||
return
|
||||
}
|
||||
|
||||
switch knowledge.ParseStatus {
|
||||
case types.ParseStatusFailed:
|
||||
// Parsing failed — notify user and stop watching
|
||||
errMsg := knowledge.ErrorMessage
|
||||
if errMsg == "" {
|
||||
errMsg = "文档解析失败"
|
||||
}
|
||||
_ = s.sendSmartReply(ctx, adapter, msg, channel,
|
||||
fmt.Sprintf("用户之前上传的文件「%s」解析失败了，错误原因：%s。请安慰用户并建议重试。", fileName, errMsg),
|
||||
fmt.Sprintf("⚠️ 文件「%s」解析失败：%s", fileName, errMsg))
|
||||
return
|
||||
|
||||
case types.ParseStatusCompleted:
|
||||
// Parsing done. If summary generation is in progress, wait for it.
|
||||
switch knowledge.SummaryStatus {
|
||||
case types.SummaryStatusNone, "":
|
||||
// No summary task configured. For image files the VLM caption
|
||||
// is stored in Description by finalizeImageKnowledge, so we
|
||||
// still show it if present.
|
||||
if knowledge.Description != "" && knowledge.Description != fileName {
|
||||
_ = s.sendSmartReply(ctx, adapter, msg, channel,
|
||||
fmt.Sprintf("用户之前上传的文件「%s」已解析完成。以下是文件的完整摘要内容：\n%s\n\n请生成一条通知消息，包含：1) 告知文件已解析完成；2) 用 Markdown 格式（标题、列表、加粗等）结构化展示上述摘要内容，不要删减或概括；3) 提示用户可以针对该文件提问。", fileName, knowledge.Description),
|
||||
fmt.Sprintf("📄 文件「%s」已解析完成。\n\n**摘要：**\n\n%s\n\n---\n可以针对该文件进行提问。", fileName, knowledge.Description))
|
||||
} else {
|
||||
_ = s.sendSmartReply(ctx, adapter, msg, channel,
|
||||
fmt.Sprintf("用户之前上传的文件「%s」已解析完成，现在可以开始针对该文件进行提问了。", fileName),
|
||||
fmt.Sprintf("📄 文件「%s」已解析完成，可以开始提问了！", fileName))
|
||||
}
|
||||
return
|
||||
|
||||
case types.SummaryStatusCompleted:
|
||||
// Summary is ready — send it
|
||||
s.sendSummaryNotification(ctx, adapter, msg, knowledge, fileName, channel)
|
||||
return
|
||||
|
||||
case types.SummaryStatusFailed:
|
||||
_ = s.sendSmartReply(ctx, adapter, msg, channel,
|
||||
fmt.Sprintf("用户之前上传的文件「%s」已解析完成，但摘要生成失败了。不过文件已可用于提问。", fileName),
|
||||
fmt.Sprintf("📄 文件「%s」已解析完成，可以开始提问了！（摘要生成失败）", fileName))
|
||||
return
|
||||
|
||||
default:
|
||||
// Still generating summary — keep polling
|
||||
}
|
||||
|
||||
default:
|
||||
// Still parsing — keep polling
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// sendSummaryNotification retrieves the summary chunk for a knowledge entry
|
||||
// and sends it as a message to the IM user.
|
||||
func (s *Service) sendSummaryNotification(
|
||||
ctx context.Context,
|
||||
adapter Adapter,
|
||||
msg *IncomingMessage,
|
||||
knowledge *types.Knowledge,
|
||||
fileName string,
|
||||
channel *IMChannel,
|
||||
) {
|
||||
// The summary is stored in the knowledge's Description field or as a
|
||||
// ChunkTypeSummary chunk. We use Description first (populated by the
|
||||
// summary generation task), falling back to a generic notice.
|
||||
summary := knowledge.Description
|
||||
if summary == "" {
|
||||
summary = knowledge.Title
|
||||
}
|
||||
|
||||
var situation, fallback string
|
||||
if summary != "" && summary != fileName {
|
||||
situation = fmt.Sprintf("用户之前上传的文件「%s」已解析完成。以下是文件的完整摘要内容：\n%s\n\n请生成一条通知消息，包含：1) 告知文件已解析完成；2) 用 Markdown 格式（标题、列表、加粗等）结构化展示上述摘要内容，不要删减或概括；3) 提示用户可以针对该文件提问。", fileName, summary)
|
||||
fallback = fmt.Sprintf("📄 文件「%s」已解析完成。\n\n**摘要：**\n\n%s\n\n---\n可以针对该文件进行提问。", fileName, summary)
|
||||
} else {
|
||||
situation = fmt.Sprintf("用户之前上传的文件「%s」已解析完成，现在可以开始针对该文件进行提问了。", fileName)
|
||||
fallback = fmt.Sprintf("📄 文件「%s」已解析完成，可以开始提问了！", fileName)
|
||||
}
|
||||
|
||||
if err := s.sendSmartReply(ctx, adapter, msg, channel, situation, fallback); err != nil {
|
||||
logger.Warnf(ctx, "[IM] Failed to send summary notification: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// fileExtension extracts the lowercase file extension from a filename.
|
||||
func fileExtension(filename string) string {
|
||||
parts := strings.Split(filename, ".")
|
||||
if len(parts) < 2 {
|
||||
return ""
|
||||
}
|
||||
return strings.ToLower(parts[len(parts)-1])
|
||||
}
|
||||
|
||||
// newInMemoryFileHeader wraps in-memory file content as a *multipart.FileHeader
|
||||
// so it can be passed to CreateKnowledgeFromFile which expects a multipart upload.
|
||||
func newInMemoryFileHeader(filename string, data []byte) *multipart.FileHeader {
|
||||
body := &bytes.Buffer{}
|
||||
writer := multipart.NewWriter(body)
|
||||
|
||||
h := make(textproto.MIMEHeader)
|
||||
h.Set("Content-Disposition", fmt.Sprintf(`form-data; name="file"; filename="%s"`, filename))
|
||||
h.Set("Content-Type", "application/octet-stream")
|
||||
|
||||
part, err := writer.CreatePart(h)
|
||||
if err != nil {
|
||||
// Fallback: return a minimal FileHeader
|
||||
return &multipart.FileHeader{Filename: filename, Size: int64(len(data))}
|
||||
}
|
||||
_, _ = part.Write(data)
|
||||
_ = writer.Close()
|
||||
|
||||
// Parse the multipart body to extract the FileHeader
|
||||
reader := multipart.NewReader(body, writer.Boundary())
|
||||
form, err := reader.ReadForm(int64(len(data)) + 1024)
|
||||
if err != nil || form == nil {
|
||||
return &multipart.FileHeader{Filename: filename, Size: int64(len(data))}
|
||||
}
|
||||
files := form.File["file"]
|
||||
if len(files) == 0 {
|
||||
return &multipart.FileHeader{Filename: filename, Size: int64(len(data))}
|
||||
}
|
||||
return files[0]
|
||||
}
|
||||
|
||||
+55
-2
@@ -1,6 +1,8 @@
|
||||
package im
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/Tencent/WeKnora/internal/types"
|
||||
@@ -18,8 +20,10 @@ type IMChannel struct {
|
||||
Name string `json:"name" gorm:"type:varchar(255);not null;default:''"`
|
||||
Enabled bool `json:"enabled" gorm:"not null;default:true"`
|
||||
Mode string `json:"mode" gorm:"type:varchar(20);not null;default:'websocket'"`
|
||||
OutputMode string `json:"output_mode" gorm:"type:varchar(20);not null;default:'stream'"`
|
||||
Credentials types.JSON `json:"credentials" gorm:"type:jsonb;not null;default:'{}'"`
|
||||
OutputMode string `json:"output_mode" gorm:"type:varchar(20);not null;default:'stream'"`
|
||||
KnowledgeBaseID string `json:"knowledge_base_id" gorm:"type:varchar(36);default:''"`
|
||||
BotIdentity string `json:"bot_identity" gorm:"type:varchar(255);not null;default:'';uniqueIndex:idx_im_channels_bot_identity,where:deleted_at IS NULL AND bot_identity != ''"`
|
||||
Credentials types.JSON `json:"credentials" gorm:"type:jsonb;not null;default:'{}'"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
DeletedAt gorm.DeletedAt `json:"deleted_at" gorm:"index"`
|
||||
@@ -39,9 +43,58 @@ func (ch *IMChannel) BeforeCreate(tx *gorm.DB) error {
|
||||
if ch.OutputMode == "" {
|
||||
ch.OutputMode = "stream"
|
||||
}
|
||||
ch.BotIdentity = ch.computeBotIdentity()
|
||||
return nil
|
||||
}
|
||||
|
||||
// BeforeSave ensures bot_identity is recomputed on every save (create + update).
|
||||
func (ch *IMChannel) BeforeSave(tx *gorm.DB) error {
|
||||
ch.BotIdentity = ch.computeBotIdentity()
|
||||
return nil
|
||||
}
|
||||
|
||||
// computeBotIdentity derives a unique bot identity string from the channel's
|
||||
// platform, mode, and credentials. Returns "" if no identity can be extracted.
|
||||
func (ch *IMChannel) computeBotIdentity() string {
|
||||
creds := make(map[string]interface{})
|
||||
if err := json.Unmarshal([]byte(ch.Credentials), &creds); err != nil {
|
||||
return ""
|
||||
}
|
||||
|
||||
str := func(key string) string {
|
||||
if v, ok := creds[key]; ok {
|
||||
switch val := v.(type) {
|
||||
case string:
|
||||
return val
|
||||
case float64:
|
||||
return fmt.Sprintf("%.0f", val)
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
switch ch.Platform {
|
||||
case "wecom":
|
||||
switch ch.Mode {
|
||||
case "websocket":
|
||||
if botID := str("bot_id"); botID != "" {
|
||||
return "wecom:ws:" + botID
|
||||
}
|
||||
case "webhook":
|
||||
corpID := str("corp_id")
|
||||
agentID := str("corp_agent_id")
|
||||
if corpID != "" && agentID != "" {
|
||||
return "wecom:wh:" + corpID + ":" + agentID
|
||||
}
|
||||
}
|
||||
case "feishu":
|
||||
if appID := str("app_id"); appID != "" {
|
||||
return "feishu:" + appID
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// ChannelSession maps an IM channel (user+chat combination) to a WeKnora session.
|
||||
// This allows the IM integration to maintain conversation continuity.
|
||||
type ChannelSession struct {
|
||||
|
||||
+176
-15
@@ -51,12 +51,14 @@ type wsFrame struct {
|
||||
}
|
||||
|
||||
// botMessage is the body of an aibot_msg_callback frame.
|
||||
// Supports text, image, file, voice, and mixed message types.
|
||||
// Reference: https://developer.work.weixin.qq.com/document/path/100719
|
||||
type botMessage struct {
|
||||
MsgID string `json:"msgid"`
|
||||
AiBotID string `json:"aibotid"`
|
||||
ChatID string `json:"chatid"`
|
||||
ChatType string `json:"chattype"` // "single" or "group"
|
||||
MsgType string `json:"msgtype"` // "text", "image", "event", ...
|
||||
MsgType string `json:"msgtype"` // "text", "image", "file", "voice", "video", "mixed", "stream"
|
||||
CreateTime int64 `json:"create_time"`
|
||||
From struct {
|
||||
UserID string `json:"userid"`
|
||||
@@ -64,6 +66,37 @@ type botMessage struct {
|
||||
Text struct {
|
||||
Content string `json:"content"`
|
||||
} `json:"text"`
|
||||
Image struct {
|
||||
URL string `json:"url"` // encrypted download URL, valid for 5 minutes
|
||||
AESKey string `json:"aeskey"` // per-message AES key for decrypting downloaded content
|
||||
} `json:"image"`
|
||||
File struct {
|
||||
URL string `json:"url"` // encrypted download URL, valid for 5 minutes
|
||||
AESKey string `json:"aeskey"` // per-message AES key for decrypting downloaded content
|
||||
} `json:"file"`
|
||||
Voice struct {
|
||||
Content string `json:"content"` // speech-to-text result
|
||||
} `json:"voice"`
|
||||
Video struct {
|
||||
URL string `json:"url"` // encrypted download URL, valid for 5 minutes
|
||||
AESKey string `json:"aeskey"` // per-message AES key for decrypting downloaded content
|
||||
} `json:"video"`
|
||||
Mixed struct {
|
||||
MsgItem []botMixedItem `json:"msg_item"`
|
||||
} `json:"mixed"`
|
||||
Quote *botMessage `json:"quote,omitempty"` // quoted message (optional)
|
||||
}
|
||||
|
||||
// botMixedItem is one element in a mixed (text+image) message.
|
||||
type botMixedItem struct {
|
||||
MsgType string `json:"msgtype"` // "text" or "image"
|
||||
Text struct {
|
||||
Content string `json:"content"`
|
||||
} `json:"text"`
|
||||
Image struct {
|
||||
URL string `json:"url"`
|
||||
AESKey string `json:"aeskey"`
|
||||
} `json:"image"`
|
||||
}
|
||||
|
||||
// streamReplyBody is the body for a streaming text reply.
|
||||
@@ -380,17 +413,18 @@ func (c *LongConnClient) heartbeatLoop(ctx context.Context) {
|
||||
}
|
||||
|
||||
func (c *LongConnClient) handleCallback(ctx context.Context, frame wsFrame) {
|
||||
// Log raw message body for debugging
|
||||
logger.Debugf(ctx, "[WeCom] Raw callback body: %s", string(frame.Body))
|
||||
|
||||
var msg botMessage
|
||||
if err := json.Unmarshal(frame.Body, &msg); err != nil {
|
||||
logger.Warnf(ctx, "[WeCom] Failed to unmarshal callback body: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
// Only handle text messages for now
|
||||
if msg.MsgType != "text" {
|
||||
logger.Infof(ctx, "[WeCom] Ignoring non-text message type: %s", msg.MsgType)
|
||||
return
|
||||
}
|
||||
logger.Debugf(ctx, "[WeCom] Parsed message: msgid=%s msgtype=%s from=%s chattype=%s text=%q image_url=%q file_url=%q voice=%q mixed_items=%d",
|
||||
msg.MsgID, msg.MsgType, msg.From.UserID, msg.ChatType,
|
||||
msg.Text.Content, msg.Image.URL, msg.File.URL, msg.Voice.Content, len(msg.Mixed.MsgItem))
|
||||
|
||||
chatType := im.ChatTypeDirect
|
||||
chatID := ""
|
||||
@@ -405,15 +439,87 @@ func (c *LongConnClient) handleCallback(ctx context.Context, frame wsFrame) {
|
||||
reqID = frame.Headers["req_id"]
|
||||
}
|
||||
|
||||
incoming := &im.IncomingMessage{
|
||||
Platform: im.PlatformWeCom,
|
||||
UserID: msg.From.UserID,
|
||||
UserName: msg.From.UserID,
|
||||
ChatID: chatID,
|
||||
ChatType: chatType,
|
||||
Content: strings.TrimSpace(msg.Text.Content),
|
||||
MessageID: msg.MsgID,
|
||||
Extra: map[string]string{"req_id": reqID},
|
||||
var incoming *im.IncomingMessage
|
||||
|
||||
switch msg.MsgType {
|
||||
case "text":
|
||||
incoming = &im.IncomingMessage{
|
||||
Platform: im.PlatformWeCom,
|
||||
MessageType: im.MessageTypeText,
|
||||
UserID: msg.From.UserID,
|
||||
UserName: msg.From.UserID,
|
||||
ChatID: chatID,
|
||||
ChatType: chatType,
|
||||
Content: strings.TrimSpace(msg.Text.Content),
|
||||
MessageID: msg.MsgID,
|
||||
Extra: map[string]string{"req_id": reqID},
|
||||
}
|
||||
|
||||
case "voice":
|
||||
// WeCom returns speech-to-text content directly — treat as text query
|
||||
if msg.Voice.Content == "" {
|
||||
logger.Infof(ctx, "[WeCom] Ignoring voice message with empty content")
|
||||
return
|
||||
}
|
||||
incoming = &im.IncomingMessage{
|
||||
Platform: im.PlatformWeCom,
|
||||
MessageType: im.MessageTypeText,
|
||||
UserID: msg.From.UserID,
|
||||
UserName: msg.From.UserID,
|
||||
ChatID: chatID,
|
||||
ChatType: chatType,
|
||||
Content: strings.TrimSpace(msg.Voice.Content),
|
||||
MessageID: msg.MsgID,
|
||||
Extra: map[string]string{"req_id": reqID},
|
||||
}
|
||||
|
||||
case "image":
|
||||
if msg.Image.URL == "" {
|
||||
logger.Infof(ctx, "[WeCom] Ignoring image message with empty URL")
|
||||
return
|
||||
}
|
||||
incoming = &im.IncomingMessage{
|
||||
Platform: im.PlatformWeCom,
|
||||
MessageType: im.MessageTypeImage,
|
||||
UserID: msg.From.UserID,
|
||||
UserName: msg.From.UserID,
|
||||
ChatID: chatID,
|
||||
ChatType: chatType,
|
||||
MessageID: msg.MsgID,
|
||||
FileKey: msg.Image.URL, // store encrypted URL in FileKey
|
||||
FileName: msg.MsgID + ".png",
|
||||
Extra: map[string]string{"req_id": reqID, "aes_key": msg.Image.AESKey},
|
||||
}
|
||||
|
||||
case "file":
|
||||
if msg.File.URL == "" {
|
||||
logger.Infof(ctx, "[WeCom] Ignoring file message with empty URL")
|
||||
return
|
||||
}
|
||||
incoming = &im.IncomingMessage{
|
||||
Platform: im.PlatformWeCom,
|
||||
MessageType: im.MessageTypeFile,
|
||||
UserID: msg.From.UserID,
|
||||
UserName: msg.From.UserID,
|
||||
ChatID: chatID,
|
||||
ChatType: chatType,
|
||||
MessageID: msg.MsgID,
|
||||
FileKey: msg.File.URL, // store encrypted URL in FileKey
|
||||
FileName: msg.MsgID, // WeCom doesn't provide file name directly
|
||||
Extra: map[string]string{"req_id": reqID, "aes_key": msg.File.AESKey},
|
||||
}
|
||||
|
||||
case "mixed":
|
||||
// Extract text parts for QA content, and detect if any images are present
|
||||
incoming = convertMixedMessage(&msg, chatID, chatType, reqID)
|
||||
if incoming == nil {
|
||||
logger.Infof(ctx, "[WeCom] Ignoring empty mixed message")
|
||||
return
|
||||
}
|
||||
|
||||
default:
|
||||
logger.Infof(ctx, "[WeCom] Ignoring unsupported message type: %s", msg.MsgType)
|
||||
return
|
||||
}
|
||||
|
||||
if err := c.handler(ctx, incoming); err != nil {
|
||||
@@ -421,6 +527,61 @@ func (c *LongConnClient) handleCallback(ctx context.Context, frame wsFrame) {
|
||||
}
|
||||
}
|
||||
|
||||
// convertMixedMessage converts a WeCom mixed (text+image) message.
|
||||
// Extracts all text content for QA; if there's only images, treat as image message.
|
||||
func convertMixedMessage(msg *botMessage, chatID string, chatType im.ChatType, reqID string) *im.IncomingMessage {
|
||||
var textParts []string
|
||||
var firstImageURL string
|
||||
var firstImageAESKey string
|
||||
|
||||
for _, item := range msg.Mixed.MsgItem {
|
||||
switch item.MsgType {
|
||||
case "text":
|
||||
if t := strings.TrimSpace(item.Text.Content); t != "" {
|
||||
textParts = append(textParts, t)
|
||||
}
|
||||
case "image":
|
||||
if firstImageURL == "" && item.Image.URL != "" {
|
||||
firstImageURL = item.Image.URL
|
||||
firstImageAESKey = item.Image.AESKey
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// If there's text content, treat as text message (QA query)
|
||||
if len(textParts) > 0 {
|
||||
return &im.IncomingMessage{
|
||||
Platform: im.PlatformWeCom,
|
||||
MessageType: im.MessageTypeText,
|
||||
UserID: msg.From.UserID,
|
||||
UserName: msg.From.UserID,
|
||||
ChatID: chatID,
|
||||
ChatType: chatType,
|
||||
Content: strings.Join(textParts, "\n"),
|
||||
MessageID: msg.MsgID,
|
||||
Extra: map[string]string{"req_id": reqID},
|
||||
}
|
||||
}
|
||||
|
||||
// Only images, treat as image message (save to KB)
|
||||
if firstImageURL != "" {
|
||||
return &im.IncomingMessage{
|
||||
Platform: im.PlatformWeCom,
|
||||
MessageType: im.MessageTypeImage,
|
||||
UserID: msg.From.UserID,
|
||||
UserName: msg.From.UserID,
|
||||
ChatID: chatID,
|
||||
ChatType: chatType,
|
||||
MessageID: msg.MsgID,
|
||||
FileKey: firstImageURL,
|
||||
FileName: msg.MsgID + ".png",
|
||||
Extra: map[string]string{"req_id": reqID, "aes_key": firstImageAESKey},
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *LongConnClient) writeJSON(v interface{}) error {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
@@ -22,7 +22,10 @@ import (
|
||||
"encoding/xml"
|
||||
"fmt"
|
||||
"io"
|
||||
"mime"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"path"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
@@ -33,7 +36,7 @@ import (
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
var httpClient = &http.Client{Timeout: 10 * time.Second}
|
||||
var httpClient = &http.Client{Timeout: 30 * time.Second}
|
||||
|
||||
// WebhookAdapter implements im.Adapter for WeCom in webhook (self-built app callback) mode.
|
||||
// Messages arrive via HTTP callback; replies are sent via the WeCom REST API.
|
||||
@@ -51,6 +54,9 @@ type WebhookAdapter struct {
|
||||
tokenExpAt time.Time
|
||||
}
|
||||
|
||||
// Compile-time check that WebhookAdapter implements im.FileDownloader.
|
||||
var _ im.FileDownloader = (*WebhookAdapter)(nil)
|
||||
|
||||
// NewWebhookAdapter creates a new WeCom webhook adapter.
|
||||
func NewWebhookAdapter(corpID, agentSecret, token, encodingAESKey string, corpAgentID int) (*WebhookAdapter, error) {
|
||||
// Decode the AES key from base64
|
||||
@@ -146,37 +152,63 @@ func (a *WebhookAdapter) ParseCallback(c *gin.Context) (*im.IncomingMessage, err
|
||||
return nil, fmt.Errorf("decrypt message: %w", err)
|
||||
}
|
||||
|
||||
// Log raw decrypted message for debugging
|
||||
logger.Debugf(c.Request.Context(), "[WeCom] Raw decrypted callback: %s", string(decrypted))
|
||||
|
||||
var msg wecomMessage
|
||||
if err := xml.Unmarshal(decrypted, &msg); err != nil {
|
||||
return nil, fmt.Errorf("unmarshal decrypted message: %w", err)
|
||||
}
|
||||
|
||||
// Only handle text messages
|
||||
if msg.MsgType != "text" {
|
||||
logger.Infof(c.Request.Context(), "[WeCom] Ignoring non-text message type: %s", msg.MsgType)
|
||||
return nil, nil
|
||||
}
|
||||
logger.Debugf(c.Request.Context(), "[WeCom] Parsed webhook message: msgid=%s msgtype=%s from=%s content=%q picurl=%q mediaid=%q",
|
||||
msg.MsgID, msg.MsgType, msg.FromUserName, msg.Content, msg.PicUrl, msg.MediaId)
|
||||
|
||||
// Determine chat type
|
||||
chatType := im.ChatTypeDirect
|
||||
chatID := ""
|
||||
content := msg.Content
|
||||
|
||||
// Check if this is a group message (has ChatId field in WeCom smart bot callback)
|
||||
// In group chats, the content may contain @bot mention prefix that should be stripped
|
||||
if msg.ChatID != "" {
|
||||
chatType = im.ChatTypeGroup
|
||||
chatID = msg.ChatID
|
||||
}
|
||||
|
||||
return &im.IncomingMessage{
|
||||
Platform: im.PlatformWeCom,
|
||||
UserID: msg.FromUserName,
|
||||
UserName: msg.FromUserName,
|
||||
ChatID: chatID,
|
||||
ChatType: chatType,
|
||||
Content: strings.TrimSpace(content),
|
||||
MessageID: msg.MsgID,
|
||||
}, nil
|
||||
switch msg.MsgType {
|
||||
case "text":
|
||||
return &im.IncomingMessage{
|
||||
Platform: im.PlatformWeCom,
|
||||
MessageType: im.MessageTypeText,
|
||||
UserID: msg.FromUserName,
|
||||
UserName: msg.FromUserName,
|
||||
ChatID: chatID,
|
||||
ChatType: chatType,
|
||||
Content: strings.TrimSpace(msg.Content),
|
||||
MessageID: msg.MsgID,
|
||||
}, nil
|
||||
|
||||
case "image":
|
||||
// Image via webhook: has PicUrl (direct download) and MediaId
|
||||
if msg.PicUrl == "" && msg.MediaId == "" {
|
||||
return nil, nil
|
||||
}
|
||||
fileKey := msg.PicUrl
|
||||
if fileKey == "" {
|
||||
fileKey = msg.MediaId
|
||||
}
|
||||
return &im.IncomingMessage{
|
||||
Platform: im.PlatformWeCom,
|
||||
MessageType: im.MessageTypeImage,
|
||||
UserID: msg.FromUserName,
|
||||
UserName: msg.FromUserName,
|
||||
ChatID: chatID,
|
||||
ChatType: chatType,
|
||||
MessageID: msg.MsgID,
|
||||
FileKey: fileKey,
|
||||
FileName: msg.MsgID + ".png",
|
||||
}, nil
|
||||
|
||||
default:
|
||||
logger.Infof(c.Request.Context(), "[WeCom] Ignoring unsupported message type: %s", msg.MsgType)
|
||||
return nil, nil
|
||||
}
|
||||
}
|
||||
|
||||
// SendReply sends a reply message via WeCom API.
|
||||
@@ -357,14 +389,160 @@ type callbackRequestBody struct {
|
||||
}
|
||||
|
||||
// wecomMessage is the decrypted WeCom message structure.
|
||||
// Supports text, image, voice, video, location, and link message types.
|
||||
// Reference: https://developer.work.weixin.qq.com/document/path/90375
|
||||
type wecomMessage struct {
|
||||
XMLName xml.Name `xml:"xml"`
|
||||
ToUserName string `xml:"ToUserName"`
|
||||
FromUserName string `xml:"FromUserName"`
|
||||
CreateTime int64 `xml:"CreateTime"`
|
||||
MsgType string `xml:"MsgType"`
|
||||
Content string `xml:"Content"`
|
||||
Content string `xml:"Content"` // text
|
||||
PicUrl string `xml:"PicUrl"` // image: download URL
|
||||
MediaId string `xml:"MediaId"` // image/voice/video: media ID for download
|
||||
Format string `xml:"Format"` // voice: audio format (amr/speex)
|
||||
ThumbMediaId string `xml:"ThumbMediaId"` // video: thumbnail media ID
|
||||
MsgID string `xml:"MsgId"`
|
||||
AgentID string `xml:"AgentID"`
|
||||
ChatID string `xml:"ChatId"`
|
||||
}
|
||||
|
||||
// ──────────────────────────────────────────────────────────────────────
|
||||
// File download support for WeCom webhook mode
|
||||
// ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
// DownloadFile downloads a file/image from WeCom.
|
||||
// For webhook mode, images come with MediaId (temporary media) which can be
|
||||
// downloaded via the GetMedia API, or PicUrl for direct download.
|
||||
func (a *WebhookAdapter) DownloadFile(ctx context.Context, msg *im.IncomingMessage) (io.ReadCloser, string, error) {
|
||||
if msg.FileKey == "" {
|
||||
return nil, "", fmt.Errorf("no file key (URL or media_id) in message")
|
||||
}
|
||||
|
||||
fileName := msg.FileName
|
||||
if fileName == "" {
|
||||
fileName = msg.FileKey
|
||||
}
|
||||
|
||||
// If FileKey looks like a URL, download directly
|
||||
if strings.HasPrefix(msg.FileKey, "http://") || strings.HasPrefix(msg.FileKey, "https://") {
|
||||
return downloadFromURL(ctx, msg.FileKey, fileName)
|
||||
}
|
||||
|
||||
// Otherwise treat as media_id, download via temporary media API
|
||||
accessToken, err := a.getAccessToken(ctx)
|
||||
if err != nil {
|
||||
return nil, "", fmt.Errorf("get access token: %w", err)
|
||||
}
|
||||
|
||||
apiURL := fmt.Sprintf("https://qyapi.weixin.qq.com/cgi-bin/media/get?access_token=%s&media_id=%s",
|
||||
accessToken, msg.FileKey)
|
||||
return downloadFromURL(ctx, apiURL, fileName)
|
||||
}
|
||||
|
||||
// downloadFromURL performs a GET request and returns the response body.
|
||||
// It tries to resolve the real filename from HTTP response headers:
|
||||
// 1. Content-Disposition: attachment; filename="xxx.pdf"
|
||||
// 2. Content-Type → extension mapping (fallback for platforms like WeCom that
|
||||
// don't provide the original filename in the callback JSON)
|
||||
func downloadFromURL(ctx context.Context, rawURL, fileName string) (io.ReadCloser, string, error) {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, rawURL, nil)
|
||||
if err != nil {
|
||||
return nil, "", fmt.Errorf("create request: %w", err)
|
||||
}
|
||||
|
||||
resp, err := httpClient.Do(req)
|
||||
if err != nil {
|
||||
return nil, "", fmt.Errorf("download: %w", err)
|
||||
}
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
resp.Body.Close()
|
||||
return nil, "", fmt.Errorf("download failed: status=%d", resp.StatusCode)
|
||||
}
|
||||
|
||||
logger.Debugf(ctx, "[WeCom] Download response: status=%d content-type=%s content-disposition=%s",
|
||||
resp.StatusCode, resp.Header.Get("Content-Type"), resp.Header.Get("Content-Disposition"))
|
||||
|
||||
// Try to extract filename from Content-Disposition header.
|
||||
// Supports both standard filename and RFC 5987 filename* parameters.
|
||||
if cd := resp.Header.Get("Content-Disposition"); cd != "" {
|
||||
if _, params, err := mime.ParseMediaType(cd); err == nil {
|
||||
// Prefer filename* (RFC 5987, already decoded by mime.ParseMediaType)
|
||||
if fn := params["filename"]; fn != "" {
|
||||
fileName = fn
|
||||
}
|
||||
} else {
|
||||
// Fallback: manual extraction for malformed headers
|
||||
if idx := strings.Index(cd, "filename="); idx >= 0 {
|
||||
extracted := strings.Trim(cd[idx+len("filename="):], "\" ")
|
||||
if extracted != "" {
|
||||
fileName = extracted
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// URL-decode the filename if it contains percent-encoded characters.
|
||||
// Some servers (e.g. WeCom COS) return URL-encoded Chinese filenames.
|
||||
if strings.Contains(fileName, "%") {
|
||||
if decoded, err := url.QueryUnescape(fileName); err == nil && decoded != "" {
|
||||
fileName = decoded
|
||||
}
|
||||
}
|
||||
|
||||
// Also try to extract a meaningful filename from the URL path itself,
|
||||
// in case Content-Disposition is missing but the URL contains the real name.
|
||||
if !strings.Contains(fileName, ".") {
|
||||
if u, err := url.Parse(rawURL); err == nil {
|
||||
base := path.Base(u.Path)
|
||||
if base != "" && base != "." && base != "/" && strings.Contains(base, ".") {
|
||||
// URL-decode the path component as well
|
||||
if decoded, err := url.QueryUnescape(base); err == nil {
|
||||
fileName = decoded
|
||||
} else {
|
||||
fileName = base
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// If filename still has no extension, try to infer from Content-Type.
|
||||
// This handles platforms (e.g. WeCom aibot) where the callback only provides
|
||||
// a hash ID as the filename without any extension.
|
||||
if !strings.Contains(fileName, ".") {
|
||||
if ext := contentTypeToExt(resp.Header.Get("Content-Type")); ext != "" {
|
||||
fileName = fileName + "." + ext
|
||||
}
|
||||
}
|
||||
|
||||
return resp.Body, fileName, nil
|
||||
}
|
||||
|
||||
// contentTypeToExt maps common Content-Type values to file extensions.
|
||||
func contentTypeToExt(ct string) string {
|
||||
// Normalize: take only the media type, ignore parameters like charset
|
||||
if idx := strings.Index(ct, ";"); idx >= 0 {
|
||||
ct = strings.TrimSpace(ct[:idx])
|
||||
}
|
||||
ct = strings.ToLower(ct)
|
||||
|
||||
mapping := map[string]string{
|
||||
"application/pdf": "pdf",
|
||||
"application/msword": "doc",
|
||||
"application/vnd.openxmlformats-officedocument.wordprocessingml.document": "docx",
|
||||
"application/vnd.ms-excel": "xls",
|
||||
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": "xlsx",
|
||||
"application/vnd.ms-powerpoint": "ppt",
|
||||
"application/vnd.openxmlformats-officedocument.presentationml.presentation": "pptx",
|
||||
"text/plain": "txt",
|
||||
"text/markdown": "md",
|
||||
"text/csv": "csv",
|
||||
"image/png": "png",
|
||||
"image/jpeg": "jpg",
|
||||
"image/gif": "gif",
|
||||
"image/webp": "webp",
|
||||
}
|
||||
|
||||
return mapping[ct]
|
||||
}
|
||||
|
||||
@@ -1,17 +1,24 @@
|
||||
package wecom
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/aes"
|
||||
"crypto/cipher"
|
||||
"encoding/base64"
|
||||
"fmt"
|
||||
"io"
|
||||
|
||||
"github.com/Tencent/WeKnora/internal/im"
|
||||
"github.com/Tencent/WeKnora/internal/logger"
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
// Compile-time checks.
|
||||
var (
|
||||
_ im.Adapter = (*WSAdapter)(nil)
|
||||
_ im.StreamSender = (*WSAdapter)(nil)
|
||||
_ im.Adapter = (*WSAdapter)(nil)
|
||||
_ im.StreamSender = (*WSAdapter)(nil)
|
||||
_ im.FileDownloader = (*WSAdapter)(nil)
|
||||
)
|
||||
|
||||
// WSAdapter implements im.Adapter and im.StreamSender for WeCom in WebSocket
|
||||
@@ -60,3 +67,107 @@ func (a *WSAdapter) SendStreamChunk(ctx context.Context, incoming *im.IncomingMe
|
||||
func (a *WSAdapter) EndStream(ctx context.Context, incoming *im.IncomingMessage, streamID string) error {
|
||||
return a.client.EndStream(ctx, incoming, streamID)
|
||||
}
|
||||
|
||||
// ── FileDownloader implementation ──
|
||||
// WeCom aibot provides AES-256-CBC encrypted URLs for image/file/video messages.
|
||||
// Each message carries its own aeskey for decryption.
|
||||
|
||||
func (a *WSAdapter) DownloadFile(ctx context.Context, msg *im.IncomingMessage) (io.ReadCloser, string, error) {
|
||||
if msg.FileKey == "" {
|
||||
return nil, "", fmt.Errorf("no file URL in message")
|
||||
}
|
||||
fileName := msg.FileName
|
||||
if fileName == "" {
|
||||
fileName = msg.FileKey
|
||||
}
|
||||
|
||||
// Download the (encrypted) file content
|
||||
reader, fileName, err := downloadFromURL(ctx, msg.FileKey, fileName)
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
|
||||
// If an AES key is provided, the downloaded content is AES-256-CBC encrypted
|
||||
// and must be decrypted before use. This is the case for WeCom aibot long
|
||||
// connection mode where each file/image message carries a per-message aeskey.
|
||||
aesKeyB64 := msg.Extra["aes_key"]
|
||||
if aesKeyB64 == "" {
|
||||
// No encryption — return raw content (e.g. webhook mode uses media API)
|
||||
return reader, fileName, nil
|
||||
}
|
||||
|
||||
// Read all encrypted content
|
||||
encryptedData, err := io.ReadAll(reader)
|
||||
reader.Close()
|
||||
if err != nil {
|
||||
return nil, "", fmt.Errorf("read encrypted file: %w", err)
|
||||
}
|
||||
|
||||
logger.Debugf(ctx, "[WeCom] Decrypting file: name=%s encrypted_size=%d aes_key_len=%d",
|
||||
fileName, len(encryptedData), len(aesKeyB64))
|
||||
|
||||
// Decrypt
|
||||
decrypted, err := decryptAESCBC(encryptedData, aesKeyB64)
|
||||
if err != nil {
|
||||
return nil, "", fmt.Errorf("decrypt file: %w", err)
|
||||
}
|
||||
|
||||
logger.Debugf(ctx, "[WeCom] File decrypted: name=%s decrypted_size=%d", fileName, len(decrypted))
|
||||
|
||||
return io.NopCloser(bytes.NewReader(decrypted)), fileName, nil
|
||||
}
|
||||
|
||||
// decryptAESCBC decrypts data encrypted with AES-256-CBC using PKCS#7 padding.
|
||||
// The aesKeyB64 is the base64-encoded AES key provided per-message by WeCom.
|
||||
// IV is the first 16 bytes of the decoded AES key.
|
||||
func decryptAESCBC(ciphertext []byte, aesKeyB64 string) ([]byte, error) {
|
||||
// WeCom's per-message aeskey is base64-encoded (43 chars → 32 bytes after decode)
|
||||
aesKey, err := base64.StdEncoding.DecodeString(aesKeyB64 + "=")
|
||||
if err != nil {
|
||||
// Try without padding
|
||||
aesKey, err = base64.RawStdEncoding.DecodeString(aesKeyB64)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("base64 decode aes key: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
if len(aesKey) < 16 {
|
||||
return nil, fmt.Errorf("aes key too short: %d bytes", len(aesKey))
|
||||
}
|
||||
|
||||
block, err := aes.NewCipher(aesKey)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("new aes cipher: %w", err)
|
||||
}
|
||||
|
||||
if len(ciphertext) < aes.BlockSize {
|
||||
return nil, fmt.Errorf("ciphertext too short: %d bytes", len(ciphertext))
|
||||
}
|
||||
if len(ciphertext)%aes.BlockSize != 0 {
|
||||
return nil, fmt.Errorf("ciphertext not a multiple of block size: %d bytes", len(ciphertext))
|
||||
}
|
||||
|
||||
// IV = first 16 bytes of the AES key
|
||||
iv := aesKey[:aes.BlockSize]
|
||||
mode := cipher.NewCBCDecrypter(block, iv)
|
||||
plaintext := make([]byte, len(ciphertext))
|
||||
mode.CryptBlocks(plaintext, ciphertext)
|
||||
|
||||
// Remove PKCS#7 padding
|
||||
if len(plaintext) == 0 {
|
||||
return nil, fmt.Errorf("empty plaintext after decryption")
|
||||
}
|
||||
padLen := int(plaintext[len(plaintext)-1])
|
||||
if padLen > aes.BlockSize || padLen == 0 || padLen > len(plaintext) {
|
||||
// No valid PKCS#7 padding — return as-is (some implementations may not pad)
|
||||
return plaintext, nil
|
||||
}
|
||||
// Verify padding bytes
|
||||
for i := 0; i < padLen; i++ {
|
||||
if plaintext[len(plaintext)-1-i] != byte(padLen) {
|
||||
// Invalid padding — return as-is
|
||||
return plaintext, nil
|
||||
}
|
||||
}
|
||||
return plaintext[:len(plaintext)-padLen], nil
|
||||
}
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
ALTER TABLE im_channels DROP COLUMN IF EXISTS knowledge_base_id;
|
||||
@@ -0,0 +1,3 @@
|
||||
-- Add knowledge_base_id column to im_channels table.
|
||||
-- When set, file messages received on this channel will be saved to the specified knowledge base.
|
||||
ALTER TABLE im_channels ADD COLUMN IF NOT EXISTS knowledge_base_id VARCHAR(36) DEFAULT '';
|
||||
@@ -0,0 +1,2 @@
|
||||
DROP INDEX IF EXISTS idx_im_channels_bot_identity;
|
||||
ALTER TABLE im_channels DROP COLUMN IF EXISTS bot_identity;
|
||||
@@ -0,0 +1,17 @@
|
||||
-- Migration: 000024_im_channel_bot_identity
|
||||
-- Description: Add bot_identity column to im_channels for duplicate bot prevention.
|
||||
-- bot_identity is a computed unique key derived from credentials, ensuring each bot
|
||||
-- can only be connected to one active (non-deleted) channel.
|
||||
DO $$ BEGIN RAISE NOTICE '[Migration 000024] Adding bot_identity column to im_channels'; END $$;
|
||||
|
||||
ALTER TABLE im_channels ADD COLUMN IF NOT EXISTS bot_identity VARCHAR(255) NOT NULL DEFAULT '';
|
||||
|
||||
-- Partial unique index: only enforce uniqueness for non-deleted rows with a non-empty bot_identity.
|
||||
-- Empty bot_identity (unknown credential format) is excluded from the constraint.
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS idx_im_channels_bot_identity
|
||||
ON im_channels (bot_identity)
|
||||
WHERE deleted_at IS NULL AND bot_identity != '';
|
||||
|
||||
COMMENT ON COLUMN im_channels.bot_identity IS 'Unique bot identity derived from credentials (e.g. wecom:ws:{bot_id}, feishu:{app_id}). Used to prevent duplicate bot bindings.';
|
||||
|
||||
DO $$ BEGIN RAISE NOTICE '[Migration 000024] bot_identity column and unique index created successfully'; END $$;
|
||||
Reference in New Issue
Block a user