686 lines
30 KiB
Python
686 lines
30 KiB
Python
"""AI 控制台 API：模型/Key 前端配置 + Agent 流式执行（SSE 事件流）。
|
||
|
||
流程：
|
||
POST /api/agent/run {prompt} → 启动 Agent 线程，返回 run_id
|
||
GET /api/agent/stream?run_id= → SSE 事件流（EventSource 订阅）：
|
||
event: delta {text, kind: content|reasoning} 流式文本增量
|
||
event: step {tool, args, image?} 工具调用完成（MCP 步骤）
|
||
event: done {answer} 完成
|
||
event: error {message} 失败
|
||
GET/POST /api/agent/config → 配置读写（key 打码回显）
|
||
|
||
单实例：同时只允许一个 Agent 运行。
|
||
"""
|
||
import asyncio
|
||
import base64
|
||
import io
|
||
import json
|
||
import os
|
||
import queue
|
||
import re
|
||
import sys
|
||
import threading
|
||
import uuid
|
||
|
||
from flask import Blueprint, Response, jsonify, request
|
||
|
||
from core.logger import get_logger
|
||
from core.models import db
|
||
from web.auth import admin_required
|
||
|
||
_log = get_logger("web.agent")
|
||
bp = Blueprint("agent", __name__)
|
||
|
||
# ---------- 配置键（app_meta） ----------
|
||
_CFG_KEYS = {"api_base": "agent_api_base",
|
||
"model": "agent_model",
|
||
"api_key": "agent_api_key",
|
||
"default_serial": "agent_default_serial"}
|
||
|
||
# ---------- 运行状态（单实例 + 事件队列） ----------
|
||
_run = {"id": None, "state": "idle", "prompt": "", "serial": "",
|
||
"answer": "", "error": "",
|
||
"history": []} # 多轮对话历史 [{role: user|assistant, content}]
|
||
_queues = {} # run_id -> queue.Queue（SSE 消费者读取）
|
||
_stop_events = {} # run_id -> threading.Event（用户中断）
|
||
_lock = threading.Lock()
|
||
_flask_app = None # web_server 注册时注入（后台线程 db 操作需 app context）
|
||
|
||
|
||
def set_app(app):
|
||
global _flask_app
|
||
_flask_app = app
|
||
|
||
|
||
def _meta_get(key):
|
||
return db.session.execute(
|
||
db.text("SELECT value FROM app_meta WHERE key=:k"), {"k": key}).scalar() or ""
|
||
|
||
|
||
def _meta_put(key, value):
|
||
db.session.execute(
|
||
db.text("INSERT OR REPLACE INTO app_meta(key,value) VALUES(:k,:v)"),
|
||
{"k": key, "v": str(value)})
|
||
|
||
|
||
def _read_cfg():
|
||
return {k: _meta_get(v) for k, v in _CFG_KEYS.items()}
|
||
|
||
|
||
# ================== 经验记忆（自进化） ==================
|
||
# agent_experience：任务成功后的操作配方，下次相似任务检索注入 system prompt。
|
||
# 原始 SQLite（CREATE IF NOT EXISTS 幂等），不进模型层迁移。
|
||
_EXP_TABLE = """
|
||
CREATE TABLE IF NOT EXISTS agent_experience (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
task_prompt TEXT DEFAULT '',
|
||
recipe TEXT DEFAULT '',
|
||
tool_seq TEXT DEFAULT '',
|
||
hits INTEGER DEFAULT 0,
|
||
created_at VARCHAR(20) DEFAULT '')"""
|
||
|
||
# 经验巡检（AI 质检）：每日定时把经验交给模型评审，疑似问题标 pending 待人工确认。
|
||
# 删除只允许人工（前端 confirm / 本 API），巡检绝不自动删。
|
||
_AUDIT_TABLE = """
|
||
CREATE TABLE IF NOT EXISTS experience_audit (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
exp_id INTEGER NOT NULL,
|
||
verdict TEXT DEFAULT '', -- keep 保留 / delete 建议删除
|
||
score REAL DEFAULT 0, -- 0-10 可用性评分
|
||
reason TEXT DEFAULT '', -- 模型给出的理由
|
||
hits INTEGER DEFAULT 0, -- 巡检时的引用次数（hits>0 被删要更谨慎）
|
||
action TEXT DEFAULT 'pending', -- pending 待人工确认 / kept 已保留 / deleted 已删除
|
||
audited_at VARCHAR(20) DEFAULT '')"""
|
||
|
||
# 巡检评审提示词：模型只判「保留/建议删除」+ 评分 + 理由，不直接删。
|
||
_AUDIT_PROMPT = """你是「手机自动化经验库」质检员。经验库存放 AI 成功操作手机后提炼的
|
||
操作套路，下次相似任务会自动注入参考。请判断下面这条经验是否值得继续保留。
|
||
|
||
判断标准（全部满足才保留）：
|
||
1. 具体可执行：步骤是明确的工具操作（de_* 或清晰的中文步骤），不是「de_tap / de_input」
|
||
这类空泛占位，也没有编造不存在的工具名
|
||
2. 不依赖易过时信息：不含写死的像素坐标（x,y 数字）这类界面一变就失效的步骤
|
||
3. 是独立任务的操作套路（如「打开X搜索Y并点赞」），不是对话追问/抱怨/单次性内容
|
||
4. 配方与下方实际操作序列大致一致，不是模型自由发挥编造
|
||
|
||
经验内容：
|
||
任务描述：{prompt}
|
||
操作配方：{recipe}
|
||
实际操作序列：{tool_seq}
|
||
被引用次数：{hits}
|
||
|
||
只输出 JSON：{{"keep": true或false, "score": 0到10的整数, "reason": "一句话中文理由"}}"""
|
||
|
||
|
||
def _ensure_exp_table():
|
||
try:
|
||
db.session.execute(db.text(_EXP_TABLE))
|
||
db.session.commit()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _bigrams(text):
|
||
"""中文/英文文本 bigram 集合（无空格分词，粗粒度相似度）。"""
|
||
t = "".join(c for c in (text or "").lower() if c.isalnum() or "\u4e00" <= c <= "\u9fff")
|
||
return {t[i:i + 2] for i in range(len(t) - 1)}
|
||
|
||
|
||
# ---------- 经验质量门槛 ----------
|
||
# 自进化经验只应保存「独立操作任务」的成功套路。多轮对话里用户的短句
|
||
# （质疑/纠正/催促，如「继续啊」「你确定我是卡1吗」「还没有完成啊」）
|
||
# 不是新任务——把执行出错被纠正的轮次存成经验会教坏后续任务。
|
||
# 用启发式过滤（零成本，可解释），配方层再校验工具名真实性。
|
||
|
||
# 任务性动词：命中任一视为有明确操作诉求（疑问/纠错短句一般不含它们）
|
||
_TASK_VERBS = ("打开", "搜索", "查看", "找到", "截图", "输入", "点击", "点开",
|
||
"发送", "安装", "卸载", "下载", "启动", "停止", "关闭", "退出",
|
||
"登录", "切换", "设置", "删除", "清理", "复制", "粘贴", "读取",
|
||
"剪贴板", "长按", "滑动", "播放", "发布", "检查", "看看",
|
||
"帮我", "给我", "请", "拍张", "查一下")
|
||
# 对话续语开头：几乎只出现在承接上一轮（「继续啊」「还有吗」）
|
||
_CONTINUE_PREFIXES = ("继续", "还有", "然后呢", "再来", "快点", "刚才",
|
||
"接着", "上一步")
|
||
# 强质疑/纠错信号（不含「为什么」——「查一下为什么」是正当任务）
|
||
_DOUBT_MARKS = ("你确定", "是不是", "不是吧", "不是吗", "怎么都", "怎么还",
|
||
"还没有", "没看到", "我说的是", "你听我说", "不对吧", "你又",
|
||
"重新来", "错了", "你说得", "你回答")
|
||
# 全部真实 MCP 工具（配方里出现不存在的 de_* 说明蒸馏模型在编造，弃存）
|
||
_KNOWN_TOOLS = frozenset({
|
||
"de_list_devices", "de_screenshot", "de_tap", "de_swipe", "de_ui_tree",
|
||
"de_tap_element", "de_read_clipboard", "de_wake", "de_press_key",
|
||
"de_open_app", "de_stop_app", "de_foreground_app", "de_type_text",
|
||
"de_set_clipboard", "de_sleep", "de_ocr", "de_tap_text", "de_list_apps",
|
||
"de_list_tasks"})
|
||
|
||
|
||
def _qualify_experience(prompt, recipe):
|
||
"""经验入库前质量门槛，返回 True=值得保存。
|
||
|
||
1) prompt 太短 / 纯续语开头 / 质疑纠错 → 非独立任务，弃
|
||
2) 疑问短句（≤30 字、以 吗/？ 结尾）且无任务动词 → 追问/反问，弃
|
||
3) 配方含不存在的 de_* 工具（蒸馏模型自由发挥）→ 弃
|
||
"""
|
||
t = "".join(c for c in (prompt or "") if not c.isspace())
|
||
if len(t) < 6:
|
||
_log.info("经验弃存：prompt 过短「%s」", t[:20])
|
||
return False
|
||
if any(t.startswith(p) for p in _CONTINUE_PREFIXES):
|
||
_log.info("经验弃存：对话续语开头「%s」", t[:20])
|
||
return False
|
||
if any(m in t for m in _DOUBT_MARKS):
|
||
_log.info("经验弃存：质疑/纠错语气「%s」", t[:20])
|
||
return False
|
||
if (t.endswith("吗") or t.endswith("？") or t.endswith("?")) \
|
||
and len(t) <= 30 and not any(v in t for v in _TASK_VERBS):
|
||
_log.info("经验弃存：无操作诉求的追问「%s」", t[:20])
|
||
return False
|
||
for name in re.findall(r"de_[a-z_]+", recipe or ""):
|
||
if name not in _KNOWN_TOOLS:
|
||
_log.info("经验弃存：配方含不存在的工具 %s", name)
|
||
return False
|
||
return True
|
||
|
||
|
||
def _find_experiences(prompt, limit=2, threshold=0.10):
|
||
"""按 bigram 重叠检索相似历史经验（prompt 与任务描述的字符相似度）。
|
||
|
||
命中的经验 hits+1（回写），让被反复参考的有效经验浮到前面。
|
||
"""
|
||
try:
|
||
if _flask_app is None:
|
||
return ""
|
||
with _flask_app.app_context():
|
||
_ensure_exp_table()
|
||
rows = db.session.execute(db.text(
|
||
"SELECT id, task_prompt, recipe, hits FROM agent_experience "
|
||
"WHERE recipe != '' ORDER BY hits DESC, id DESC LIMIT 50")).fetchall()
|
||
except Exception:
|
||
return ""
|
||
if not rows:
|
||
return ""
|
||
cur = _bigrams(prompt)
|
||
if not cur:
|
||
return ""
|
||
scored = []
|
||
for row_id, task_prompt, recipe, hits in rows:
|
||
sim = len(cur & _bigrams(task_prompt)) / len(cur)
|
||
if sim >= threshold:
|
||
scored.append((sim, hits or 0, recipe, row_id))
|
||
scored.sort(key=lambda x: (-x[0], -x[1]))
|
||
if scored:
|
||
# hits 回写（尽力而为，失败不影响检索）
|
||
try:
|
||
with _flask_app.app_context():
|
||
for _sim, _hits, _recipe, row_id in scored:
|
||
db.session.execute(db.text(
|
||
"UPDATE agent_experience SET hits=hits+1 WHERE id=:i"),
|
||
{"i": row_id})
|
||
db.session.commit()
|
||
except Exception:
|
||
pass
|
||
parts = []
|
||
for sim, _hits, recipe, _row_id in scored[:limit]:
|
||
parts.append(f"- {recipe[:600]}")
|
||
return "\n".join(parts)
|
||
|
||
|
||
def _distill_experience(cfg, prompt, tool_seq):
|
||
"""任务完成后用模型把操作序列提炼为可复用配方（失败静默，不阻塞）。"""
|
||
try:
|
||
import httpx
|
||
body = {
|
||
"model": cfg.get("model") or "deepseek-v4-flash-vision-exp",
|
||
"messages": [{"role": "user",
|
||
"content": "以下是一次成功的手机自动化操作记录。请提炼成简洁的"
|
||
"「操作配方」（2-6 步，每步：目标 → 用哪个工具），"
|
||
"供下次同类任务参考。不要解释，直接输出配方。\n"
|
||
f"任务：{prompt[:300]}\n操作序列：{tool_seq[:800]}"}],
|
||
"max_tokens": 600,
|
||
}
|
||
headers = {"Authorization": f"Bearer {cfg.get('api_key', '')}",
|
||
"Content-Type": "application/json"}
|
||
r = httpx.post(f"{(cfg.get('api_base') or 'https://api.deepseek.com').rstrip('/')}/chat/completions",
|
||
json=body, headers=headers, timeout=25)
|
||
if r.status_code != 200:
|
||
return ""
|
||
j = r.json()
|
||
recipe = ((j.get("choices") or [{}])[0].get("message") or {}).get("content") or ""
|
||
return recipe.strip()[:1500]
|
||
except Exception as e:
|
||
_log.warning(f"经验提炼失败: {e}")
|
||
return ""
|
||
|
||
|
||
def _save_experience(prompt, recipe, tool_seq):
|
||
"""保存经验（后台线程调用，包 app context）。返回是否保存成功。"""
|
||
try:
|
||
if _flask_app is None:
|
||
return False
|
||
with _flask_app.app_context():
|
||
_ensure_exp_table()
|
||
from datetime import datetime
|
||
db.session.execute(db.text(
|
||
"INSERT INTO agent_experience(task_prompt, recipe, tool_seq, hits, created_at) "
|
||
"VALUES(:p, :r, :t, 0, :c)"),
|
||
{"p": prompt[:500], "r": recipe, "t": tool_seq[:1000],
|
||
"c": datetime.now().strftime("%Y-%m-%d %H:%M")})
|
||
db.session.commit()
|
||
_log.info("经验已保存（配方 %d 字符）", len(recipe))
|
||
return True
|
||
except Exception as e:
|
||
_log.warning(f"经验保存失败: {e}")
|
||
return False
|
||
|
||
|
||
# ================== 经验巡检（AI 质检，删除需人工确认） ==================
|
||
_audit_state = {"running": False, "last": "", "last_summary": ""}
|
||
|
||
|
||
def _ensure_audit_table():
|
||
try:
|
||
db.session.execute(db.text(_AUDIT_TABLE))
|
||
db.session.commit()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _review_one_exp(cfg, prompt, recipe, tool_seq, hits):
|
||
"""调模型评审单条经验，返回 (verdict, score, reason)。
|
||
|
||
模型偶尔输出解释文字不带 JSON——重试一次（强调只输出 JSON），
|
||
仍失败则保守返回保留。
|
||
"""
|
||
import httpx
|
||
base_prompt = _AUDIT_PROMPT.format(prompt=(prompt or "")[:400],
|
||
recipe=(recipe or "")[:600],
|
||
tool_seq=(tool_seq or "")[:600],
|
||
hits=hits or 0)
|
||
for attempt in (1, 2):
|
||
try:
|
||
content = base_prompt if attempt == 1 else (
|
||
base_prompt + "\n注意：只输出一个 JSON 对象，不要输出任何其它文字、解释或代码块标记。")
|
||
body = {
|
||
"model": cfg.get("model") or "deepseek-v4-flash-vision-exp",
|
||
"messages": [{"role": "user", "content": content}],
|
||
"max_tokens": 200,
|
||
}
|
||
headers = {"Authorization": f"Bearer {cfg.get('api_key', '')}",
|
||
"Content-Type": "application/json"}
|
||
r = httpx.post(f"{(cfg.get('api_base') or 'https://api.deepseek.com').rstrip('/')}/chat/completions",
|
||
json=body, headers=headers, timeout=20)
|
||
if r.status_code != 200:
|
||
raise RuntimeError(f"HTTP {r.status_code}")
|
||
text = (((r.json() or {}).get("choices") or [{}])[0]
|
||
.get("message") or {}).get("content") or ""
|
||
m = re.search(r"\{.*\}", text, re.S)
|
||
if not m:
|
||
raise RuntimeError("响应无 JSON")
|
||
j = json.loads(m.group(0))
|
||
keep = bool(j.get("keep", True))
|
||
score = max(0, min(10, int(j.get("score", 5) or 5)))
|
||
reason = str(j.get("reason", ""))[:200]
|
||
return ("keep" if keep else "delete"), score, reason
|
||
except Exception as e:
|
||
_log.warning(f"经验评审第 {attempt} 次失败: {e}")
|
||
return "keep", 0, "评审失败自动保留（两次尝试均无有效 JSON）"
|
||
|
||
|
||
def run_experience_audit():
|
||
"""巡检一轮：逐条评审经验 → 写 experience_audit。
|
||
|
||
规则：
|
||
- 最新已 kept / deleted 的经验不再重复评审（人工拍板过的尊重）
|
||
- 评审 keep → 直接 action=kept；delete → action=pending（等人工确认）
|
||
- 绝不自动删除；无 api_key 配置时跳过并记日志
|
||
由 web_server 每日定时调度或 API 手动触发（后台线程）。
|
||
"""
|
||
if _audit_state["running"]:
|
||
_log.info("经验巡检已在运行，跳过本次触发")
|
||
return
|
||
_audit_state["running"] = True
|
||
try:
|
||
if _flask_app is None:
|
||
return
|
||
with _flask_app.app_context():
|
||
_ensure_exp_table()
|
||
_ensure_audit_table()
|
||
cfg = _read_cfg()
|
||
if not cfg.get("api_key"):
|
||
_log.info("经验巡检跳过：未配置 API Key")
|
||
return
|
||
rows = db.session.execute(db.text(
|
||
"SELECT id, task_prompt, recipe, tool_seq, hits FROM agent_experience "
|
||
"ORDER BY id")).fetchall()
|
||
if not rows:
|
||
_log.info("经验巡检：经验库为空")
|
||
return
|
||
from datetime import datetime
|
||
now = datetime.now().strftime("%Y-%m-%d %H:%M")
|
||
reviewed = suggested = 0
|
||
for exp_id, prompt, recipe, tool_seq, hits in rows:
|
||
# 最新一次人工/评审结论：kept / deleted 的不再打扰
|
||
last = db.session.execute(db.text(
|
||
"SELECT action FROM experience_audit WHERE exp_id=:e "
|
||
"ORDER BY id DESC LIMIT 1"), {"e": exp_id}).scalar()
|
||
if last in ("kept", "deleted"):
|
||
continue
|
||
verdict, score, reason = _review_one_exp(
|
||
cfg, prompt, recipe, tool_seq, hits)
|
||
action = "kept" if verdict == "keep" else "pending"
|
||
db.session.execute(db.text(
|
||
"INSERT INTO experience_audit"
|
||
"(exp_id, verdict, score, reason, hits, action, audited_at) "
|
||
"VALUES(:e, :v, :s, :r, :h, :a, :t)"),
|
||
{"e": exp_id, "v": verdict, "s": score, "r": reason,
|
||
"h": hits or 0, "a": action, "t": now})
|
||
reviewed += 1
|
||
suggested += 1 if action == "pending" else 0
|
||
db.session.commit()
|
||
summary = f"评审 {reviewed} 条，建议删除 {suggested} 条（待人工确认）"
|
||
_audit_state.update(last=now, last_summary=summary)
|
||
_log.info("经验巡检完成: %s", summary)
|
||
except Exception as e:
|
||
_log.warning(f"经验巡检异常: {e}")
|
||
finally:
|
||
_audit_state["running"] = False
|
||
|
||
|
||
# ================== 经验库管理 API ==================
|
||
@bp.route("/api/agent/experience")
|
||
@admin_required
|
||
def agent_experience_list():
|
||
"""经验库列表（含最近一次巡检结论），供 AI 控制台「🧠 经验库」面板管理。"""
|
||
_ensure_exp_table()
|
||
_ensure_audit_table()
|
||
rows = db.session.execute(db.text(
|
||
"SELECT e.id, e.task_prompt, e.recipe, e.tool_seq, e.hits, e.created_at, "
|
||
"a.verdict, a.score, a.reason, a.action, a.audited_at "
|
||
"FROM agent_experience e LEFT JOIN experience_audit a ON a.id = "
|
||
"(SELECT id FROM experience_audit WHERE exp_id = e.id ORDER BY id DESC LIMIT 1) "
|
||
"ORDER BY e.id DESC")).fetchall()
|
||
exps = []
|
||
for r in rows:
|
||
(eid, task_prompt, recipe, tool_seq, hits, created_at,
|
||
verdict, score, reason, action, audited_at) = r
|
||
audit = None
|
||
if action:
|
||
audit = {"verdict": verdict, "score": round(score or 0, 1),
|
||
"reason": reason, "action": action, "at": audited_at}
|
||
exps.append({"id": eid, "task_prompt": task_prompt or "",
|
||
"recipe": recipe or "", "tool_seq": tool_seq or "",
|
||
"hits": hits or 0, "created_at": created_at or "",
|
||
"audit": audit})
|
||
return jsonify({"ok": True, "running": _audit_state["running"],
|
||
"last": _audit_state["last"],
|
||
"last_summary": _audit_state["last_summary"],
|
||
"experiences": exps})
|
||
|
||
|
||
@bp.route("/api/agent/experience/delete", methods=["POST"])
|
||
@admin_required
|
||
def agent_experience_delete():
|
||
"""人工确认删除经验（真删；前端有二次确认，巡检绝不自动调它）。"""
|
||
data = request.json or {}
|
||
exp_id = int(data.get("id") or 0)
|
||
if exp_id <= 0:
|
||
return jsonify({"ok": False, "error": "缺少 id"}), 400
|
||
db.session.execute(db.text(
|
||
"DELETE FROM agent_experience WHERE id=:i"), {"i": exp_id})
|
||
db.session.execute(db.text(
|
||
"UPDATE experience_audit SET action='deleted' "
|
||
"WHERE exp_id=:i AND action='pending'"), {"i": exp_id})
|
||
db.session.commit()
|
||
_log.info("人工确认删除经验 #%d", exp_id)
|
||
return jsonify({"ok": True, "msg": f"经验 #{exp_id} 已删除"})
|
||
|
||
|
||
@bp.route("/api/agent/experience/audit", methods=["POST"])
|
||
@admin_required
|
||
def agent_experience_audit_now():
|
||
"""手动触发一轮经验巡检（后台线程；进行中返回 409）。"""
|
||
if _audit_state["running"]:
|
||
return jsonify({"ok": False, "error": "巡检正在进行中"}), 409
|
||
threading.Thread(target=run_experience_audit, daemon=True).start()
|
||
return jsonify({"ok": True, "msg": "巡检已启动，完成后刷新列表查看建议"})
|
||
|
||
|
||
@bp.route("/api/agent/experience/keep", methods=["POST"])
|
||
@admin_required
|
||
def agent_experience_keep():
|
||
"""人工保留：撤销建议删除（该条后续巡检不再重复建议）。"""
|
||
data = request.json or {}
|
||
exp_id = int(data.get("id") or 0)
|
||
if exp_id <= 0:
|
||
return jsonify({"ok": False, "error": "缺少 id"}), 400
|
||
db.session.execute(db.text(
|
||
"UPDATE experience_audit SET action='kept' "
|
||
"WHERE exp_id=:i AND action='pending'"), {"i": exp_id})
|
||
db.session.commit()
|
||
return jsonify({"ok": True, "msg": f"经验 #{exp_id} 已保留"})
|
||
|
||
@bp.route("/api/agent/config", methods=["GET"])
|
||
@admin_required
|
||
def agent_config_get():
|
||
"""读 Agent 配置（key 打码返回）。"""
|
||
cfg = _read_cfg()
|
||
if cfg["api_key"]:
|
||
k = cfg["api_key"]
|
||
cfg["api_key_masked"] = k[:6] + "***" + k[-4:]
|
||
return jsonify({"ok": True, **cfg})
|
||
|
||
|
||
@bp.route("/api/agent/config", methods=["POST"])
|
||
@admin_required
|
||
def agent_config_save():
|
||
"""保存 Agent 配置：{api_base?, model?, api_key?, default_serial?} 部分更新。"""
|
||
data = request.json or {}
|
||
for key, meta_key in _CFG_KEYS.items():
|
||
if key in data and data[key] is not None:
|
||
_meta_put(meta_key, str(data[key]).strip())
|
||
db.session.commit()
|
||
return jsonify({"ok": True, "msg": "已保存"})
|
||
|
||
|
||
# ================== 运行 ==================
|
||
@bp.route("/api/agent/run", methods=["POST"])
|
||
@admin_required
|
||
def agent_run():
|
||
"""启动 Agent：{prompt, serial?}。运行中返回 409。"""
|
||
data = request.json or {}
|
||
prompt = (data.get("prompt") or "").strip()
|
||
if not prompt:
|
||
return jsonify({"ok": False, "error": "请输入指令"}), 400
|
||
serial = (data.get("serial") or "").strip()
|
||
cfg = _read_cfg()
|
||
if not cfg.get("api_key"):
|
||
return jsonify({"ok": False, "error": "请先在配置区填写 API Key"}), 400
|
||
if not cfg.get("model"):
|
||
return jsonify({"ok": False, "error": "请先填写模型名"}), 400
|
||
with _lock:
|
||
if _run["state"] == "running":
|
||
return jsonify({"ok": False, "error": "已有 Agent 运行中，请等待完成"}), 409
|
||
run_id = uuid.uuid4().hex[:8]
|
||
_run.update(id=run_id, state="running", prompt=prompt,
|
||
serial=(data.get("serial") or "").strip(),
|
||
answer="", error="")
|
||
# history 保留（同会话多轮对话），由前端「清空对话」调用 clear 重置
|
||
_queues[run_id] = queue.Queue()
|
||
_stop_events[run_id] = threading.Event()
|
||
_log.info(f"Agent 启动: {prompt[:60]}")
|
||
threading.Thread(target=_agent_thread,
|
||
args=(run_id, prompt, serial, cfg),
|
||
daemon=True).start()
|
||
return jsonify({"ok": True, "run_id": run_id})
|
||
|
||
|
||
@bp.route("/api/agent/stream")
|
||
@admin_required
|
||
def agent_stream():
|
||
"""SSE 事件流（EventSource）：delta/step/done/error。"""
|
||
run_id = request.args.get("run_id", "")
|
||
with _lock:
|
||
if run_id != _run["id"]:
|
||
return jsonify({"ok": False, "error": "run_id 不存在"}), 404
|
||
q = _queues.get(run_id)
|
||
|
||
def gen():
|
||
while True:
|
||
try:
|
||
evt = q.get(timeout=15)
|
||
except queue.Empty:
|
||
yield ": keepalive\n\n" # 心跳防超时
|
||
continue
|
||
if evt is None:
|
||
break
|
||
kind, payload = evt
|
||
yield f"event: {kind}\ndata: {json.dumps(payload, ensure_ascii=False)}\n\n"
|
||
if kind in ("done", "error"):
|
||
break
|
||
|
||
return Response(gen(), mimetype="text/event-stream",
|
||
headers={"Cache-Control": "no-cache",
|
||
"X-Accel-Buffering": "no"})
|
||
|
||
|
||
@bp.route("/api/agent/stop", methods=["POST"])
|
||
@admin_required
|
||
def agent_stop():
|
||
"""中断当前运行的 Agent（下一个检查点生效，通常在数秒内）。"""
|
||
with _lock:
|
||
if _run["state"] != "running":
|
||
return jsonify({"ok": False, "error": "当前没有运行中的任务"}), 400
|
||
evt = _stop_events.get(_run["id"])
|
||
if evt:
|
||
evt.set()
|
||
_log.info("用户请求中断 Agent")
|
||
return jsonify({"ok": True, "msg": "已请求停止"})
|
||
|
||
|
||
@bp.route("/api/agent/clear", methods=["POST"])
|
||
@admin_required
|
||
def agent_clear():
|
||
"""清空对话历史。"""
|
||
with _lock:
|
||
_run["history"] = []
|
||
return jsonify({"ok": True, "msg": "已清空"})
|
||
|
||
|
||
def _shrink_image(b64, width=220, quality=50):
|
||
"""截图降采样（SSE step 事件用，控制传输体积）。失败原样返回。"""
|
||
try:
|
||
from PIL import Image
|
||
img = Image.open(io.BytesIO(base64.b64decode(b64)))
|
||
if img.width > width:
|
||
img = img.resize((width, int(img.height * width / img.width)))
|
||
buf = io.BytesIO()
|
||
img.convert("RGB").save(buf, "JPEG", quality=quality)
|
||
return base64.b64encode(buf.getvalue()).decode()
|
||
except Exception:
|
||
return b64
|
||
|
||
|
||
def _agent_thread(run_id, prompt, serial, cfg):
|
||
"""后台线程：Agent 流式执行，事件推入队列供 SSE 消费。"""
|
||
q = _queues.get(run_id)
|
||
try:
|
||
# 平台进程 cwd=/app（含 mcp_agent 包），进程内 import
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
from mcp_agent.agent import Agent
|
||
|
||
def on_delta(text, kind):
|
||
q.put(("delta", {"text": text, "kind": kind}))
|
||
|
||
tool_seq = [] # 本轮工具序列（经验提炼用）
|
||
|
||
def on_tool(step):
|
||
rec = {"tool": step.get("tool"),
|
||
"args": str(step.get("args"))[:200]}
|
||
if step.get("image"):
|
||
rec["image"] = _shrink_image(step["image"])
|
||
q.put(("step", rec))
|
||
# 记录精简工具序列
|
||
try:
|
||
args = step.get("args") or {}
|
||
brief = {k: v for k, v in args.items() if k != "serial"}
|
||
tool_seq.append(f"{step.get('tool')}({str(brief)[:60]})")
|
||
except Exception:
|
||
pass
|
||
|
||
agent = Agent()
|
||
agent.s.api_base = cfg.get("api_base") or agent.s.api_base
|
||
agent.s.model = cfg.get("model") or agent.s.model
|
||
agent.s.api_key = cfg.get("api_key") or agent.s.api_key
|
||
agent.s.default_serial = cfg.get("default_serial") or agent.s.default_serial
|
||
|
||
with _lock:
|
||
history = list(_run.get("history") or [])
|
||
target = serial or cfg.get("default_serial") or ""
|
||
stop_evt = _stop_events.get(run_id)
|
||
|
||
# 经验检索：相似历史任务的操作配方注入 system（自进化记忆）
|
||
exp_ctx = _find_experiences(prompt)
|
||
if exp_ctx:
|
||
_log.info("命中历史经验，注入参考配方")
|
||
q.put(("step", {"tool": "🧠 经验记忆",
|
||
"args": f"命中 {exp_ctx.count(chr(10) + '- ')} 条同类历史经验，已注入参考",
|
||
"image": None}))
|
||
|
||
async def _execute():
|
||
await agent._load_tools()
|
||
return await agent.run_stream(prompt, target,
|
||
history=history,
|
||
on_delta=on_delta, on_tool=on_tool,
|
||
should_stop=lambda: bool(
|
||
stop_evt and stop_evt.is_set()),
|
||
extra_context=exp_ctx)
|
||
|
||
# 整体超时保护：卡死时结束，释放单实例
|
||
answer = asyncio.run(asyncio.wait_for(_execute(), timeout=900))
|
||
with _lock:
|
||
_run["state"] = "done"
|
||
_run["answer"] = answer
|
||
# 追加本轮进历史（多轮连续性；上限 12 轮防 token 膨胀）
|
||
hist = _run.setdefault("history", [])
|
||
hist.append({"role": "user", "content": prompt[:2000]})
|
||
hist.append({"role": "assistant", "content": (answer or "")[:4000]})
|
||
_run["history"] = hist[-24:]
|
||
|
||
# 自进化：成功执行过工具则提炼配方写入经验。必须在 done 之前完成——
|
||
# done 发出后 SSE 关流，用户就看不到「已写入经验」的提示了。
|
||
# 提炼/保存失败静默（不阻塞、不影响结果），只在成功时推送 🧠 卡片。
|
||
if tool_seq:
|
||
try:
|
||
recipe = _distill_experience(cfg, prompt, " -> ".join(tool_seq))
|
||
if recipe and "配方" not in recipe[:50]:
|
||
if _qualify_experience(prompt, recipe):
|
||
if _save_experience(prompt, recipe, " -> ".join(tool_seq)):
|
||
_log.info("经验已写入记忆库，随事件流提示")
|
||
q.put(("step", {"tool": "🧠 经验记忆",
|
||
"args": "本轮操作已提炼为经验并写入记忆库"
|
||
"（下次相似任务会自动参考）",
|
||
"image": None}))
|
||
except Exception as e:
|
||
_log.warning(f"经验保存异常: {e}")
|
||
q.put(("done", {"answer": answer}))
|
||
except Exception as e:
|
||
_log.warning(f"Agent 运行异常: {e}")
|
||
# 诊断：打印消息结构（tool_calls 与 tool 消息配对检查）
|
||
try:
|
||
roles = [m.get("role", "?") for m in agent.messages]
|
||
tcs = sum(1 for m in agent.messages
|
||
if m.get("tool_calls") and isinstance(m.get("tool_calls"), list))
|
||
tools_msg = sum(1 for m in agent.messages if m.get("role") == "tool")
|
||
_log.warning(f"诊断 messages: {len(agent.messages)} 条 roles={roles[-8:]} "
|
||
f"tool_calls消息={tcs} tool回应={tools_msg}")
|
||
except Exception:
|
||
pass
|
||
with _lock:
|
||
_run["state"] = "error"
|
||
_run["error"] = f"{type(e).__name__}: {str(e)[:200]}"
|
||
q.put(("error", {"message": str(e)[:200]}))
|
||
finally:
|
||
q.put(None) # 关闭 SSE
|
||
_queues.pop(run_id, None)
|
||
_stop_events.pop(run_id, None)
|