mirror of
https://github.com/kuaifan/dootask.git
synced 2026-07-22 14:08:24 +00:00
fix(search): 评审修复——重建自愈加探针表与锁内复检，向量模型变更自动重建
- 重建前用探针表验证 ai 端点，探不通不动真表不写 marker（下次自愈重试） - 锁内复检 marker，消除并发进程重复重建窗口 - 检测 embedding 模型变更（vector:model 对比查询侧缓存）自动触发重建 - Ihttp 返回体 code 非 0/200 时视为失败（401/500 不再被当成功） - 查询向量缓存键掺入模型指纹，换模型后不再命中旧维度缓存 - 批量写入按行数+字节双预算分块、块内延迟渲染 SQL，降低峰值内存 - user 同步加脏检查，字段未变不重复触发向量重算 - 日志截断超长 SQL；文档与注释同步 KB_INGEST_TOKEN 退役、截断长度对齐
This commit is contained in:
kuaifan
parent
746337c2f6
commit
f8eaab5a50
@ -916,10 +916,6 @@ class AI
|
||||
return Base::retSuccess("success", []);
|
||||
}
|
||||
|
||||
$host = config('dootask.ai_host', 'ai');
|
||||
$port = (int) config('dootask.ai_port', 5001);
|
||||
$url = "http://{$host}:{$port}/embeddings";
|
||||
|
||||
$post = json_encode(["input" => $texts]);
|
||||
$headers = [
|
||||
'Content-Type' => 'application/json',
|
||||
@ -927,16 +923,30 @@ class AI
|
||||
];
|
||||
$timeout = $count > 1 ? 120 : 30;
|
||||
|
||||
$res = Ihttp::ihttp_request($url, $post, $headers, $timeout);
|
||||
$res = Ihttp::ihttp_request(self::embeddingsUrl(), $post, $headers, $timeout);
|
||||
if (Base::isError($res)) {
|
||||
return Base::retError("Embedding 接口请求失败", $res);
|
||||
}
|
||||
|
||||
$resData = Base::json2array($res['data']);
|
||||
|
||||
// 先识别端点的业务错误：Ihttp 对 401/500 等带响应体的状态同样返回成功标志，
|
||||
// 必须按响应里的 code 字段判错，否则鉴权错配等故障只会报"格式错误"难以定位
|
||||
$resCode = intval($resData['code'] ?? 0);
|
||||
if ($resCode !== 0 && $resCode !== 200) {
|
||||
return Base::retError("Embedding 接口错误 [{$resCode}]: " . ($resData['error'] ?? 'unknown'), $resData);
|
||||
}
|
||||
|
||||
if (empty($resData['data']) || !is_array($resData['data'])) {
|
||||
return Base::retError("Embedding 接口返回数据格式错误", $resData);
|
||||
}
|
||||
|
||||
// 记录当前实际生效的向量模型（查询缓存键与模型变化检测依赖它）
|
||||
$model = (string) ($resData['model'] ?? '');
|
||||
if ($model !== '' && Cache::get('ai:embedding_model') !== $model) {
|
||||
Cache::forever('ai:embedding_model', $model);
|
||||
}
|
||||
|
||||
// 按 index 回填，保证与输入顺序对齐，缺失位置留 []
|
||||
$vectors = array_fill(0, $count, []);
|
||||
foreach ($resData['data'] as $item) {
|
||||
@ -953,6 +963,19 @@ class AI
|
||||
return Base::retSuccess("success", $vectors);
|
||||
}
|
||||
|
||||
/**
|
||||
* ai 插件 /embeddings 端点地址
|
||||
*
|
||||
* 查询侧（本类）与 Manticore 表定义（ManticoreBase::vectorColumnDDL）共用的唯一来源，
|
||||
* 两侧必须指向同一端点，否则查询向量与存量向量来自不同模型导致语义搜索错乱。
|
||||
*/
|
||||
public static function embeddingsUrl(): string
|
||||
{
|
||||
$host = config('dootask.ai_host', 'ai');
|
||||
$port = (int) config('dootask.ai_port', 5001);
|
||||
return "http://{$host}:{$port}/embeddings";
|
||||
}
|
||||
|
||||
/**
|
||||
* 通过 ai 插件的免费向量模型获取文本的 Embedding 向量
|
||||
*
|
||||
@ -973,8 +996,10 @@ class AI
|
||||
// 截断过长的文本（约 32K 字符）
|
||||
$text = mb_substr($text, 0, 30000);
|
||||
|
||||
// 缓存键换命名空间（embeddingV2）：避免历史 1536 维缓存污染新的向量维度
|
||||
$cacheKey = "embeddingV2::" . md5($text);
|
||||
// 缓存键带上当前生效的模型标识：切换向量模型后旧查询缓存自动失效，
|
||||
// 避免最长 7 天内用旧模型向量去搜新模型索引（模型未知时为空串）
|
||||
$model = (string) Cache::get('ai:embedding_model', '');
|
||||
$cacheKey = "embeddingV2::" . md5($model . '|' . $text);
|
||||
if ($noCache) {
|
||||
Cache::forget($cacheKey);
|
||||
}
|
||||
|
||||
@ -110,7 +110,7 @@ class Apps
|
||||
* 获取（必要时生成并持久化）应用的独立密钥 APP_SECRET。
|
||||
*
|
||||
* 与全局 APP_KEY 不同，APP_SECRET 每个已安装应用独立、唯一，持久化在应用自身的
|
||||
* docker/appstore/config/{appid}/config.yml（与 KB_INGEST_TOKEN 等每应用参数同源），
|
||||
* docker/appstore/config/{appid}/config.yml（与其它每应用安装参数同源），
|
||||
* 由 appstore 安装链路按内置 compose 变量 APP_SECRET 注入插件容器。
|
||||
* 此处主程序侧负责生成/持久化与校验；首次需要时若不存在则惰性生成，保证主程序可独立验证。
|
||||
*
|
||||
|
||||
@ -114,7 +114,9 @@ class ManticoreBase
|
||||
");
|
||||
|
||||
$expected = self::schemaMarker();
|
||||
if (self::kvGetPdo($pdo, 'vector:schema') === $expected && self::allVectorTablesExist($pdo)) {
|
||||
if (self::kvGetPdo($pdo, 'vector:schema') === $expected
|
||||
&& self::allVectorTablesExist($pdo)
|
||||
&& !self::embeddingModelChanged($pdo)) {
|
||||
return;
|
||||
}
|
||||
|
||||
@ -124,41 +126,52 @@ class ManticoreBase
|
||||
return;
|
||||
}
|
||||
try {
|
||||
// 拿到锁时可能另一进程刚完成重建（marker 写入在锁内），先复检避免重复重建
|
||||
if (self::kvGetPdo($pdo, 'vector:schema') === $expected
|
||||
&& self::allVectorTablesExist($pdo)
|
||||
&& !self::embeddingModelChanged($pdo)) {
|
||||
return;
|
||||
}
|
||||
|
||||
// 先用一次性探针表验证 ai 端点可用（CREATE 会向端点探测维度、未就绪则抛错），
|
||||
// 通过后才 DROP 现有表——避免 ai 未就绪时销毁旧索引后建不回来
|
||||
$pdo->exec("DROP TABLE IF EXISTS _schema_probe");
|
||||
$pdo->exec("CREATE TABLE _schema_probe (t TEXT, " . self::vectorColumnDDL('t') . ")");
|
||||
$pdo->exec("DROP TABLE IF EXISTS _schema_probe");
|
||||
|
||||
foreach (self::vectorTableDDLs() as $table => $ddl) {
|
||||
$pdo->exec("DROP TABLE IF EXISTS {$table}");
|
||||
// CREATE 时引擎会调用 ai 端点探测向量维度，ai 未就绪则抛错
|
||||
$pdo->exec($ddl);
|
||||
}
|
||||
|
||||
self::resetSyncPointersPdo($pdo);
|
||||
// 清理旧向量管道遗留键（vector:dim 与 vector:*LastId 指针）
|
||||
$legacy = ["'vector:dim'"];
|
||||
foreach (self::VECTOR_TABLES as $t) {
|
||||
$legacy[] = "'vector:manticore" . ucfirst($t) . "LastId'";
|
||||
}
|
||||
$pdo->exec("DELETE FROM key_values WHERE k IN (" . implode(',', $legacy) . ")");
|
||||
self::rememberEmbeddingModel($pdo);
|
||||
// marker 最后写入且在锁内：写入即代表重建完整成功
|
||||
self::kvSetPdo($pdo, 'vector:schema', $expected);
|
||||
Log::info("Manticore vector tables rebuilt for auto-embeddings schema {$expected}");
|
||||
} catch (\Throwable $e) {
|
||||
// 建表失败（如 ai 插件未就绪/未升级）：不写 marker，下个进程重试，可自愈
|
||||
// 重建失败（如 ai 插件未就绪/未升级）：不写 marker，下个进程重试，可自愈
|
||||
Log::error('Manticore schema rebuild failed: ' . $e->getMessage());
|
||||
return;
|
||||
} finally {
|
||||
$lock->release();
|
||||
}
|
||||
|
||||
self::resetSyncPointersPdo($pdo);
|
||||
// 清理旧向量管道遗留键（vector:dim 与 vector:*LastId 指针）
|
||||
$legacy = ["'vector:dim'"];
|
||||
foreach (self::VECTOR_TABLES as $t) {
|
||||
$legacy[] = "'vector:manticore" . ucfirst($t) . "LastId'";
|
||||
}
|
||||
$pdo->exec("DELETE FROM key_values WHERE k IN (" . implode(',', $legacy) . ")");
|
||||
self::kvSetPdo($pdo, 'vector:schema', $expected);
|
||||
Log::info("Manticore vector tables rebuilt for auto-embeddings schema {$expected}");
|
||||
} catch (\Throwable $e) {
|
||||
Log::error('Manticore initializeTables failed: ' . $e->getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* ai 插件向量化端点地址（引擎 Auto Embeddings 与主程序查询侧共用）
|
||||
* ai 插件向量化端点地址（唯一来源在 AI::embeddingsUrl，查询侧与表定义共用）
|
||||
*/
|
||||
private static function embeddingsApiUrl(): string
|
||||
{
|
||||
$host = config('dootask.ai_host', 'ai');
|
||||
$port = (int) config('dootask.ai_port', 5001);
|
||||
return "http://{$host}:{$port}/embeddings";
|
||||
return AI::embeddingsUrl();
|
||||
}
|
||||
|
||||
/**
|
||||
@ -180,6 +193,42 @@ class ManticoreBase
|
||||
. self::embeddingsApiUrl() . '|' . hash('sha256', (string) config('app.key')));
|
||||
}
|
||||
|
||||
/**
|
||||
* 检测 ai 插件实际生效的向量模型是否与建表时不一致（不一致需重建，否则维度可能不匹配）。
|
||||
*
|
||||
* 实际模型（EMBEDDING_MODEL env）对主程序不可见，由查询侧在成功请求后
|
||||
* 写入缓存 ai:embedding_model；建表时的模型记录在 key_values 的 vector:model。
|
||||
* 任一侧未知时不触发（返回 false），已知且存量缺失时顺手补记。
|
||||
*/
|
||||
private static function embeddingModelChanged(PDO $pdo): bool
|
||||
{
|
||||
$live = (string) Cache::get('ai:embedding_model', '');
|
||||
if ($live === '') {
|
||||
return false;
|
||||
}
|
||||
$stored = self::kvGetPdo($pdo, 'vector:model');
|
||||
if ($stored === null || $stored === '') {
|
||||
// 旧部署/首次：补记当前模型，不触发重建
|
||||
self::kvSetPdo($pdo, 'vector:model', $live);
|
||||
return false;
|
||||
}
|
||||
return $stored !== $live;
|
||||
}
|
||||
|
||||
/**
|
||||
* 重建成功后记录当前生效的向量模型（优先取查询侧维护的缓存值）
|
||||
*/
|
||||
private static function rememberEmbeddingModel(PDO $pdo): void
|
||||
{
|
||||
$live = (string) Cache::get('ai:embedding_model', '');
|
||||
if ($live !== '') {
|
||||
self::kvSetPdo($pdo, 'vector:model', $live);
|
||||
} else {
|
||||
// 未知则清掉存量，待查询侧探得后由 embeddingModelChanged 补记
|
||||
$pdo->exec("DELETE FROM key_values WHERE k = 'vector:model'");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 生成 Auto Embeddings 向量列定义（引擎按 FROM 字段自动生成/更新向量）
|
||||
*
|
||||
@ -419,13 +468,17 @@ class ManticoreBase
|
||||
*/
|
||||
public function executeRaw(string $sql): bool
|
||||
{
|
||||
// 日志上下文只保留 SQL 前 2KB：多行批量语句可达数 MB，完整写入会淹没日志
|
||||
$sqlPreview = strlen($sql) > 2048
|
||||
? substr($sql, 0, 2048) . ' ...[+' . (strlen($sql) - 2048) . ' bytes]'
|
||||
: $sql;
|
||||
return $this->runWithRetry(
|
||||
function (PDO $pdo) use ($sql) {
|
||||
$pdo->exec($sql);
|
||||
return true;
|
||||
},
|
||||
false,
|
||||
['sql' => $sql]
|
||||
['sql' => $sqlPreview]
|
||||
);
|
||||
}
|
||||
|
||||
@ -2107,17 +2160,19 @@ class ManticoreBase
|
||||
$pk = $config['pk'];
|
||||
$instance = new self();
|
||||
|
||||
// 预构建每行的内联值，剔除无主键行
|
||||
// 剔除无主键行；按内容长度估算分块（不在此渲染 SQL，块内惰性渲染以压低内存峰值）
|
||||
$pending = [];
|
||||
$fieldListRef = null;
|
||||
foreach ($rows as $row) {
|
||||
if (($row[$pk] ?? 0) <= 0) {
|
||||
continue;
|
||||
}
|
||||
[$fieldList, $valueList] = $instance->buildRowValues($config, $row);
|
||||
$fieldListRef = $fieldList;
|
||||
$valuesSql = '(' . implode(', ', $valueList) . ')';
|
||||
$pending[] = ['pk' => $row[$pk], 'sql' => $valuesSql, 'bytes' => strlen($valuesSql), 'row' => $row];
|
||||
$bytes = 64;
|
||||
foreach ($row as $value) {
|
||||
if (is_string($value)) {
|
||||
$bytes += strlen($value);
|
||||
}
|
||||
}
|
||||
$pending[] = ['pk' => $row[$pk], 'bytes' => $bytes, 'row' => $row];
|
||||
}
|
||||
if (empty($pending)) {
|
||||
return 0;
|
||||
@ -2143,9 +2198,20 @@ class ManticoreBase
|
||||
|
||||
$successCount = 0;
|
||||
foreach ($chunks as $chunk) {
|
||||
// 块内渲染，执行后即释放，内存峰值 = 原始行 + 单块 SQL
|
||||
$fieldListRef = null;
|
||||
$valuesSql = [];
|
||||
foreach ($chunk as $item) {
|
||||
[$fieldList, $valueList] = $instance->buildRowValues($config, $item['row']);
|
||||
$fieldListRef = $fieldList;
|
||||
$valuesSql[] = '(' . implode(', ', $valueList) . ')';
|
||||
}
|
||||
$sql = "REPLACE INTO {$table} (" . implode(', ', $fieldListRef) . ") VALUES "
|
||||
. implode(', ', array_column($chunk, 'sql'));
|
||||
if ($instance->executeRaw($sql)) {
|
||||
. implode(', ', $valuesSql);
|
||||
unset($valuesSql);
|
||||
$ok = $instance->executeRaw($sql);
|
||||
unset($sql);
|
||||
if ($ok) {
|
||||
$successCount += count($chunk);
|
||||
ManticoreSyncFailure::removeSuccessBatch($type, array_column($chunk, 'pk'), 'sync');
|
||||
} else {
|
||||
|
||||
@ -37,7 +37,9 @@ class ManticoreFile
|
||||
public const SEARCHABLE_TYPES = ['document', 'word', 'excel', 'ppt', 'txt', 'md', 'text', 'code'];
|
||||
|
||||
/**
|
||||
* 最大内容长度（字符）- 提取后的文本内容限制
|
||||
* 最大内容长度（字符）- 提取后的文本内容限制（服务全文检索范围）。
|
||||
* 注意：向量化输入在 ai 插件侧另有 30000 字符上限（main.py _EMBEDDING_INPUT_MAX_CHARS），
|
||||
* 超出部分只参与全文检索、不参与语义向量。
|
||||
*/
|
||||
public const MAX_CONTENT_LENGTH = 100000; // 100K 字符
|
||||
|
||||
|
||||
@ -37,7 +37,8 @@ class ManticoreMsg
|
||||
public const INDEXABLE_TYPES = ['text', 'file', 'record', 'meeting', 'vote'];
|
||||
|
||||
/**
|
||||
* 最大内容长度（字符）
|
||||
* 最大内容长度（字符）。向量化输入在 ai 插件侧另有 30000 字符上限
|
||||
* （main.py _EMBEDDING_INPUT_MAX_CHARS），超出部分只参与全文检索
|
||||
*/
|
||||
public const MAX_CONTENT_LENGTH = 50000; // 50K 字符
|
||||
|
||||
|
||||
@ -41,7 +41,8 @@ use Illuminate\Support\Facades\Log;
|
||||
class ManticoreTask
|
||||
{
|
||||
/**
|
||||
* 最大内容长度（字符）
|
||||
* 最大内容长度（字符）。向量化输入在 ai 插件侧另有 30000 字符上限
|
||||
* （main.py _EMBEDDING_INPUT_MAX_CHARS），超出部分只参与全文检索
|
||||
*/
|
||||
public const MAX_CONTENT_LENGTH = 50000; // 50K 字符
|
||||
|
||||
|
||||
@ -143,7 +143,13 @@ class ManticoreUser
|
||||
}
|
||||
|
||||
try {
|
||||
return ManticoreBase::upsertUserVector(self::buildRow($user));
|
||||
$row = self::buildRow($user);
|
||||
// 脏检查：与已索引行完全一致则跳过。标签点赞/识别等高频事件经常不改变
|
||||
// Top-10 标签文本，跳过可省一次真实的向量化调用与整行重写
|
||||
if (self::rowUnchanged($row)) {
|
||||
return true;
|
||||
}
|
||||
return ManticoreBase::upsertUserVector($row);
|
||||
} catch (\Exception $e) {
|
||||
Log::error('Manticore user sync error: ' . $e->getMessage(), [
|
||||
'userid' => $user->userid,
|
||||
@ -153,6 +159,26 @@ class ManticoreUser
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 判断待写入行与当前已索引行是否完全一致（文本字段逐一比较）
|
||||
*/
|
||||
private static function rowUnchanged(array $row): bool
|
||||
{
|
||||
$existing = (new ManticoreBase())->queryOne(
|
||||
"SELECT nickname, email, profession, tags, introduction FROM user_vectors WHERE userid = ?",
|
||||
[$row['userid']]
|
||||
);
|
||||
if (!$existing) {
|
||||
return false;
|
||||
}
|
||||
foreach (['nickname', 'email', 'profession', 'tags', 'introduction'] as $field) {
|
||||
if ((string) ($existing[$field] ?? '') !== (string) $row[$field]) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* 构建用户索引行数据（含标签 Top 10）
|
||||
*
|
||||
|
||||
@ -67,10 +67,12 @@ volumes:
|
||||
内容同步机制：容器每次启动按文件 hash 对账（reconcile），自动增量收敛新增/变更/删除的 markdown——客户实例更新 DooTask 后重启插件容器即生效。需要免重启即时生效时手动触发：
|
||||
```bash
|
||||
curl -X POST 'http://ai-service/kb/reindex' \
|
||||
-H "X-Ingest-Token: $KB_INGEST_TOKEN" \
|
||||
-H "Authorization: Bearer $APP_KEY" \
|
||||
-d '{"mode":"reconcile"}'
|
||||
```
|
||||
|
||||
（`$APP_KEY` 为主程序全局密钥；ai 插件 0.5.8 起以它取代旧的 `KB_INGEST_TOKEN` 鉴权。）
|
||||
|
||||
## 用检索打点与用户反馈数据迭代内容
|
||||
|
||||
主程序记录了两类质量数据（mariadb，表前缀以实际 `DB_PREFIX` 为准，下例用 `pre_`）：
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user