AI Worker 项目管理平台 MVP：项目/任务看板 + Worker注册 + 单任务执行 + HITL审核 + 成本记账

engine.py

@@ -0,0 +1,176 @@
+# -*- coding: utf-8 -*-
+"""
+任务执行引擎：后台线程执行单次 LLM 调用，记录日志与成本，
+完成后进入「待审核」或直接「已完成」。
+"""
+import threading
+import traceback
+import db
+import llm_gateway
+import config
+
+
+def _log(task_id, level, message):
+    try:
+        db.w('INSERT INTO task_logs (task_id, level, message, created_at) VALUES (?,?,?,?)',
+             (task_id, level, message, db.now()))
+    except Exception:
+        pass
+
+
+def _set_task(task_id, **fields):
+    if not fields:
+        return
+    fields['updated_at'] = db.now()
+    sets = ', '.join(f'{k}=?' for k in fields)
+    db.w(f'UPDATE tasks SET {sets} WHERE id=?', (*fields.values(), task_id))
+
+
+def _cost_record(task, worker, usage):
+    db.w(
+        'INSERT INTO cost_records (task_id, project_id, worker_id, provider, model, '
+        'prompt_tokens, completion_tokens, total_tokens, cost, created_at) '
+        'VALUES (?,?,?,?,?,?,?,?,?,?)',
+        (task['id'], task['project_id'], worker['id'], worker['provider'],
+         worker['model'], usage['prompt_tokens'], usage['completion_tokens'],
+         usage['total_tokens'], usage['cost'], db.now()))
+
+
+def pick_worker_auto(task):
+    """自动路由：按模型输入单价升序挑选 enabled Worker"""
+    rows = db.q('SELECT * FROM workers WHERE status="enabled" ORDER BY id')
+    if not rows:
+        return None
+    best, best_price = None, None
+    for r in rows:
+        pin, pout = llm_gateway.model_price(r['model'])
+        price = pin + pout * 0.5
+        if best_price is None or price < best_price:
+            best, best_price = r, price
+    return best
+
+
+def check_worker_limits(worker):
+    """成本上限预检：返回 (ok, reason)"""
+    if worker['task_cost_limit'] and worker['task_cost_limit'] > 0:
+        used = db.task_worker_cost(worker['id'])
+        if used >= worker['task_cost_limit']:
+            return False, f'该 Worker 累计成本 {used:.4f} 元已达单任务上限 {worker["task_cost_limit"]} 元'
+    if worker['monthly_cost_limit'] and worker['monthly_cost_limit'] > 0:
+        used = db.monthly_worker_cost(worker['id'])
+        if used >= worker['monthly_cost_limit']:
+            return False, f'该 Worker 本月成本 {used:.4f} 元已达月度上限 {worker["monthly_cost_limit"]} 元'
+    return True, ''
+
+
+def _check_project_budget(task):
+    proj = db.q('SELECT * FROM projects WHERE id=?', (task['project_id'],), one=True)
+    if proj and proj['budget_limit'] and proj['budget_limit'] > 0:
+        rows = db.q('SELECT COALESCE(SUM(cost),0) AS t FROM cost_records WHERE project_id=?',
+                    (task['project_id'],))
+        used = rows[0]['t'] if rows else 0
+        if used >= proj['budget_limit']:
+            return False, f'项目预算已用完（{used:.2f}/{proj["budget_limit"]:.2f} 元）'
+    return True, ''
+
+
+def run_task(task_id):
+    """在后台线程中执行任务"""
+    task = db.q('SELECT * FROM tasks WHERE id=?', (task_id,), one=True)
+    if not task:
+        return
+    if task['status'] == 'running':
+        return
+
+    # 确定 Worker
+    worker = None
+    if task['worker_id']:
+        worker = db.q('SELECT * FROM workers WHERE id=?', (task['worker_id'],), one=True)
+        if not worker or worker['status'] != 'enabled':
+            _set_task(task_id, status='failed', error='指定 Worker 不存在或已停用',
+                      finished_at=db.now())
+            _log(task_id, 'error', '指定 Worker 不存在或已停用')
+            return
+    else:
+        worker = pick_worker_auto(task)
+        if not worker:
+            _set_task(task_id, status='failed', error='无可用 Worker（自动路由失败）',
+                      finished_at=db.now())
+            _log(task_id, 'error', '自动路由失败：无可用 Worker')
+            return
+        _set_task(task_id, worker_id=worker['id'])
+        _log(task_id, 'info', f'自动路由 → Worker「{worker["name"]}」（{worker["provider"]}/{worker["model"]}）')
+
+    # 预算/成本预检
+    ok, reason = check_worker_limits(worker)
+    if not ok:
+        _set_task(task_id, status='failed', error=reason, finished_at=db.now())
+        _log(task_id, 'error', reason)
+        return
+    ok, reason = _check_project_budget(task)
+    if not ok:
+        _set_task(task_id, status='failed', error=reason, finished_at=db.now())
+        _log(task_id, 'error', reason)
+        return
+
+    _set_task(task_id, status='running', started_at=db.now(), error='')
+    _log(task_id, 'info', f'开始执行：Worker「{worker["name"]}」 模型 {worker["provider"]}/{worker["model"]}')
+
+    messages = []
+    if worker['system_prompt']:
+        messages.append({'role': 'system', 'content': worker['system_prompt']})
+    messages.append({'role': 'user', 'content': task['description'] or task['title']})
+
+    try:
+        usage = llm_gateway.chat(
+            worker['provider'], worker['model'], messages,
+            temperature=worker['temperature'], max_tokens=worker['max_tokens'],
+            base_url=worker['base_url'] or None, api_key=worker['api_key'] or None)
+    except Exception as e:
+        _set_task(task_id, status='failed', error=str(e), finished_at=db.now())
+        _log(task_id, 'error', f'执行失败: {e}')
+        return
+
+    _cost_record(task, worker, usage)
+    _log(task_id, 'success',
+         f'执行完成：{usage["total_tokens"]} tokens（输入 {usage["prompt_tokens"]} / 输出 {usage["completion_tokens"]}），'
+         f'成本 ¥{usage["cost"]:.6f}')
+
+    new_status = 'review' if task['review_required'] else 'done'
+    fields = {
+        'status': new_status,
+        'output_text': usage['text'],
+        'output_version': task['output_version'] + 1,
+        'finished_at': db.now(),
+    }
+    _set_task(task_id, **fields)
+    if new_status == 'review':
+        _log(task_id, 'info', '产出已提交，等待人工审核（HITL）')
+    else:
+        _log(task_id, 'success', '任务完成（无需审核）')
+
+
+class TaskRunner:
+    def __init__(self):
+        self._threads = {}
+
+    def submit(self, task_id):
+        if task_id in self._threads and self._threads[task_id].is_alive():
+            return False
+        t = threading.Thread(target=self._safe_run, args=(task_id,), daemon=True)
+        self._threads[task_id] = t
+        t.start()
+        return True
+
+    def _safe_run(self, task_id):
+        try:
+            run_task(task_id)
+        except Exception:
+            _log(task_id, 'error', '引擎异常: ' + traceback.format_exc())
+            try:
+                _set_task(task_id, status='failed', error='引擎异常，见日志', finished_at=db.now())
+            except Exception:
+                pass
+
+
+runner = TaskRunner()