fuzhongyun
/
ai_scheduler
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: 群聊使用知识库流程基本串通

This commit is contained in:
fuzhongyun 2026-01-26 15:08:33 +08:00
parent 855156374e
commit 634bca5c60
10 changed files with 403 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
config/config_env.yaml
View File

@ -6,7 +6,7 @@ server:
ollama:
base_url: "http://192.168.6.115:11434"
model: "qwen3:8b"
generate_model: "qwen3:8b"
generate_model: "deepseek-v3.2:cloud"
mapping_model: "qwen3:8b"
vl_model: "qwen2.5vl:7b"
timeout: "120s"

4
internal/biz/group_config.go
View File

@ -480,7 +480,7 @@ func (g *GroupConfigBiz) GetReportCache(ctx context.Context, day time.Time, tota
func (g *GroupConfigBiz) handleKnowledgeV2(ctx context.Context, rec *entitys.Recognize, groupConfig *model.AiBotGroupConfig, callback *chatbot.BotCallbackDataModel) (err error) {
// 请求知识库工具
knowledgeBase := knowledge_base.New(g.conf.KnowledgeConfig)
knowledgeResp, err := knowledgeBase.Call(&knowledge_base.ChatRequest{
knowledgeResp, err := knowledgeBase.Query(&knowledge_base.QueryRequest{
TenantID: constants.KnowledgeTenantIdDefault, // 后续动态接参
Query: rec.UserContent.Text,
Mode: constants.KnowledgeModeMix,
@ -615,7 +615,7 @@ func (g *GroupConfigBiz) shouldCreateIssueHandlingGroup(ctx context.Context, rec
"action_id": tea.String("create_group"),
"button_display": tea.String("true"),
"group_scope": tea.String(strings.TrimSpace(rec.UserContent.Text)),
// "_CARD_DEBUG_TOOL_ENTRY": tea.String("show"), // debug字段
"_CARD_DEBUG_TOOL_ENTRY": tea.String(constants.CardDebugToolEntryShow), // 调试字段
},
},
ImGroupOpenSpaceModel: &card_1_0.CreateAndDeliverRequestImGroupOpenSpaceModel{

2
internal/config/config.go
View File

@ -262,8 +262,6 @@ type KnowledgeConfig struct {
TenantID string `mapstructure:"tenant_id"`
// 模式
Mode string `mapstructure:"mode"`
// 是否流式
Stream bool `mapstructure:"stream"`
// 是否思考
Think bool `mapstructure:"think"`
// 是否仅RAG

7
internal/data/constants/dingtalk.go
View File

@ -131,3 +131,10 @@ const (
// 模板群机器人ID
GroupTemplateRobotIdIssueHandling string = "VqgJYpB91j3RnB217690607273471011" // 问题处理群模板机器人ID
)
// 群模板机器人 - 主应用机器人映射
var GroupTemplateRobotIdMap = map[string]string{
GroupTemplateRobotIdIssueHandling: "ding5wwvnf9hxeyjau7t",
}
const CardDebugToolEntryShow string = "show" // 卡片调试工具 [show：展示 hide：隐藏]

42
internal/domain/tools/common/knowledge_base/client.go
View File

@ -17,7 +17,8 @@ func New(cfg config.KnowledgeConfig) *Client {
return &Client{cfg: cfg}
}
func (c *Client) Call(req *ChatRequest) (io.ReadCloser, error) {
// 查询知识库
func (c *Client) Query(req *QueryRequest) (io.ReadCloser, error) {
if req == nil {
return nil, fmt.Errorf("req is nil")
}
@ -30,9 +31,6 @@ func (c *Client) Call(req *ChatRequest) (io.ReadCloser, error) {
if req.Mode == "" {
req.Mode = c.cfg.Mode
}
if !req.Stream {
req.Stream = c.cfg.Stream // 仅支持流式输出
}
if !req.Think {
req.Think = c.cfg.Think
}
@ -76,3 +74,39 @@ func (c *Client) Call(req *ChatRequest) (io.ReadCloser, error) {
return rsp.Body, nil
}
// IngestText 向知识库中注入文本
func (c *Client) IngestText(req *IngestTextRequest) error {
if req == nil {
return fmt.Errorf("req is nil")
}
if req.TenantID == "" {
return fmt.Errorf("tenantID is empty")
}
if req.Text == "" {
return fmt.Errorf("text is empty")
}
baseURL := strings.TrimRight(c.cfg.BaseURL, "/")
rsp, err := (&l_request.Request{
Method: "POST",
Url: baseURL + "/ingest/text",
Headers: map[string]string{
"Content-Type": "application/json",
"X-Tenant-ID": req.TenantID,
},
Json: map[string]interface{}{
"text": req.Text,
},
}).Send()
if err != nil {
return err
}
if rsp.StatusCode != http.StatusOK {
return fmt.Errorf("knowledge base returned status %d: %s", rsp.StatusCode, rsp.Text)
}
return nil
}

4
internal/domain/tools/common/knowledge_base/client_test.go
View File

@ -8,7 +8,7 @@ import (
)
func TestCall(t *testing.T) {
req := &ChatRequest{
req := &QueryRequest{
TenantID: "admin_test_qa",
Query: "lightRAG 的优势?",
Mode: "naive",
@ -18,7 +18,7 @@ func TestCall(t *testing.T) {
}
client := New(config.KnowledgeConfig{BaseURL: "http://127.0.0.1:9600"})
resp, err := client.Call(req)
resp, err := client.Query(req)
if err != nil {
t.Errorf("Call failed: %v", err)
}

27
internal/domain/tools/common/knowledge_base/parse.go
View File

@ -9,6 +9,7 @@ import (
type openAIChunk struct {
Choices []struct {
Delta *Delta `json:"delta"`
Message *Message `json:"message"`
FinishReason *string `json:"finish_reason"`
} `json:"choices"`
}
@ -19,6 +20,12 @@ type Delta struct {
XRagStatus string `json:"x_rag_status"` // rag命中状态 hit|miss
}
type Message struct {
Role string `json:"role"` // 角色
Content string `json:"content"` // 内容
XRagStatus string `json:"x_rag_status"` // rag命中状态 hit|miss
}
func ParseOpenAIStreamData(dataLine string) (delta *Delta, done bool, err error) {
data := strings.TrimSpace(strings.TrimPrefix(dataLine, "data:"))
if data == "" {
@ -46,3 +53,23 @@ func ParseOpenAIStreamData(dataLine string) (delta *Delta, done bool, err error)
return nil, false, nil
}
func ParseOpenAIHTTPData(body string) (message *Message, done bool, err error) {
data := strings.TrimSpace(body)
if data == "" {
return nil, false, nil
}
var resp openAIChunk
if err := json.Unmarshal([]byte(data), &resp); err != nil {
return nil, false, fmt.Errorf("unmarshal openai stream chunk failed: %w", err)
}
for _, c := range resp.Choices {
if c.Message != nil {
return c.Message, true, nil // 只输出第一个message
}
}
return nil, false, nil
}

7
internal/domain/tools/common/knowledge_base/type.go
View File

@ -1,6 +1,6 @@
package knowledge_base
type ChatRequest struct {
type QueryRequest struct {
TenantID string // 租户 ID
Query string // 查询内容
Mode string // 模式，默认 naive 可选：[bypass|naive|local|global|hybrid|mix]
@ -8,3 +8,8 @@ type ChatRequest struct {
Think bool // 是否开启思考模式
OnlyRAG bool // 是否仅开启 RAG 模式
}
type IngestTextRequest struct {
TenantID string // 租户 ID
Text string // 要注入的文本内容
}

62
internal/pkg/util/safe_pool.go Normal file
View File

@ -0,0 +1,62 @@
package util
import (
"os"
"runtime/debug"
"sync"
"time"
"github.com/bytedance/gopkg/util/gopool"
"github.com/go-kratos/kratos/v2/log"
)
var (
logger *log.Helper
once sync.Once
)
// getLogger 懒加载获取日志器
func getLogger() *log.Helper {
once.Do(func() {
// 如果没有手动初始化，使用默认的标准输出日志器
if logger == nil {
stdLogger := log.With(log.NewStdLogger(os.Stdout),
"ts", log.DefaultTimestamp,
"caller", log.DefaultCaller,
"component", "safe_pool",
)
logger = log.NewHelper(stdLogger)
}
})
return logger
}
// InitSafePool 初始化安全协程池（可选，如果不调用会使用默认日志器）
func InitSafePool(l log.Logger) {
logger = log.NewHelper(l)
}
// SafeGo 安全执行协程
// taskName: 协程任务名称，用于日志记录
// fn: 要执行的函数
func SafeGo(taskName string, fn func()) {
gopool.Go(func() {
defer func() {
if r := recover(); r != nil {
stack := debug.Stack()
getLogger().Errorf("协程 [%s] 发生panic: %v\n堆栈信息:\n%s", taskName, r, string(stack))
}
}()
// 记录协程开始执行
getLogger().Infof("协程 [%s] 开始执行", taskName)
start := time.Now()
// 执行用户函数
fn()
// 记录协程执行完成
duration := time.Since(start)
getLogger().Infof("协程 [%s] 执行完成，耗时: %v", taskName, duration)
})
}

244
internal/services/callback.go
View File

@ -6,7 +6,9 @@ import (
"ai_scheduler/internal/config"
"ai_scheduler/internal/data/constants"
errorcode "ai_scheduler/internal/data/error"
"ai_scheduler/internal/data/impl"
"ai_scheduler/internal/domain/component/callback"
"ai_scheduler/internal/domain/tools/common/knowledge_base"
"ai_scheduler/internal/entitys"
"ai_scheduler/internal/gateway"
"ai_scheduler/internal/pkg"
@ -14,9 +16,11 @@ import (
"ai_scheduler/internal/pkg/util"
"ai_scheduler/internal/pkg/utils_ollama"
"ai_scheduler/internal/tool_callback"
"bufio"
"context"
"encoding/json"
"fmt"
"io"
"net/url"
"strings"
"time"
@ -41,6 +45,7 @@ type CallbackService struct {
callbackManager callback.Manager
dingTalkBotBiz *biz.DingTalkBotBiz
ollamaClient *utils_ollama.Client
botConfigImpl *impl.BotConfigImpl
}
func NewCallbackService(
@ -53,6 +58,7 @@ func NewCallbackService(
callbackManager callback.Manager,
dingTalkBotBiz *biz.DingTalkBotBiz,
ollamaClient *utils_ollama.Client,
botConfigImpl *impl.BotConfigImpl,
) *CallbackService {
return &CallbackService{
cfg: cfg,
@ -64,6 +70,7 @@ func NewCallbackService(
callbackManager: callbackManager,
dingTalkBotBiz: dingTalkBotBiz,
ollamaClient: ollamaClient,
botConfigImpl: botConfigImpl,
}
}
@ -434,11 +441,16 @@ func (s *CallbackService) CallbackDingtalkRobot(c *fiber.Ctx) (err error) {
// issueHandling 问题处理群机器人回调
// 能力1： 通过[内容提取] 宏，分析用户QA问题，调出QA表单卡片
// 能力2： 通过[QA收集] 宏，收集用户反馈，写入知识库
// 能力3： 通过[知识库查询] 宏，查询知识库，返回答案
func (s *CallbackService) issueHandling(c *fiber.Ctx, data chatbot.BotCallbackDataModel) error {
// 宏解析
// 能力1、2：分析用户QA问题，写入知识库
if strings.Contains(data.Text.Content, "[内容提取]") || strings.Contains(data.Text.Content, "[QA收集]") {
s.issueHandlingExtractContent(data)
}
// 能力3：查询知识库，返回答案
if strings.Contains(data.Text.Content, "[知识库查询]") {
s.issueHandlingQueryKnowledgeBase(data)
}
return nil
}
@ -551,7 +563,7 @@ func (s *CallbackService) issueHandlingExtractContent(data chatbot.BotCallbackDa
"textarea_display": tea.String("normal"),
"action_id": tea.String("collect_qa"),
"tenant_id": tea.String(constants.KnowledgeTenantIdDefault),
"_CARD_DEBUG_TOOL_ENTRY": tea.String("show"), // debug字段
"_CARD_DEBUG_TOOL_ENTRY": tea.String(constants.CardDebugToolEntryShow), // 调试字段
},
},
ImGroupOpenSpaceModel: &card_1_0.CreateAndDeliverRequestImGroupOpenSpaceModel{
@ -569,6 +581,165 @@ func (s *CallbackService) issueHandlingExtractContent(data chatbot.BotCallbackDa
}
// 问题处理群机器人查询知识库
func (s *CallbackService) issueHandlingQueryKnowledgeBase(data chatbot.BotCallbackDataModel) {
// 获取应用主机器人
mainRobotCode := data.RobotCode
if robotCode, ok := constants.GroupTemplateRobotIdMap[data.RobotCode]; ok {
mainRobotCode = robotCode
}
// 获取应用机器人配置
robotConfig, err := s.botConfigImpl.GetRobotConfig(mainRobotCode)
if err != nil {
log.Errorf("应用机器人配置不存在: %s, err: %v", mainRobotCode, err)
return
}
// 创建卡片
outTrackId := constants.BuildCardOutTrackId(data.ConversationId, mainRobotCode)
_, err = s.dingtalkCardClient.CreateAndDeliver(
dingtalk.AppKey{
AppKey: robotConfig.ClientId,
AppSecret: robotConfig.ClientSecret,
},
&card_1_0.CreateAndDeliverRequest{
CardTemplateId: tea.String(constants.DingtalkCardTplBaseMsg),
CardData: &card_1_0.CreateAndDeliverRequestCardData{
CardParamMap: map[string]*string{
"title": tea.String(data.Text.Content),
"markdown": tea.String("知识库检索中..."),
},
},
OutTrackId: tea.String(outTrackId),
ImGroupOpenSpaceModel: &card_1_0.CreateAndDeliverRequestImGroupOpenSpaceModel{
SupportForward: tea.Bool(false),
},
OpenSpaceId: tea.String("dtv1.card//im_group." + data.ConversationId),
ImGroupOpenDeliverModel: &card_1_0.CreateAndDeliverRequestImGroupOpenDeliverModel{
RobotCode: tea.String(data.RobotCode),
Recipients: []*string{
tea.String(data.SenderStaffId),
},
},
},
)
// 查询知识库
knowledgeBase := knowledge_base.New(s.cfg.KnowledgeConfig)
knowledgeResp, err := knowledgeBase.Query(&knowledge_base.QueryRequest{
TenantID: constants.KnowledgeTenantIdDefault,
Query: data.Text.Content,
Mode: constants.KnowledgeModeMix,
Stream: false,
Think: false,
OnlyRAG: true,
})
if err != nil {
log.Errorf("查询知识库失败: %v", err)
return
}
knowledgeRespBytes, err := io.ReadAll(knowledgeResp)
if err != nil {
log.Errorf("读取知识库响应失败: %v", err)
return
}
// 卡片更新
message, isRetrieved, err := knowledge_base.ParseOpenAIHTTPData(string(knowledgeRespBytes))
if err != nil {
log.Errorf("读取知识库 SSE 数据失败: %v", err)
return
}
content := message.Content
if !isRetrieved {
content = "知识库未检测到匹配信息，请核查知识库数据是否正确。"
}
// 卡片更新
_, err = s.dingtalkCardClient.UpdateCard(
dingtalk.AppKey{
AppKey: robotConfig.ClientId,
AppSecret: robotConfig.ClientSecret,
},
&card_1_0.UpdateCardRequest{
OutTrackId: tea.String(outTrackId),
CardData: &card_1_0.UpdateCardRequestCardData{
CardParamMap: map[string]*string{
"markdown": tea.String(content),
},
},
CardUpdateOptions: &card_1_0.UpdateCardRequestCardUpdateOptions{
UpdateCardDataByKey: tea.Bool(true),
},
},
)
if err != nil {
log.Errorf("更新卡片失败: %v", err)
return
}
return
}
// 读取知识库 SSE 数据
func (s *CallbackService) readKnowledgeSSE(resp io.ReadCloser, channel chan string) (isRetrieved bool, err error) {
scanner := bufio.NewScanner(resp)
var buffer strings.Builder
for scanner.Scan() {
line := scanner.Text()
delta, done, err := knowledge_base.ParseOpenAIStreamData(line)
if err != nil {
return false, fmt.Errorf("解析SSE数据失败: %w", err)
}
if done {
break
}
if delta == nil {
continue
}
// 知识库未命中 输出提示后中断
if delta.XRagStatus == constants.KnowledgeRagStatusMiss {
var missContent string = "知识库未检测到匹配信息，即将为您创建群聊解决问题。"
channel <- missContent
return false, nil
}
// 推理内容
if delta.ReasoningContent != "" {
channel <- delta.ReasoningContent
continue
}
// 输出内容 - 段落
// 存入缓冲区
buffer.WriteString(delta.Content)
content := buffer.String()
// 检查是否有换行符，按段落输出
if idx := strings.LastIndex(content, "\n"); idx != -1 {
// 发送直到最后一个换行符的内容
toSend := content[:idx+1]
channel <- toSend
// 重置缓冲区，保留剩余部分
remaining := content[idx+1:]
buffer.Reset()
buffer.WriteString(remaining)
}
}
if err := scanner.Err(); err != nil {
return true, fmt.Errorf("读取SSE流中断: %w", err)
}
// 发送缓冲区剩余内容（仅在段落模式下需要）
if buffer.Len() > 0 {
channel <- buffer.String()
}
return true, nil
}
// CallbackDingtalkCard 处理钉钉卡片回调
// 钉钉 callbackRouteKey: gateway.dev.cdlsxd.cn-dingtalk-card
// 钉钉 apiSecret: aB3dE7fG9hI2jK4L5M6N7O8P9Q0R1S2T
@ -613,7 +784,71 @@ func (s *CallbackService) CallbackDingtalkCard(c *fiber.Ctx) error {
// 问题处理群机器人 QA 收集
func (s *CallbackService) issueHandlingCollectQA(data card.CardRequest) *card.CardResponse {
if data.CardActionData.CardPrivateData.Params["submit"] != "submit" {
// 确认提交，文本写入知识库
if data.CardActionData.CardPrivateData.Params["submit"] == "submit" {
content := data.CardActionData.CardPrivateData.Params["QA_details"].(string)
tenantID := data.CardActionData.CardPrivateData.Params["tenant_id"].(string)
// 协程执行耗时操作，防止阻塞
util.SafeGo("inject_knowledge_base", func() {
knowledgeBase := knowledge_base.New(s.cfg.KnowledgeConfig)
err := knowledgeBase.IngestText(&knowledge_base.IngestTextRequest{
TenantID: tenantID,
Text: content,
})
if err != nil {
log.Errorf("注入知识库失败: %v", err)
} else {
log.Infof("注入知识库成功: tenantID=%s", tenantID)
}
// 解析当前卡片的 ConversationId 和 robotCode
conversationId, robotCode := constants.ParseCardOutTrackId(data.OutTrackId)
// 获取主应用机器人（这里可能是群模板机器人）
mainRobotId := robotCode
if robotCode, ok := constants.GroupTemplateRobotIdMap[robotCode]; ok {
mainRobotId = robotCode
}
// 获取 robot 配置
robotConfig, err := s.botConfigImpl.GetRobotConfig(mainRobotId)
if err != nil {
log.Errorf("获取 robot 配置失败: %v", err)
return
}
// 发送卡片通知用户注入成功
outTrackId := constants.BuildCardOutTrackId(conversationId, robotCode)
s.dingtalkCardClient.CreateAndDeliver(
dingtalk.AppKey{
AppKey: robotConfig.ClientId,
AppSecret: robotConfig.ClientSecret,
},
&card_1_0.CreateAndDeliverRequest{
CardTemplateId: tea.String(constants.DingtalkCardTplBaseMsg),
OutTrackId: tea.String(outTrackId),
CardData: &card_1_0.CreateAndDeliverRequestCardData{
CardParamMap: map[string]*string{
"title": tea.String("QA知识收集结果"),
"markdown": tea.String("[Get] **成功**"),
},
},
ImGroupOpenSpaceModel: &card_1_0.CreateAndDeliverRequestImGroupOpenSpaceModel{
SupportForward: tea.Bool(false),
},
OpenSpaceId: tea.String("dtv1.card//im_group." + conversationId),
ImGroupOpenDeliverModel: &card_1_0.CreateAndDeliverRequestImGroupOpenDeliverModel{
RobotCode: tea.String(robotCode),
Recipients: []*string{
tea.String(data.UserId),
},
},
},
)
})
}
// 取消提交，禁用输入框
resp := &card.CardResponse{
CardUpdateOptions: &card.CardUpdateOptions{
@ -627,7 +862,4 @@ func (s *CallbackService) issueHandlingCollectQA(data card.CardRequest) *card.Ca
}
return resp
}
return nil
}