512 lines
28 KiB
Python
512 lines
28 KiB
Python
"""Restartable knowledge processing worker. Run separately from the HTTP server."""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import json
|
||
import time
|
||
import threading
|
||
from collections import Counter
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
from pathlib import Path
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
import admin_backend
|
||
from archive_store import json_text, new_id, utc_now
|
||
from knowledge_processing import classify, feed, finish
|
||
from knowledge_model_output import ModelDiscard, ModelOutputError, normalize_output, original_draft, split_batch
|
||
from knowledge_models import ModelCallError
|
||
from knowledge_store import AUTO_RUN, KnowledgeStore, job_scope, load_config
|
||
|
||
# 每页读多少条消息。规则整理一步一页；模型整理一步最多读 MAX_SCAN_PAGES 页，攒够一轮并发要用的问答
|
||
# （线上 100 条消息只能切出一两段问答，只读一页的话并发和合并都用不上）。
|
||
SCAN_BATCH = 200
|
||
MAX_SCAN_PAGES = 25
|
||
LEASE_SECONDS = 300
|
||
# 整理批次默认一次请求合并 8 段问答；旧版单账号任务保持一段一请求。
|
||
DEFAULT_BATCH = 8
|
||
# 超时、限流、5xx 这类临时错误：退避后自动重试，连续这么多次还不行才停下来等人处理。
|
||
MAX_TRANSIENT = 6
|
||
|
||
|
||
def batch_size(options) -> int:
|
||
value = options.get('model_batch')
|
||
if value is None:
|
||
value = DEFAULT_BATCH if options.get('run_id') else 1
|
||
return max(1, min(20, int(value)))
|
||
|
||
|
||
def enhance(candidate, database, options=None, *, tenant_id="default", task_id=""):
|
||
"""Optional extraction via the chosen model, using only redacted evidence.
|
||
|
||
每次调用自己解析出口、自己开事件循环：并发时各线程互不共享连接和信号量。
|
||
模型判定没有可复用知识时抛 ModelDiscard；格式不对抛 ModelOutputError；其余失败抛 ValueError。
|
||
"""
|
||
from model_usage import UsageRecorder
|
||
from knowledge_models import extract, resolve_model, safe_model_error
|
||
import httpx
|
||
options = dict(options or {})
|
||
outlet = resolve_model(database, options)
|
||
deadline = options['model_timeout_seconds']
|
||
# A separate catalog instance: task timeouts do not change interactive chat settings.
|
||
outlet.config = {**outlet.config, 'timeout_ms': deadline * 1000}
|
||
allow_discard = bool(options.get('model_filter', True))
|
||
|
||
async def run():
|
||
async with UsageRecorder(database, tenant_id=tenant_id, task_id=task_id,
|
||
purpose="knowledge") as recorder:
|
||
async with httpx.AsyncClient() as client:
|
||
return await extract(client, outlet, candidate, deadline=deadline,
|
||
allow_discard=allow_discard, usage_recorder=recorder)
|
||
try:
|
||
result, prompt_chars = asyncio.run(run())
|
||
except (httpx.HTTPError, TimeoutError) as exc:
|
||
raise ModelCallError(safe_model_error(type(exc).__name__), transient=True) from None
|
||
if result.get("error"):
|
||
raise ModelCallError(safe_model_error(result['error']), transient=is_transient(result['error']))
|
||
return normalize_output(candidate, result.get('text'), prompt_chars, allow_discard=allow_discard)
|
||
|
||
|
||
def enhance_batch(candidates, database, options=None, *, tenant_id="default", task_id=""):
|
||
"""一次请求整理几段问答，返回与 candidates 一一对应的 (结果, 值)：
|
||
ok → (草稿, 输入字数, 输出字数)；discard → ModelDiscard；format → ModelOutputError；
|
||
single → 这段没拿到结果（输出被截断、缺项、上限被拒），下一轮单独重试。请求本身失败抛 ModelCallError。"""
|
||
if len(candidates) == 1:
|
||
try:
|
||
return [('ok', enhance(candidates[0], database, options, tenant_id=tenant_id, task_id=task_id))]
|
||
except ModelDiscard as exc:
|
||
return [('discard', exc)]
|
||
except ModelOutputError as exc:
|
||
return [('format', exc)]
|
||
from model_usage import UsageRecorder
|
||
from knowledge_models import _max_tokens_rejected, extract_batch, resolve_model, safe_model_error
|
||
import httpx
|
||
options = dict(options or {})
|
||
outlet = resolve_model(database, options)
|
||
deadline = options['model_timeout_seconds']
|
||
outlet.config = {**outlet.config, 'timeout_ms': deadline * 1000}
|
||
allow_discard = bool(options.get('model_filter', True))
|
||
|
||
async def run():
|
||
async with UsageRecorder(database, tenant_id=tenant_id, task_id=task_id,
|
||
purpose="knowledge") as recorder:
|
||
async with httpx.AsyncClient() as client:
|
||
return await extract_batch(client, outlet, candidates, deadline=deadline,
|
||
allow_discard=allow_discard, usage_recorder=recorder)
|
||
try:
|
||
result, prompt_chars = asyncio.run(run())
|
||
except (httpx.HTTPError, TimeoutError) as exc:
|
||
raise ModelCallError(safe_model_error(type(exc).__name__), transient=True) from None
|
||
if result.get("error"):
|
||
if _max_tokens_rejected(result['error']):
|
||
return [('single', None)] * len(candidates)
|
||
raise ModelCallError(safe_model_error(result['error']), transient=is_transient(result['error']))
|
||
try:
|
||
parts = split_batch(result.get('text'), len(candidates))
|
||
except ModelOutputError:
|
||
return [('single', None)] * len(candidates)
|
||
share = prompt_chars // len(candidates)
|
||
outcomes = []
|
||
for index, candidate in enumerate(candidates):
|
||
raw = parts.get(str(index + 1))
|
||
if raw is None:
|
||
outcomes.append(('single', None))
|
||
continue
|
||
try:
|
||
outcomes.append(('ok', normalize_output(candidate, raw, share, allow_discard=allow_discard)))
|
||
except ModelDiscard as exc:
|
||
outcomes.append(('discard', exc))
|
||
except ModelOutputError as exc:
|
||
outcomes.append(('format', exc))
|
||
return outcomes
|
||
|
||
|
||
def is_transient(error):
|
||
from knowledge_models import is_transient as check
|
||
return check(error)
|
||
|
||
|
||
class KnowledgeWorker:
|
||
def __init__(self, store):
|
||
self.store = store
|
||
self.owner = new_id()
|
||
|
||
def claim(self):
|
||
now = time.time()
|
||
with self.store.database.connect() as db:
|
||
db.execute("BEGIN IMMEDIATE")
|
||
# 临时错误后退避中的任务（retry_after 未到）先跳过，让别的账号接着跑
|
||
job = db.execute("SELECT * FROM knowledge_job WHERE (status='queued' AND "
|
||
"COALESCE(json_extract(options_json,'$.retry_after'),0)<=?) OR "
|
||
"(status='running' AND lease_until<?) ORDER BY created_at LIMIT 1", (now, now)).fetchone()
|
||
if job is None:
|
||
return None
|
||
db.execute("UPDATE knowledge_job SET status='running',lease_owner=?,lease_until=?,updated_at=? WHERE id=?",
|
||
(self.owner, now + LEASE_SECONDS, utc_now(), job["id"]))
|
||
return dict(job)
|
||
|
||
def rows(self, job, batch_size):
|
||
options = json.loads(job["options_json"])
|
||
cursor = json.loads(job["cursor_json"])
|
||
where, args = job_scope(job["tenant_id"], options, job["cutoff_at"])
|
||
if cursor:
|
||
# 按扫描索引 (tenant, source, conversation, sent_at, id) 的顺序翻页；不指定来源的任务
|
||
# 若按会话排序，SQLite 每一批都要把整个账号的消息重排一遍。
|
||
if "source_account_id" in cursor:
|
||
where.append("(m.source_account_id,m.conversation_id,m.sent_at,m.id)>(?,?,?,?)")
|
||
args.extend([cursor["source_account_id"], cursor["conversation_id"], cursor["sent_at"], cursor["id"]])
|
||
else:
|
||
where.append("(m.conversation_id,m.sent_at,m.id)>(?,?,?)")
|
||
args.extend([cursor["conversation_id"], cursor["sent_at"], cursor["id"]])
|
||
with self.store.database.connect() as db:
|
||
rows = db.execute("""SELECT m.id,m.source_account_id,m.conversation_id,m.sender_person_id,m.message_type,
|
||
m.direction,m.sent_at,m.sent_at_epoch,v.content,v.status,v.version_no,p.display_name sender_name
|
||
FROM archive_message m JOIN archive_conversation c ON c.id=m.conversation_id
|
||
LEFT JOIN archive_person p ON p.id=m.sender_person_id
|
||
JOIN archive_message_version v ON v.message_id=m.id AND v.version_no=(
|
||
SELECT MAX(v2.version_no) FROM archive_message_version v2
|
||
WHERE v2.message_id=m.id AND v2.created_at<=?)
|
||
WHERE """ + " AND ".join(where) +
|
||
" ORDER BY m.source_account_id,m.conversation_id,m.sent_at,m.id LIMIT ?",
|
||
[job["cutoff_at"], *args, batch_size]).fetchall()
|
||
# 单聊里收发方向以归档为准：桌面端导入时已按"发送者是不是本账号"判定过。以前再要求
|
||
# 发送者身份能对上来源账号，身份没关联的账号几乎所有客服回复都被当成"方向不明"丢掉了。
|
||
return [dict(raw) for raw in rows]
|
||
|
||
def enqueue_incremental(self):
|
||
# Wait for a quiet period so a live, still-incomplete answer is not extracted.
|
||
cutoff = (datetime.now(timezone.utc) - timedelta(minutes=30)).isoformat(timespec="milliseconds")
|
||
config = load_config(self.store.database)
|
||
with self.store.database.connect() as db:
|
||
db.execute("BEGIN IMMEDIATE")
|
||
dirty = db.execute("""SELECT d.*,w.created_by FROM knowledge_dirty d JOIN knowledge_watch w
|
||
ON w.tenant_id=d.tenant_id AND w.source_account_id=d.source_account_id
|
||
WHERE w.enabled=1 AND d.changed_at<? ORDER BY d.changed_at LIMIT 1""", (cutoff,)).fetchone()
|
||
if dirty is None:
|
||
# 持续整理每个会话一个小任务，留一个月够排查，再久就清掉
|
||
old = (datetime.now(timezone.utc) - timedelta(days=30)).isoformat(timespec="milliseconds")
|
||
db.execute("DELETE FROM knowledge_job WHERE id IN (SELECT id FROM knowledge_job WHERE status='completed' "
|
||
"AND (run_id=? OR json_extract(options_json,'$.incremental') IS 1) AND updated_at<? LIMIT 200)",
|
||
(AUTO_RUN, old))
|
||
return False
|
||
options = {"source_account_id": dirty["source_account_id"], "conversation_id": dirty["conversation_id"],
|
||
"engine": "rules", "max_messages": 0, "max_model_calls": 0,
|
||
"staff_confirmed": True, "incremental": True, "run_id": AUTO_RUN}
|
||
if config.get("incremental_engine") == "model":
|
||
options.update(engine="model", model_provider_id=config.get("default_model_provider_id", ""),
|
||
model_concurrency=config.get("model_concurrency", 4),
|
||
model_batch=config.get("model_batch", DEFAULT_BATCH),
|
||
model_filter=config.get("model_filter", True), model_timeout_seconds=90)
|
||
now = utc_now()
|
||
estimate = self.store._scope_counts(db, dirty["tenant_id"], options, now, quality=False)
|
||
options.update(total_messages=estimate["total_messages"], available_messages=estimate["available_messages"],
|
||
source_name=estimate["source_name"])
|
||
db.execute("INSERT INTO knowledge_job(id,tenant_id,options_json,cutoff_at,created_by,created_at,updated_at,run_id) "
|
||
"VALUES (?,?,?,?,?,?,?,?)", (new_id(), dirty["tenant_id"], json_text(options), now,
|
||
dirty["created_by"], now, now, AUTO_RUN))
|
||
db.execute("DELETE FROM knowledge_dirty WHERE tenant_id=? AND source_account_id=? AND conversation_id=?",
|
||
(dirty["tenant_id"], dirty["source_account_id"], dirty["conversation_id"]))
|
||
return True
|
||
|
||
def _owned(self, db, job):
|
||
return db.execute("SELECT * FROM knowledge_job WHERE id=? AND status='running' AND lease_owner=?",
|
||
(job['id'], self.owner)).fetchone()
|
||
|
||
def scan(self, job, options, state, pending):
|
||
"""读一批消息，切成问答候选，连同游标一起落库（付费调用之前先保存进度）。"""
|
||
limit = options.get('max_messages') or 0
|
||
remaining = limit - job['processed'] if limit else None
|
||
size = SCAN_BATCH if remaining is None else min(SCAN_BATCH, remaining)
|
||
fetched = self.rows(job, size + 1) if size > 0 else []
|
||
rows = fetched[:size]
|
||
skipped = Counter()
|
||
for row in rows:
|
||
use, reason = classify(row)
|
||
if use == 'skip':
|
||
skipped[reason] += 1
|
||
state, emitted = feed(state, row)
|
||
pending.extend(emitted)
|
||
natural_end = size > 0 and len(fetched) <= size
|
||
if natural_end:
|
||
last = finish(state)
|
||
if last:
|
||
pending.append(last)
|
||
state = {}
|
||
done = natural_end or (remaining is not None and len(rows) >= remaining)
|
||
if done:
|
||
options['completion_reason'] = 'source_exhausted' if natural_end else 'message_limit'
|
||
state = {}
|
||
reasons = options.setdefault('skip_reasons', {})
|
||
for reason, count in skipped.items():
|
||
reasons[reason] = reasons.get(reason, 0) + count
|
||
cursor = ({k: rows[-1][k] for k in ('source_account_id', 'conversation_id', 'sent_at', 'id')}
|
||
if rows else json.loads(job['cursor_json']))
|
||
# Persist the scan and redacted candidates before any paid request.
|
||
# Pending candidates survive failure, pause and process restarts.
|
||
with self.store.database.connect() as db:
|
||
updated = db.execute("""UPDATE knowledge_job SET cursor_json=?,state_json=?,
|
||
processed=processed+?,skipped=skipped+?,options_json=?,lease_until=?,updated_at=?
|
||
WHERE id=? AND status='running' AND lease_owner=?""", (
|
||
json_text(cursor), json_text({**state, '_pending': pending, '_scan_done': done}), len(rows),
|
||
sum(skipped.values()), json_text(options), time.time() + LEASE_SECONDS, utc_now(), job['id'], self.owner))
|
||
if not updated.rowcount:
|
||
return None
|
||
# 一步里连读几页时，下一页要从这里接着读
|
||
job['cursor_json'] = json_text(cursor)
|
||
job['processed'] += len(rows)
|
||
return state, pending, done
|
||
|
||
def save_rules(self, job, options, state, pending, done):
|
||
"""规则草稿不花钱：一批一个事务存完。"""
|
||
with self.store.database.connect() as db:
|
||
db.execute('BEGIN IMMEDIATE')
|
||
if not self._owned(db, job):
|
||
return False
|
||
counts = Counter(self.store.add_draft(db, job['tenant_id'], job['id'], candidate) for candidate in pending)
|
||
options['updated_items'] = options.get('updated_items', 0) + counts['updated']
|
||
db.execute("""UPDATE knowledge_job SET state_json=?,created_items=created_items+?,duplicates=duplicates+?,
|
||
options_json=?,updated_at=? WHERE id=?""", (json_text({**state, '_pending': [], '_scan_done': done}),
|
||
counts['created'], counts[''], json_text(options), utc_now(), job['id']))
|
||
return True
|
||
|
||
def _budget(self, db, job, options, wanted):
|
||
"""这一轮还能发几次模型请求。批次有总上限时按整个批次算，旧任务按单个任务算；0 = 不限。"""
|
||
run_id = job.get('run_id') or ''
|
||
if run_id and run_id != AUTO_RUN:
|
||
run = db.execute("SELECT options_json FROM knowledge_run WHERE id=?", (run_id,)).fetchone()
|
||
cap = int(json.loads(run[0]).get('max_model_calls') or 0) if run else 0
|
||
used = db.execute("SELECT COALESCE(SUM(model_calls),0) FROM knowledge_job WHERE run_id=?", (run_id,)).fetchone()[0]
|
||
else:
|
||
cap = int(options.get('max_model_calls') or 0)
|
||
used = db.execute("SELECT model_calls FROM knowledge_job WHERE id=?", (job['id'],)).fetchone()[0]
|
||
if not cap:
|
||
return wanted
|
||
if used >= cap:
|
||
raise ValueError('模型调用预算已用完；已保存进度，请提高调用总上限（或填 0 不限）或改用规则整理后继续')
|
||
return min(wanted, cap - used)
|
||
|
||
def _call(self, job, options, group):
|
||
try:
|
||
return 'done', enhance_batch(group, self.store.database, options, tenant_id=job["tenant_id"], task_id=job["id"])
|
||
except ModelCallError as exc:
|
||
return 'error', exc
|
||
except ValueError as exc:
|
||
# 模型停用、本地校验之类：不是等一会儿就能好的
|
||
return 'error', ModelCallError(str(exc))
|
||
except Exception as exc: # noqa: BLE001 - surfaced as the job error, never the provider body
|
||
return 'error', ModelCallError(type(exc).__name__)
|
||
|
||
@staticmethod
|
||
def _groups(pending, slots, size):
|
||
"""从待整理的问答前面依次切出最多 slots 组，每组最多 size 段；要求单独重试的自成一组。"""
|
||
groups, rest = [], list(pending)
|
||
while rest and len(groups) < slots:
|
||
if size == 1 or rest[0].get('_single'):
|
||
groups.append([rest.pop(0)])
|
||
continue
|
||
group = []
|
||
while rest and len(group) < size and not rest[0].get('_single'):
|
||
group.append(rest.pop(0))
|
||
groups.append(group)
|
||
return groups
|
||
|
||
def save_model(self, job, options, state, pending, done):
|
||
"""并发调用模型整理候选问答（默认 4 路，每次请求合并若干段）；每一轮成功的结果先落库，再处理失败。"""
|
||
concurrency = max(1, min(16, int(options.get('model_concurrency') or 1)))
|
||
size = batch_size(options)
|
||
with ThreadPoolExecutor(max_workers=concurrency, thread_name_prefix='knowledge-model') as pool:
|
||
while pending:
|
||
with self.store.database.connect() as db:
|
||
db.execute('BEGIN IMMEDIATE')
|
||
if not self._owned(db, job):
|
||
return False
|
||
groups = self._groups(pending, concurrency, size)
|
||
groups = groups[:self._budget(db, job, options, len(groups))]
|
||
# Every call is bounded at <=181s, below the renewed lease.
|
||
db.execute('UPDATE knowledge_job SET model_calls=model_calls+?,lease_until=? WHERE id=?',
|
||
(len(groups), time.time() + LEASE_SECONDS, job['id']))
|
||
taken = sum(len(group) for group in groups)
|
||
results = list(pool.map(lambda group: self._call(job, options, group), groups))
|
||
error = None
|
||
with self.store.database.connect() as db:
|
||
db.execute('BEGIN IMMEDIATE')
|
||
if not self._owned(db, job):
|
||
return False
|
||
counts, chars, kept = Counter(), [0, 0], []
|
||
for group, (kind, value) in zip(groups, results):
|
||
if kind == 'error':
|
||
# The request failed: these candidates stay pending. A timed-out batch is
|
||
# retried one by one, smaller requests finish sooner.
|
||
error = error or value
|
||
split = value.transient and len(group) > 1 and '超时' in str(value)
|
||
kept.extend({**candidate, '_single': True} if split else candidate for candidate in group)
|
||
continue
|
||
for candidate, (outcome, item) in zip(group, value):
|
||
if outcome == 'single':
|
||
kept.append({**candidate, '_single': True})
|
||
continue
|
||
if outcome == 'ok':
|
||
draft, input_chars, output_chars = item
|
||
elif outcome == 'discard':
|
||
draft, input_chars, output_chars = None, item.input_chars, item.output_chars
|
||
counts['filtered'] += 1
|
||
options['model_last_filtered'] = item.reason
|
||
else:
|
||
# A bad response affects this draft only.
|
||
draft, input_chars, output_chars = original_draft(candidate, item), item.input_chars, item.output_chars
|
||
options['model_format_fallbacks'] = options.get('model_format_fallbacks', 0) + 1
|
||
options['model_last_warning'] = str(item)
|
||
chars[0] += input_chars
|
||
chars[1] += output_chars
|
||
if draft is not None:
|
||
draft.pop('_single', None)
|
||
counts[self.store.add_draft(db, job['tenant_id'], job['id'], draft, replace=True)] += 1
|
||
pending = kept + pending[taken:]
|
||
options['updated_items'] = options.get('updated_items', 0) + counts['updated']
|
||
options['model_filtered'] = options.get('model_filtered', 0) + counts['filtered']
|
||
if error is None:
|
||
options.pop('retry_after', None)
|
||
options['transient_failures'] = 0
|
||
db.execute("""UPDATE knowledge_job SET state_json=?,created_items=created_items+?,duplicates=duplicates+?,
|
||
input_chars=input_chars+?,output_chars=output_chars+?,options_json=?,updated_at=? WHERE id=?""",
|
||
(json_text({**state, '_pending': pending, '_scan_done': done}), counts['created'], counts[''],
|
||
chars[0], chars[1], json_text(options), utc_now(), job['id']))
|
||
if error is not None:
|
||
raise error
|
||
return True
|
||
|
||
def _retry_later(self, job, error):
|
||
"""临时错误：保存好的进度不动，任务放回队列，退避一段时间再接着整理（30 秒起，每次翻倍，最多 10 分钟）。"""
|
||
with self.store.database.connect() as db:
|
||
db.execute('BEGIN IMMEDIATE')
|
||
row = self._owned(db, job)
|
||
if row is None:
|
||
return True
|
||
options = json.loads(row['options_json'])
|
||
failures = int(options.get('transient_failures') or 0) + 1
|
||
if failures > MAX_TRANSIENT:
|
||
return False
|
||
delay = min(600, 30 * 2 ** (failures - 1))
|
||
options.update(transient_failures=failures, retry_after=time.time() + delay)
|
||
note = f'{str(error)[:160]}；{delay} 秒后自动重试（第 {failures}/{MAX_TRANSIENT} 次）'
|
||
db.execute("""UPDATE knowledge_job SET status='queued',options_json=?,error=?,lease_owner='',lease_until=0,
|
||
updated_at=? WHERE id=? AND status='running' AND lease_owner=?""",
|
||
(json_text(options), note, utc_now(), job['id'], self.owner))
|
||
return True
|
||
|
||
def step(self):
|
||
job = self.claim()
|
||
if job is None:
|
||
return self.enqueue_incremental()
|
||
try:
|
||
options = json.loads(job['options_json'])
|
||
if options['engine'] == 'model' and not options.get('model_provider_id'):
|
||
from knowledge_models import resolve_model
|
||
resolve_model(self.store.database, options)
|
||
state = json.loads(job['state_json'])
|
||
pending = state.pop('_pending', [])
|
||
done = state.pop('_scan_done', False)
|
||
if not pending and not done:
|
||
# 规则整理一步一页；模型整理多读几页，攒够一轮并发 × 合并所需的问答再调用
|
||
target = (max(1, min(16, int(options.get('model_concurrency') or 1))) * batch_size(options)
|
||
if options['engine'] == 'model' else 1)
|
||
pages = 0
|
||
while not done and len(pending) < target and pages < (MAX_SCAN_PAGES if target > 1 else 1):
|
||
scanned = self.scan(job, options, state, pending)
|
||
if scanned is None:
|
||
return True
|
||
state, pending, done = scanned
|
||
pages += 1
|
||
if pending:
|
||
saved = (self.save_model(job, options, state, pending, done) if options['engine'] == 'model'
|
||
else self.save_rules(job, options, state, pending, done))
|
||
if not saved:
|
||
return True
|
||
with self.store.database.connect() as db:
|
||
db.execute("""UPDATE knowledge_job SET status=?,state_json=?,lease_owner='',lease_until=0,error='',
|
||
options_json=?,updated_at=? WHERE id=? AND status='running' AND lease_owner=?""",
|
||
('completed' if done else 'queued', json_text({} if done else state),
|
||
json_text(options), utc_now(), job['id'], self.owner))
|
||
except Exception as exc:
|
||
if isinstance(exc, ModelCallError) and exc.transient and self._retry_later(job, exc):
|
||
return True
|
||
# Only local validation messages may be persisted; provider bodies stay private.
|
||
error = str(exc)[:240] if isinstance(exc, ValueError) and not isinstance(exc, json.JSONDecodeError) else type(exc).__name__
|
||
if isinstance(exc, ModelCallError) and exc.transient:
|
||
error = f'{error}（已自动重试 {MAX_TRANSIENT} 次仍失败）'[:240]
|
||
with self.store.database.connect() as db:
|
||
db.execute("UPDATE knowledge_job SET status='failed',error=?,lease_owner='',lease_until=0,updated_at=? "
|
||
"WHERE id=? AND lease_owner=? AND status='running'", (error, utc_now(), job['id'], self.owner))
|
||
return True
|
||
|
||
def reindex(store, tenant):
|
||
from archive_store import safe_scope
|
||
from knowledge_retriever import VectorIndex
|
||
tenant = safe_scope(tenant)
|
||
index = VectorIndex()
|
||
if not index.configured:
|
||
raise ValueError("请先配置语义检索服务")
|
||
cursor, count = '', 0
|
||
while True:
|
||
with store.database.connect() as db:
|
||
rows = [dict(r) for r in db.execute("SELECT * FROM knowledge_item WHERE tenant_id=? AND status='published' "
|
||
"AND id>? ORDER BY id LIMIT 100", (tenant, cursor))]
|
||
if not rows:
|
||
return count
|
||
cursor = rows[-1]['id']
|
||
for item in rows:
|
||
profile = index.publish(item)
|
||
with store.database.connect() as db:
|
||
db.execute("BEGIN IMMEDIATE")
|
||
if store._fresh(db, item['id'], tenant):
|
||
result = db.execute("UPDATE knowledge_item SET vector_status='ready',vector_profile=? "
|
||
"WHERE id=? AND tenant_id=? AND revision=? AND status='published'",
|
||
(profile, item['id'], tenant, item['revision']))
|
||
count += result.rowcount
|
||
print(f"已重建 {count} 条向量索引", flush=True)
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="聊天知识加工 Worker")
|
||
parser.add_argument("--db", default="backend.db")
|
||
parser.add_argument("--once", action="store_true", help="处理一个批次后退出")
|
||
parser.add_argument("--reindex", action="store_true", help="为已发布知识重建向量索引")
|
||
parser.add_argument("--tenant", default="", help="重建索引时指定租户")
|
||
args = parser.parse_args()
|
||
if args.reindex and not args.tenant:
|
||
parser.error("--reindex 必须指定 --tenant")
|
||
database = admin_backend.Database(Path(args.db).resolve())
|
||
database.migrate()
|
||
store = KnowledgeStore(database)
|
||
store.initialize()
|
||
if args.reindex:
|
||
reindex(store, args.tenant)
|
||
return
|
||
from knowledge_review import KnowledgeReviewWorker
|
||
review_worker = KnowledgeReviewWorker(store)
|
||
def review_loop():
|
||
while True:
|
||
try:
|
||
worked = review_worker.step()
|
||
except Exception as exc:
|
||
# Keep processing isolated from model requests; do not log content.
|
||
print(f'批量审核暂不可用: {type(exc).__name__}', flush=True)
|
||
worked = False
|
||
time.sleep(0.05 if worked else 2)
|
||
if args.once:
|
||
review_worker.step()
|
||
else:
|
||
threading.Thread(target=review_loop, name='knowledge-review', daemon=True).start()
|
||
worker = KnowledgeWorker(store)
|
||
print("知识加工 Worker 已启动", flush=True)
|
||
while True:
|
||
worked = worker.step()
|
||
if args.once:
|
||
break
|
||
if not worked:
|
||
time.sleep(2)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|