主动式 Agent 四件套

"""
主动式 Agent 四件套：调度 + 幂等 + 审批 + 核验
环境：Python 3.10+，依赖 pip install apscheduler pytz requests
环境变量：AIHCC_API_KEY（https://aihcc.cloud 注册即得）/ DEEPSEEK_API_KEY（可选，不配走离线演示）
"""
import datetime
import json
import os
import re
import sqlite3
import uuid

import pytz
import requests
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger

TZ = pytz.timezone('Asia/Shanghai')
DB = 'agent_tasks.db'
PENDING = 'pending_actions.jsonl'


# ---------- 配置：先加载 .env 再读变量（顺序反了就拿不到 key）----------
def load_env(path: str) -> bool:
    """极简 .env 加载器；已存在的环境变量优先，不覆盖。
    KEY= 空值行跳过：setdefault 会让它抢先占位，顶掉后面文件里真正的值。"""
    if not os.path.isfile(path):
        return False
    with open(path, encoding='utf-8') as f:
        for line in f:
            line = line.strip()
            if not line or line.startswith('#') or '=' not in line:
                continue
            k, v = line.split('=', 1)
            k, v = k.strip(), v.strip().strip('"\'')
            if not k or not v:
                continue
            os.environ.setdefault(k, v)
    return True


ENV_FILE = os.environ.get('ENV_FILE', '.env')
LOCAL_ENV = os.path.join(os.path.dirname(os.path.abspath(__file__)), '.env.local')
for _p in (LOCAL_ENV, ENV_FILE):      # 先加载本地：setdefault 不覆盖，靠前的生效
    load_env(_p)

AIHCC_BASE = os.environ.get('AIHCC_BASE') or 'https://aihcc.cloud/api'
LLM_BASE = os.environ.get('LLM_BASE') or 'https://api.deepseek.com/v1'
LLM_MODEL = os.environ.get('LLM_MODEL') or 'deepseek-v4-pro'


def _find_key(candidates, patterns):
    """先精确名，再正则兜底。返回 (变量名, 值)。
    每次调用都现取 os.environ——import 期固化成全局变量时 .env 还没加载。"""
    for name in candidates:
        v = os.environ.get(name)
        if v:
            return name, v
    for k, v in sorted(os.environ.items()):
        if v and any(re.fullmatch(p, k) for p in patterns):
            return k, v
    return '', ''


def llm_key():
    """LLM_BASE 指向哪家，就找那家的变量。改 LLM_BASE 即可换供应商。"""
    if 'deepseek' in LLM_BASE.lower():
        return _find_key(('DEEPSEEK_API_KEY',), ('DEEPSEEK.*KEY',))
    return _find_key(('OPENAI_API_KEY',), ('OPENAI.*KEY',))


def hallucc_key():
    return _find_key(('AIHCC_API_KEY', 'HALLUC_API_KEY'),
                     ('AIHCC.*KEY', 'HALLUC.*KEY'))


def now() -> datetime.datetime:
    """统一取 Asia/Shanghai 时间，别用裸的 datetime.now()。"""
    return datetime.datetime.now(TZ)


def notify(text: str):
    """推送通知：替换成企业微信/钉钉群机器人 webhook 即可。"""
    print(f'[NOTIFY] {text}')


# ---------- 四件套之四：核验（出口安检，HallucC 公开接口）----------
MAX_CHARS = 50_000     # /detect 的 text 上限，超出部分会被服务端截断


def verify_brief(text: str, strict: bool = False, domain: str = 'general') -> dict:
    """调 HallucC /detect 做逐句事实核验。
    契约要点（实测）：POST /detect（没有 /v1），鉴权 Authorization: Bearer hc_xxx；
    存疑数是 summary.red（没有 doubtful）；没有 annotated_text，标注要自己渲染。"""
    _, key = hallucc_key()
    resp = requests.post(f'{AIHCC_BASE}/detect',
                         headers={'Authorization': f'Bearer {key}'},
                         json={'text': text[:MAX_CHARS], 'strict': strict,
                               'domain': domain},
                         timeout=120)
    resp.raise_for_status()
    data = resp.json()
    # 200 但 body 是错误结构（额度/内部错）时，别当成「核验通过」
    if not isinstance(data.get('summary'), dict) or not data.get('claims'):
        raise RuntimeError(f"/detect 返回了非检测结构: "
                           f"{json.dumps(data, ensure_ascii=False)[:200]}")
    return data


MARKS = {'supported': '🟢', 'refuted': '🔴', 'unverified': '🟡'}


def compose_report(report: dict, raw: str) -> str:
    """把 /detect 响应渲染成可读晨报（接口本身不返回标注文本）。"""
    s = report.get('summary') or {}
    head = (f"晨报核验：共 {s.get('total', 0)} 句 ｜ "
            f"🟢可信 {s.get('green', 0)} 🟡疑似 {s.get('yellow', 0)} "
            f"🔴存疑 {s.get('red', 0)} ｜ 幻觉风险 {s.get('hallucination_risk', 0):.2f}")
    lines = [head, '', raw, '', '—— 逐句核验 ——']
    for c in report.get('claims') or []:
        v = c.get('verification') or {}
        mark = MARKS.get(v.get('status'), '⚪')
        line = f"{mark} {c.get('claim', {}).get('text', '')}" \
               f"（把握 {v.get('confidence', 0):.0%}）"
        srcs = v.get('sources') or []
        if srcs:
            line += f" 依据：{srcs[0].get('title', '')}"
        fix = (c.get('correction') or {}).get('corrected')
        if fix:
            line += f"\n  建议改为：{fix}"
        lines.append(line)
    return '\n'.join(lines)


# ---------- LLM（OpenAI 兼容；未配 key 返回离线演示文本，先跑通骨架）----------
def agent_run(prompt: str, output_json: bool = False):
    """返回类型与 output_json 一致：True -> dict，False -> str。"""
    _, key = llm_key()
    if not key:
        off = ('{"level":"alert","summary":"订单积压 37 单，超过阈值 20"}' if output_json
               else '今日要闻：（离线演示文本，未配置 LLM key）')
        return json.loads(off) if output_json else off
    payload = {'model': LLM_MODEL, 'temperature': 0.2,
               'messages': [{'role': 'user', 'content': prompt}]}
    if output_json:
        payload['response_format'] = {'type': 'json_object'}
    resp = requests.post(f'{LLM_BASE}/chat/completions',
                         headers={'Authorization': f'Bearer {key}'},
                         json=payload, timeout=60)
    resp.raise_for_status()
    content = resp.json()['choices'][0]['message']['content']
    return json.loads(content) if output_json else content


# ---------- 四件套之二：幂等（SQLite 原子占坑）----------
def init_db():
    with sqlite3.connect(DB) as conn:
        conn.execute('''CREATE TABLE IF NOT EXISTS task_log(
            task TEXT, idem_key TEXT,
            status TEXT DEFAULT 'done',
            created_at TEXT DEFAULT (datetime('now','localtime')),
            UNIQUE(task, idem_key)
        )''')
        conn.commit()


def acquire_lock(task: str, idem_key: str) -> bool:
    """原子抢占：INSERT 成功=拿到执行权；UNIQUE 冲突=已执行过。"""
    conn = sqlite3.connect(DB)
    try:
        cur = conn.execute('INSERT OR IGNORE INTO task_log(task, idem_key) VALUES(?,?)',
                           (task, idem_key))
        conn.commit()
        return cur.rowcount == 1
    finally:
        conn.close()          # with sqlite3.connect() 只 commit/rollback，不负责 close


def brief_key() -> str:
    return f'brief-{now().date().isoformat()}'     # 粒度到天：一天最多一份晨报


def patrol_key() -> str:
    return f"patrol-{now().strftime('%Y%m%d%H%M')}"  # 粒度到分钟：窗口过了就翻篇


def mark_failed(task: str, idem_key: str) -> None:
    """执行失败时释放幂等锁，让下一个窗口有机会重试。
    否则「抢到锁 -> 中途崩了」会变成「任务永久跳过」，而且没有任何提示。"""
    conn = sqlite3.connect(DB)
    try:
        conn.execute("DELETE FROM task_log WHERE task=? AND idem_key=? AND status='done'",
                     (task, idem_key))
        conn.commit()
    finally:
        conn.close()


# ---------- 四件套之三：审批（不可逆动作转人工工单）----------
RISK_LEVELS = {'query_data': 'low', 'generate_report': 'low',
               'send_email': 'high', 'delete_file': 'high', 'update_order': 'high'}
HANDLERS = {}


def handler(name):
    """装饰器：登记 Agent 可调用的动作实现。"""
    def deco(fn):
        HANDLERS[name] = fn
        return fn
    return deco


@handler('query_data')
def _query_data():
    return {'rows': [], 'note': '接你自己的数据源'}


@handler('generate_report')
def _generate_report(title=''):
    return {'report': f'报告：{title}'}


@handler('send_email')
def _send_email(to='', subject='', body=''):
    return {'sent': False, 'to': to, 'note': '演示：不真的发邮件'}


@handler('delete_file')
def _delete_file(path=''):
    return {'deleted': False, 'path': path, 'note': '演示：不真的删'}


@handler('update_order')
def _update_order(order_id=None, status=''):
    return {'updated': False, 'order_id': order_id}


def guarded_call(agent_id: str, tool: str, args: dict):
    """Agent 所有动作的统一出口：低危直接执行，高危转人工审批。"""
    if RISK_LEVELS.get(tool, 'high') == 'high':     # 未登记默认高危
        return request_approval(agent_id, tool, args)
    if tool not in HANDLERS:                        # 登记成 low 却没实现 -> 别崩
        return {'error': f'no handler for low-risk tool {tool!r}'}
    return HANDLERS[tool](**args)


def request_approval(agent_id: str, tool: str, args: dict) -> str:
    """高危动作不执行，落盘成待审批工单并推送提醒，返回工单号。"""
    ticket = uuid.uuid4().hex[:8]
    record = {'id': ticket, 'agent': agent_id, 'tool': tool, 'args': args,
              'status': 'pending', 'created_at': now().isoformat(timespec='seconds')}
    with open(PENDING, 'a', encoding='utf-8') as f:
        f.write(json.dumps(record, ensure_ascii=False) + '\n')
    notify(f'【待审批 {ticket}】{tool} {args}')
    return ticket


def review(ticket: str, approve: bool) -> str:
    """人工审批入口：只有 pending 状态才真正放行，状态全程留痕。"""
    if not os.path.exists(PENDING):
        return 'not_found'
    with open(PENDING, encoding='utf-8') as f:
        entries = [json.loads(line) for line in f if line.strip()]
    outcome = 'not_found'
    for e in entries:
        if e['id'] != ticket:
            continue
        if e['status'] != 'pending':
            outcome = f"already_{e['status']}"        # 防重复审批
            break
        if approve:
            if e['tool'] not in HANDLERS:
                e['status'] = 'failed'                # 放行但没人执行 -> 别假装成功
                outcome = e['status']
                break
            HANDLERS[e['tool']](**e['args'])
            e['status'] = 'approved'
        else:
            e['status'] = 'rejected'
        e['reviewed_at'] = now().isoformat(timespec='seconds')
        outcome = e['status']
        break
    # 原子替换：先写临时文件再 rename，进程中途挂掉不会把审计日志截断
    tmp = PENDING + '.tmp'
    with open(tmp, 'w', encoding='utf-8') as f:
        for e in entries:
            f.write(json.dumps(e, ensure_ascii=False) + '\n')
    os.replace(tmp, PENDING)
    return outcome


# ---------- 主动任务：晨报（核验后才发）+ 巡检 ----------
def morning_brief():
    key = brief_key()
    if not acquire_lock('morning_brief', key):
        return                          # 幂等锁：今天已生成过，直接跳过
    try:
        brief = agent_run('汇总今日行业要闻与团队待办，生成 300 字晨报')
        if not isinstance(brief, str):
            raise RuntimeError(f'LLM 返回了非文本结果: {type(brief).__name__}')
    except Exception as exc:
        mark_failed('morning_brief', key)   # 抢了锁却没跑成 -> 释放，允许重试
        notify(f'[晨报生成失败] {exc.__class__.__name__}: {exc}')
        return

    try:
        report = verify_brief(brief)
        if (report.get('summary') or {}).get('total', 0) == 0:
            # 一句话都没抽出来 = 其实没核验过，不能当「核验通过」
            text = f'⚠️ 本次未抽到任何可核验声明，本晨报未经有效核验\n\n{brief}'
        else:
            text = compose_report(report, brief)
            risky = (report.get('summary') or {}).get('red', 0)
            if risky:
                text = (f'⚠️ 本晨报含 {risky} 句存疑内容，已逐句标注，请过目后再转发\n\n{text}')
    except (requests.RequestException, ValueError, RuntimeError,
            PermissionError, OSError) as exc:
        # 核验不可用：不静默放行，明确标注「未经核验」。
        # ValueError 必须收——resp.json() 抛的 JSONDecodeError 是 ValueError 子类，
        # 不是 requests.RequestException（实测 issubclass 为 False）。
        text = f'⚠️ 核验服务不可用（{exc.__class__.__name__}），本晨报未经核验\n\n{brief}'
    notify(text)


def patrol():
    """每 15 分钟巡检：Agent 查数据源，仅触发阈值时才通知。"""
    key = patrol_key()
    if not acquire_lock('patrol', key):
        return
    try:
        result = agent_run('巡检数据源，异常时输出 {"level":"alert","summary":"..."}',
                           output_json=True)
    except Exception:
        mark_failed('patrol', key)
        return
    if not isinstance(result, dict):
        mark_failed('patrol', key)
        return
    if result.get('level') == 'alert':
        notify(f"[巡检告警] {result.get('summary', '')}")


# ---------- 四件套之一：调度（时区钉死，任务防堆叠）----------
def build_scheduler() -> BlockingScheduler:
    scheduler = BlockingScheduler(timezone=TZ)
    scheduler.add_job(morning_brief, CronTrigger(hour=8, minute=30, timezone=TZ),
                      id='brief', max_instances=1, coalesce=True)
    scheduler.add_job(patrol, 'interval', minutes=15, timezone=TZ,
                      id='patrol', max_instances=1, coalesce=True)
    return scheduler


def next_fire(trigger, prev=None, now_dt=None):
    """触发器自身的下一次触发时刻——start() 之前也能算。
    别用 get_job('brief').next_run_time：job 在 start() 前是 pending 的，会 AttributeError。"""
    return trigger.get_next_fire_time(prev, now_dt or now())


if __name__ == '__main__':
    init_db()
    scheduler = build_scheduler()
    print('主动式 Agent 已启动：晨报每天 08:30（Asia/Shanghai），巡检每 15 分钟')
    scheduler.start()      # 阻塞式启动，适合单职责的常驻进程