Skip to content

feat: 支持网络恢复补跑与 OneBot 群文件 Provider #4

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
123kaze wants to merge 2 commits into
base: main
Choose a base branch
from
Open

feat: 支持网络恢复补跑与 OneBot 群文件 Provider #4

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
529a7f5
feat: 完善网络恢复后的热点生成流程
123kaze Jun 13, 2026
e2ad065
feat: add onebot group file provider
123kaze Jun 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
179 changes: 150 additions & 29 deletions applications/ai-model/src/tasks/AISummarize.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
import "reflect-metadata";
import type { SessionDigestCoverage } from "@root/common/services/database/AgcDbAccessService";

import { injectable, inject } from "tsyringe";
import { agendaInstance } from "@root/common/scheduler/agenda";
import { TaskHandlerTypes, TaskParameters } from "@root/common/scheduler/@types/Tasks";
Expand All @@ -19,6 +21,15 @@ import {
PooledTaskResult
} from "../services/generators/text/PooledTextGeneratorService";

const MIN_SUMMARY_MESSAGE_COUNT = 10;

interface TaskContext {
groupId: string;
sessionId: string;
latestMessageTimestamp: number;
messageCount: number;
}

/**
* AI 摘要任务处理器
* 负责对群聊消息进行 AI 摘要生成
Expand Down Expand Up @@ -65,14 +76,9 @@ export class AISummarizeTaskHandler {

await ctxBuilder.init();

// 任务上下文类型定义
interface TaskContext {
groupId: string;
sessionId: string;
}

// 收集所有需要处理的任务
const allTasks: PooledTask<TaskContext>[] = [];
const activeSessionGraceMs = config.preprocessors.TimeoutSplitter.timeoutInMinutes * 60 * 1000;

for (const groupId of attrs.groupIds) {
/* 1. 获取指定时间范围内的消息 */
Expand Down Expand Up @@ -102,44 +108,104 @@ export class AISummarizeTaskHandler {
for (const msg of msgs) {
const { sessionId } = msg;

// 如果 sessionId 已经被生成过摘要，跳过
if (!(await this.agcDbAccessService.isSessionIdSummarized(sessionId))) {
if (!sessions[sessionId]) {
sessions[sessionId] = [];
}
sessions[sessionId].push(msg);
if (!sessions[sessionId]) {
sessions[sessionId] = [];
}
sessions[sessionId].push(msg);
}
if (Object.keys(sessions).length === 0) {
this.LOGGER.info(`群 ${groupId} 在指定时间范围内无消息，跳过`);
continue;
}
// 考虑到最后一个session可能正在发生，还没有闭合，因此需要删掉

// 最新 session 在静默时间不足时先跳过，避免把仍在发生的对话截断。
// 网络恢复补跑需要尽快追上断网期间遗漏的消息，因此允许调用方显式跳过这层保护。
const newestSessionId = msgs[msgs.length - 1].sessionId;
const newestSessionMessages = sessions[newestSessionId];

delete sessions[newestSessionId];
this.LOGGER.debug(`删掉了最后一个sessionId为 ${newestSessionId} 的session`);
this.LOGGER.info(`分组完成，共 ${Object.keys(sessions).length} 个需要处理的session`);
if (newestSessionMessages) {
const newestSessionLatestTimestamp =
newestSessionMessages[newestSessionMessages.length - 1].timestamp;
const idleTime = attrs.endTimeStamp - newestSessionLatestTimestamp;

// 3. 删掉消息量不够的session
if (attrs.ignoreActiveSessionGrace) {
this.LOGGER.info(
`已启用网络恢复补跑模式，最新 session ${newestSessionId} 即使静默时间不足也纳入摘要`
);
} else if (idleTime < activeSessionGraceMs) {
delete sessions[newestSessionId];
this.LOGGER.debug(
`最新 session ${newestSessionId} 静默时间不足 ${Math.ceil(activeSessionGraceMs / 60000)} 分钟，暂不摘要`
);
} else {
this.LOGGER.debug(`最新 session ${newestSessionId} 已静默足够久，纳入摘要`);
}
}
this.LOGGER.info(`分组完成，共 ${Object.keys(sessions).length} 个候选 session`);

const sessionMessagesToSummarize: Record<string, ProcessedChatMessageWithRawMessage[]> = {};
const sessionDigestMetadata: Record<
string,
{
latestMessageTimestamp: number;
messageCount: number;
}
> = {};

// 3. 过滤掉摘要已经覆盖的 session，并只保留新增消息片段
for (const sessionId in sessions) {
if (sessions[sessionId].length <= 10) {
const fullSessionMessages =
await this.imDbAccessService.getProcessedChatMessagesBySessionId(sessionId);

if (fullSessionMessages.length === 0) {
this.LOGGER.warning(`session ${sessionId} 未找到完整消息，跳过`);
continue;
}

const latestMessageTimestamp =
fullSessionMessages[fullSessionMessages.length - 1].timestamp;
const messageCount = fullSessionMessages.length;

if (
await this.agcDbAccessService.isSessionDigestFresh(
sessionId,
latestMessageTimestamp,
messageCount
)
) {
this.LOGGER.info(`session ${sessionId} 已经摘要到最新消息，跳过`);
continue;
}

const coverage = await this.agcDbAccessService.getSessionDigestCoverage(sessionId);
const messagesToSummarize = this._getMessagesToSummarizeByCoverage(
fullSessionMessages,
coverage
);

if (messagesToSummarize.length <= MIN_SUMMARY_MESSAGE_COUNT) {
this.LOGGER.warning(
`session ${sessionId} 消息数量不足，消息数量为${sessions[sessionId].length}，跳过`
`session ${sessionId} 新增可摘要消息数量不足，消息数量为 ${messagesToSummarize.length}，跳过`
);
delete sessions[sessionId];
continue;
}

sessionMessagesToSummarize[sessionId] = messagesToSummarize;
sessionDigestMetadata[sessionId] = {
latestMessageTimestamp,
messageCount
};
}

/* 4. 构建任务列表 */
for (const sessionId in sessions) {
for (const sessionId in sessionMessagesToSummarize) {
this.LOGGER.info(
`准备处理session ${sessionId} ，该session内共 ${sessions[sessionId].length} 条消息`
`准备处理 session ${sessionId}，本次新增可摘要消息共 ${sessionMessagesToSummarize[sessionId].length} 条`
);

// 构建上下文
const ctx = await ctxBuilder.buildCtx(
sessions[sessionId],
sessionMessagesToSummarize[sessionId],
config.groupConfigs[groupId].groupIntroduction
);

Expand All @@ -148,13 +214,29 @@ export class AISummarizeTaskHandler {
allTasks.push({
input: ctx,
modelNames: config.groupConfigs[groupId].aiModels,
context: { groupId, sessionId },
context: {
groupId,
sessionId,
latestMessageTimestamp: sessionDigestMetadata[sessionId].latestMessageTimestamp,
messageCount: sessionDigestMetadata[sessionId].messageCount
},
checkJsonFormat: true
});
}
}

this.LOGGER.info(`共收集到 ${allTasks.length} 个任务，开始并行处理（并行度=5）`);
this.LOGGER.info(
`共收集到 ${allTasks.length} 个任务，开始并行处理（并行度=${config.ai.maxConcurrentRequests}）`
);

if (allTasks.length === 0) {
this.LOGGER.info("没有需要生成摘要的 session，任务完成");
pooledTextGeneratorService.dispose();
ctxBuilder.dispose();
this.LOGGER.success(`🥳任务完成: ${job.attrs.name}`);

return;
}

// 并行处理所有任务，每个任务完成时回调
let completedCount = 0;
Expand All @@ -164,7 +246,7 @@ export class AISummarizeTaskHandler {
async (result: PooledTaskResult<TaskContext>) => {
await job.touch(); // 保证任务存活
completedCount++;
const { sessionId } = result.context;
const { sessionId, latestMessageTimestamp, messageCount } = result.context;

if (!result.isSuccess) {
this.LOGGER.error(
Expand All @@ -189,7 +271,12 @@ export class AISummarizeTaskHandler {
this.LOGGER.warning(
`session ${sessionId} 生成摘要长度过短，长度为 ${resultStr.length}，跳过`
);
console.log(resultStr);

return;
}

if (results.length === 0) {
this.LOGGER.warning(`session ${sessionId} 生成摘要为空，跳过`);

return;
}
Expand All @@ -203,8 +290,15 @@ export class AISummarizeTaskHandler {
Object.assign(resultItem, { updateTime: Date.now() });
}

// 存储摘要结果
await this.agcDbAccessService.storeAIDigestResults(results as AIDigestResult[]);
// 存储摘要结果，并记录本次摘要覆盖到的消息范围
await this.agcDbAccessService.storeAIDigestResultsWithSessionMetadata(
sessionId,
results as AIDigestResult[],
{
summarizedUntil: latestMessageTimestamp,
summarizedMessageCount: messageCount
}
);
this.LOGGER.success(`session ${sessionId} 存储摘要成功！`);
} catch (error) {
this.LOGGER.error(
Expand All @@ -226,4 +320,31 @@ export class AISummarizeTaskHandler {
}
);
}

/**
* 根据摘要覆盖范围切出本次需要摘要的消息
* @param messages 完整 session 消息
* @param coverage 摘要覆盖范围
* @returns 本次需要摘要的消息
*/
private _getMessagesToSummarizeByCoverage(
messages: ProcessedChatMessageWithRawMessage[],
coverage: SessionDigestCoverage | null
): ProcessedChatMessageWithRawMessage[] {
if (!coverage) {
return messages;
}

const messagesAfterTimestamp = messages.filter(msg => msg.timestamp > coverage.summarizedUntil);

if (messagesAfterTimestamp.length > 0) {
return messagesAfterTimestamp;
}

if (coverage.summarizedMessageCount !== null && coverage.summarizedMessageCount < messages.length) {
return messages.slice(coverage.summarizedMessageCount);
}

return [];
}
}
32 changes: 24 additions & 8 deletions applications/ai-model/src/tasks/InterestScore.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -118,20 +118,36 @@ export class InterestScoreTaskHandler {
})
);

const batchSize = config.ai.embedding.batchSize;
const totalBatchCount = Math.ceil(filteredDigestResults.length / batchSize);

// 构建所有话题详情文本
const topics = filteredDigestResults.map(
digestResult => `话题：${digestResult.topic} 正文内容：${digestResult.detail}`
);

// 批量获取所有话题的分数
await job.touch(); // 保证任务存活
const scores = await rater.scoreTopics(argArr, topics);
for (let i = 0; i < filteredDigestResults.length; i += batchSize) {
const currentBatchDigestResults = filteredDigestResults.slice(i, i + batchSize);
const currentBatchTopics = topics.slice(i, i + batchSize);
const currentBatchIndex = Math.floor(i / batchSize) + 1;

this.LOGGER.info(
`处理兴趣度评分批次 ${currentBatchIndex}/${totalBatchCount}，当前批次共 ${currentBatchTopics.length} 条`
);

await job.touch();
const scores = await rater.scoreTopics(argArr, currentBatchTopics);

for (let j = 0; j < currentBatchDigestResults.length; j++) {
await this.interestScoreDbAccessService.storeInterestScoreResult(
currentBatchDigestResults[j].topicId,
scores[j]
);
}

// 存储所有分数结果
for (let i = 0; i < filteredDigestResults.length; i++) {
await this.interestScoreDbAccessService.storeInterestScoreResult(
filteredDigestResults[i].topicId,
scores[i]
await job.touch();
this.LOGGER.info(
`兴趣度评分批次 ${currentBatchIndex}/${totalBatchCount} 已写入 ${currentBatchDigestResults.length} 条结果`
);
}

Expand Down
27 changes: 27 additions & 0 deletions applications/data-provider/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
- **消息解析**：解析protobuf格式的消息内容
- **消息格式化**：将原始消息转换为结构化数据
- **数据提供**：通过统一的接口向其他服务提供消息数据
- **群文件获取**：可选接入 OneBot/NapCat HTTP API，列出群文件并下载到本地目录

## 技术栈

Expand All @@ -21,6 +22,7 @@ src/
├── di/ # 依赖注入容器
├── providers/ # 数据提供者实现
│ ├── contracts/ # 提供者接口定义
│ ├── OneBotProvider/ # OneBot/NapCat 群文件提供者
│ └── QQProvider/ # QQ消息提供者
│ ├── parsers/ # protobuf消息解析器
│ ├── @types/ # 类型定义
Expand All @@ -36,6 +38,31 @@ QQ Provider负责从QQ数据库中读取消息并进行解析：
- 处理消息正文、图片、语音等多种消息类型
- 使用protobuf解析消息内容

## OneBot 文件 Provider 说明

OneBot 文件 Provider 通过 OneBot/NapCat HTTP API 获取群文件能力，适合后续构建群文件知识库：

- 支持 `get_group_root_files`、`get_group_file_list`、`get_group_files` 多种文件列表接口
- 支持 `get_group_file_url`、`get_file_url` 获取临时下载链接
- 下载文件时按群号保存到本地目录，并清洗危险文件名字符
- 该 Provider 不会自动接入当前聊天消息 Pipeline，需要由后续群文件同步任务显式调用

配置示例：

```json
{
"dataProviders": {
"OneBot": {
"enabled": true,
"baseURL": "http://127.0.0.1:3000",
"accessToken": "",
"downloadDirectory": "/path/to/group-files",
"requestTimeoutMs": 30000
}
}
}
```

## 开发命令

```bash
Expand Down
17 changes: 17 additions & 0 deletions applications/data-provider/src/di/container.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ import { container } from "tsyringe";

import { ProvideDataTaskHandler } from "../tasks/ProvideDataTask";
import { QQProvider } from "../providers/QQProvider/QQProvider";
import { OneBotFileProvider } from "../providers/OneBotProvider/OneBotFileProvider";

import { DATA_PROVIDER_TOKENS } from "./tokens";

Expand All @@ -26,6 +27,22 @@ export function getQQProvider(): QQProvider {
return container.resolve<QQProvider>(DATA_PROVIDER_TOKENS.QQProvider);
}

/**
* 注册 OneBotFileProvider 到 DI 容器
*/
export function registerOneBotFileProvider(): void {
container.register(DATA_PROVIDER_TOKENS.OneBotFileProvider, { useClass: OneBotFileProvider });
}

/**
* 从 DI 容器获取 OneBotFileProvider 实例
* 每次调用返回新实例（非单例）
* @returns OneBotFileProvider 实例
*/
export function getOneBotFileProvider(): OneBotFileProvider {
return container.resolve<OneBotFileProvider>(DATA_PROVIDER_TOKENS.OneBotFileProvider);
}

/**
* 注册任务处理器到 DI 容器
*/
Expand Down
4 changes: 3 additions & 1 deletion applications/data-provider/src/di/tokens.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,5 +12,7 @@ export const DATA_PROVIDER_TOKENS = {
/** 数据提供任务处理器 */
ProvideDataTaskHandler: Symbol.for("ProvideDataTaskHandler"),
/** QQ 消息提供者 */
QQProvider: Symbol.for("QQProvider")
QQProvider: Symbol.for("QQProvider"),
/** OneBot/NapCat 群文件提供者 */
OneBotFileProvider: Symbol.for("OneBotFileProvider")
} as const;
Loading