hz4th_coder
c56745a8ab
1. 精细化统计：cost_records 新增 calls/cached_tokens/latency_ms/first_token_ms；
用量明细报表(项目×智能体矩阵) + 成本报表细化(输入/输出/缓存命中/调用次数)
2. 从参考项目中新建：内置3个测试项目(文案/Python/调研)，一键复制目标+任务
3. 大模型接口库(llm_endpoints)：专门配置接口(地址/密钥/模型/定价)，
计费支持按token(逐模型)与按调用次数；创建AI Worker直接选用；
AI Worker团队(worker_teams)：打包Worker，对话/建项目可直接选团队
4. 对话导航融合仪表盘：可选大模型/AI Worker/团队，默认主力AI Worker(⭐可设)，SSE流式
5. 系统工作目录：默认data/workspace可改绝对路径；项目与多Agent协作均在其下
建唯一工作目录；手动输入目录已存在则列出信息并需手动确认
6. 任务执行超时改为流式单token返回超时+首字延迟超时，设置页可配；
所有模型输出SSE按token接收
338 lines
14 KiB
Python
338 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
任务执行引擎 V1：
|
||
- DAG 依赖校验 + 下游自动触发（串行/并行）
|
||
- RAG 知识库上下文注入
|
||
- 预算告警（项目预算使用率阈值）
|
||
- 事件通知（待审核/完成/失败）与告警记录
|
||
"""
|
||
import json
|
||
import threading
|
||
import traceback
|
||
import db
|
||
import llm_gateway
|
||
import config
|
||
import rag
|
||
import notify
|
||
import delivery
|
||
|
||
|
||
def _notify_failed(task, message):
|
||
"""任务失败：写告警 + 推送渠道 + 邮件通知送达者（项目难关）"""
|
||
notify.notify('task_failed', f'任务失败：{task["title"]}',
|
||
f'项目 #{task["project_id"]} 任务「{task["title"]}」{message}',
|
||
save_alert=True, level='warn', atype='task_failed')
|
||
try:
|
||
proj = db.q('SELECT * FROM projects WHERE id=?', (task['project_id'],), one=True)
|
||
if proj and (proj.get('deliver_email') or '').strip():
|
||
delivery.notify_blocker(proj, task['title'], message)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
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, cached_tokens, total_tokens, calls, cost, '
|
||
'latency_ms, first_token_ms, created_at) '
|
||
'VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)',
|
||
(task['id'], task['project_id'], worker['id'], worker['provider'],
|
||
worker['model'], usage['prompt_tokens'], usage['completion_tokens'],
|
||
usage.get('cached_tokens', 0), usage['total_tokens'], usage.get('calls', 1),
|
||
usage['cost'], usage.get('elapsed_ms', 0), usage.get('first_token_ms') or 0,
|
||
db.now()))
|
||
|
||
|
||
def _deps(task):
|
||
try:
|
||
return json.loads(task.get('depends_on') or '[]')
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
def check_dependencies(task):
|
||
"""DAG 依赖检查：返回 (ok, blockers)"""
|
||
blockers = []
|
||
for dep_id in _deps(task):
|
||
dep = db.q('SELECT id, title, status FROM tasks WHERE id=?', (dep_id,), one=True)
|
||
if not dep:
|
||
blockers.append(f'# {dep_id}（已删除）')
|
||
elif dep['status'] != 'done':
|
||
blockers.append(f'「{dep["title"]}」(#{dep_id}) {dep["status"]}')
|
||
return (not blockers), blockers
|
||
|
||
|
||
def pick_worker_auto(task):
|
||
"""自动路由：按模型单次调用成本升序挑选 enabled Worker（V3.5 用接口库计价）"""
|
||
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:
|
||
try:
|
||
price = llm_gateway.worker_unit_price(r)
|
||
except Exception:
|
||
continue
|
||
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 _project_cost(project_id):
|
||
rows = db.q('SELECT COALESCE(SUM(cost),0) AS t FROM cost_records WHERE project_id=?',
|
||
(project_id,))
|
||
return rows[0]['t'] if rows else 0.0
|
||
|
||
|
||
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:
|
||
used = _project_cost(task['project_id'])
|
||
if used >= proj['budget_limit']:
|
||
return False, f'项目预算已用完（{used:.2f}/{proj["budget_limit"]:.2f} 元）'
|
||
return True, ''
|
||
|
||
|
||
def _budget_alert(project_id):
|
||
"""预算使用率告警（每次任务完成后检查，避免重复刷屏）"""
|
||
proj = db.q('SELECT * FROM projects WHERE id=?', (project_id,), one=True)
|
||
if not proj or not proj['budget_limit'] or proj['budget_limit'] <= 0:
|
||
return
|
||
used = _project_cost(project_id)
|
||
ratio = used / proj['budget_limit']
|
||
if ratio >= config.BUDGET_ALERT_RATIO:
|
||
# 同项目 1 小时内只告警一次，避免刷屏
|
||
dup = db.q('SELECT COUNT(*) c FROM alerts WHERE type="budget" AND detail LIKE ? '
|
||
'AND created_at > ?', (f'项目「{proj["name"]}」%', db.now() - 3600))
|
||
if dup[0]['c'] == 0:
|
||
notify.notify('budget_alert',
|
||
f'预算告警：项目「{proj["name"]}」已使用 {ratio*100:.0f}%',
|
||
f'已花费 ¥{used:.2f} / 预算 ¥{proj["budget_limit"]:.2f}，'
|
||
f'超过阈值 {config.BUDGET_ALERT_RATIO*100:.0f}%，请关注成本控制。',
|
||
save_alert=True, level='warn', atype='budget')
|
||
|
||
|
||
def _build_messages(task, worker):
|
||
"""构造提示词：任务指令 + RAG 知识库上下文
|
||
支持图片注入：任务描述中的  或 图片:URL 会转成多模态消息（视觉 Worker）。"""
|
||
import re as _re
|
||
messages = []
|
||
if worker['system_prompt']:
|
||
messages.append({'role': 'system', 'content': worker['system_prompt']})
|
||
user_text = task['description'] or task['title']
|
||
ctx, refs = rag.build_context(task['project_id'], user_text)
|
||
if ctx:
|
||
user_text = f'{ctx}\n\n----\n\n任务指令：{user_text}'
|
||
_log(task['id'], 'info', f'📚 RAG 知识库命中 {len(refs)} 个片段：' + '；'.join(refs[:5]))
|
||
# 提取图片 URL（ 或 图片:url 或 image:url）
|
||
img_urls = []
|
||
for m in _re.finditer(r'!\[[^\]]*\]\(([^)\s]+)\)', user_text):
|
||
img_urls.append(m.group(1))
|
||
for m in _re.finditer(r'(?:图片|image)\s*[:：]\s*(https?://\S+)', user_text, _re.I):
|
||
img_urls.append(m.group(1))
|
||
if img_urls:
|
||
content = [{'type': 'text', 'text': user_text}]
|
||
for u in img_urls:
|
||
content.append({'type': 'image_url', 'image_url': {'url': u}})
|
||
messages.append({'role': 'user', 'content': content})
|
||
_log(task['id'], 'info', f'🖼️ 检测到 {len(img_urls)} 张图片，已注入多模态消息')
|
||
else:
|
||
messages.append({'role': 'user', 'content': user_text})
|
||
return messages
|
||
|
||
|
||
def _trigger_downstream(task):
|
||
"""DAG：任务完成后自动触发所有就绪的下游任务"""
|
||
rows = db.q('SELECT * FROM tasks WHERE deleted=0 AND status IN ("todo","failed")')
|
||
triggered = []
|
||
for t in rows:
|
||
deps = _deps(t)
|
||
if task['id'] not in deps:
|
||
continue
|
||
ok, blockers = check_dependencies(t)
|
||
if ok:
|
||
if runner.submit(t['id']):
|
||
triggered.append(t)
|
||
else:
|
||
_log(t['id'], 'info', f'⏳ 等待前置任务完成：' + '、'.join(blockers))
|
||
return triggered
|
||
|
||
|
||
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
|
||
|
||
# DAG 依赖检查
|
||
ok, blockers = check_dependencies(task)
|
||
if not ok:
|
||
_set_task(task_id, status='failed', error='前置任务未完成：' + '、'.join(blockers),
|
||
finished_at=db.now())
|
||
_log(task_id, 'error', '❌ 依赖未满足，无法执行：' + '、'.join(blockers))
|
||
_notify_failed(task, '因依赖未完成被拒绝执行：' + '、'.join(blockers))
|
||
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 不存在或已停用')
|
||
_notify_failed(task, '指定 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')
|
||
_notify_failed(task, '自动路由失败：没有可用的 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)
|
||
_notify_failed(task, 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)
|
||
_notify_failed(task, 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"]}')
|
||
|
||
try:
|
||
usage = llm_gateway.chat_worker(worker, _build_messages(task, worker))
|
||
except llm_gateway.LLMError as e:
|
||
err_msg = str(e)
|
||
if e.partial_text:
|
||
# 流式超时但已有部分产出：保留部分产出，标记失败并说明原因
|
||
db.w('UPDATE tasks SET output_text=?, status="failed", error=?, finished_at=? WHERE id=?',
|
||
(e.partial_text[:200000], err_msg, db.now(), task_id))
|
||
_log(task_id, 'warn', f'流式输出中断（已收到部分内容）：{err_msg}')
|
||
else:
|
||
_set_task(task_id, status='failed', error=err_msg, finished_at=db.now())
|
||
_log(task_id, 'error', f'执行失败: {err_msg}')
|
||
_notify_failed(task, f'执行出错：{err_msg[:300]}')
|
||
return
|
||
except Exception as e:
|
||
_set_task(task_id, status='failed', error=str(e), finished_at=db.now())
|
||
_log(task_id, 'error', f'执行失败: {e}')
|
||
_notify_failed(task, f'执行出错：{str(e)[:300]}')
|
||
return
|
||
|
||
_cost_record(task, worker, usage)
|
||
_log(task_id, 'success',
|
||
f'执行完成：{usage["total_tokens"]} tokens（输入 {usage["prompt_tokens"]} / 输出 {usage["completion_tokens"]} / 缓存命中 {usage.get("cached_tokens", 0)}），'
|
||
f'首字 {usage.get("first_token_ms") or "—"}ms · 总耗时 {usage.get("elapsed_ms", 0)}ms，成本 ¥{usage["cost"]:.6f}')
|
||
|
||
# V3.4：产出若为完整网页文档 → 落盘工作目录，供 Demo 真实展示
|
||
try:
|
||
saved = delivery.save_task_output(task, usage['text'])
|
||
if saved:
|
||
_log(task_id, 'info', f'🌐 网页产出已保存到工作目录：' + '、'.join(saved))
|
||
except Exception:
|
||
pass
|
||
|
||
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）')
|
||
notify.notify('task_review', f'任务待审核：{task["title"]}',
|
||
f'项目 #{task["project_id"]} 任务「{task["title"]}」已完成，等待人工验收。\n'
|
||
f'模型 {worker["provider"]}/{worker["model"]} · {usage["total_tokens"]} tokens · ¥{usage["cost"]:.4f}',
|
||
save_alert=True, level='info', atype='task_review')
|
||
else:
|
||
_log(task_id, 'success', '任务完成（无需审核）')
|
||
notify.notify('task_done', f'任务完成：{task["title"]}',
|
||
f'项目 #{task["project_id"]} 任务「{task["title"]}」执行完毕，'
|
||
f'成本 ¥{usage["cost"]:.4f}，tokens {usage["total_tokens"]}',
|
||
save_alert=False)
|
||
|
||
_budget_alert(task['project_id'])
|
||
|
||
if new_status == 'done':
|
||
# V3：无需审核的任务直接完成后，检查项目是否全部完成 → 自动交付并通知送达者
|
||
try:
|
||
delivery.auto_complete_if_ready(task['project_id'])
|
||
except Exception:
|
||
pass
|
||
|
||
# DAG：触发下游就绪任务
|
||
downstream = _trigger_downstream(task)
|
||
for t in downstream:
|
||
_log(t['id'], 'info', f'🔗 前置任务「{task["title"]}」已完成，自动触发执行')
|
||
|
||
|
||
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()
|