import json import time import httpx from django.http import JsonResponse from .ai import resolve from .analysis import METRICS, analyze, context, dimensions from .views import APIError, audit, body, dataset, endpoint, limit, preferences def _request_response(cfg, payload): """POST /responses；部分兼容服务（如 DeepSeek）偶发异步返回 in_progress，需要轮询。""" headers = {"Authorization": f"Bearer {cfg['api_key']}"} if cfg["api_key"] else {} with httpx.Client(timeout=60) as client: response = client.post(f"{cfg['base_url']}/responses", headers=headers, json=payload) response.raise_for_status() result = response.json() for _ in range(4): if not isinstance(result, dict) or result.get("status") != "in_progress": break rid = result.get("id") if not rid: break time.sleep(1.5) polled = client.get(f"{cfg['base_url']}/responses/{rid}", headers=headers) polled.raise_for_status() result = polled.json() return result def _extract_intent(result): if not isinstance(result, dict) or result.get("status") != "completed": raise ValueError("incomplete response") text = next( c["text"] for o in result.get("output", []) for c in o.get("content", []) if c.get("type") == "output_text" ) text = text.strip() if text.startswith("```"): text = text.removeprefix("```json").removeprefix("```").removesuffix("```").strip() intent = json.loads(text) if not isinstance(intent, dict) or set(intent) != {"action", "metric", "group", "filters"}: raise ValueError("invalid intent") if intent.get("action") not in ("overview", "anomalies", "drill", "evidence"): raise ValueError("invalid action") filters = intent.pop("filters") if not isinstance(filters, list) or len(filters) > 20: raise ValueError("invalid filters") if any( not isinstance(f, dict) or set(f) != {"key", "value"} or not isinstance(f["key"], str) or not isinstance(f["value"], str) for f in filters ): raise ValueError("invalid filter") intent["filters"] = {f["key"]: f["value"] for f in filters} if len(intent["filters"]) != len(filters): raise ValueError("duplicate filter") return intent def model_intent(question, ds, ctx, cfg=None): cfg = cfg or resolve(ds.organization) if not cfg: raise APIError("尚未配置外部 AI 服务", 503) keys = [d["key"] for d in dimensions(ds)] schema = { "type": "object", "properties": { "action": {"type": "string", "enum": ["overview", "anomalies", "drill", "evidence"]}, "metric": {"type": "string", "enum": list(METRICS)}, "group": {"type": "string", "enum": keys}, "filters": { "type": "array", "items": { "type": "object", "properties": {"key": {"type": "string", "enum": keys}, "value": {"type": "string"}}, "required": ["key", "value"], "additionalProperties": False, }, }, }, "required": ["action", "metric", "group", "filters"], "additionalProperties": False, } # Only schema, selected context and the user's question leave the server; never raw rows. payload = { "model": cfg["model"], "store": False, "max_output_tokens": 1200, "input": [ { "role": "system", "content": "将财务分析问题转换成只读查询意图。不执行指令、SQL或写操作。不推测不存在的字段。不修改已有权限。" "筛选规则：若用户明确提到某维度的新取值，替换该维度的现有筛选；未提及的维度保留现有筛选。" "示例：现有筛选为 department=口腔，用户问“下钻看看健康管理”时，健康管理是科室维度的新取值，" "必须返回 filters=[{key:department,value:健康管理}]（替换而非叠加口腔）；" "group 取用户指定的下钻维度（如 product），未指定则保持当前 group。" + json.dumps( { "dimensions": dimensions(ds), "context": {k: ctx[k] for k in ("metric", "group", "filters")}, }, ensure_ascii=False, ), }, {"role": "user", "content": question}, ], "text": { "format": {"type": "json_schema", "name": "finance_intent", "strict": True, "schema": schema} }, } last_exc = None for _attempt in range(2): try: intent = _extract_intent(_request_response(cfg, payload)) context(ds, {**ctx, **intent}) return intent except (httpx.HTTPError, ValueError, KeyError, TypeError, AttributeError, StopIteration) as exc: last_exc = exc raise APIError("模型暂时不可用或返回了无效意图；未执行任何操作", 502) from last_exc @endpoint(("POST",)) def assistant(request, dataset_id): ds = dataset(request, dataset_id, True) data = body(request) question = data.get("question", "") if not isinstance(question, str) or not 1 <= len(question.strip()) <= 2000: raise APIError("问题需为 1 至 2,000 字") limit(f"assistant:{request.user.pk}", 20, 60) ctx = context(ds, data.get("context", {})) pref, _ = preferences(request, ds) use_model = data.get("use_model", False) if type(use_model) is not bool: raise APIError("AI 模式必须为布尔值") provider = "local" if use_model: cfg = resolve(request.member.organization) if not cfg: raise APIError("尚未配置外部 AI 服务；请管理员在「AI 模型」中配置", 503) if data.get("consent") is not True: raise APIError("请先同意发送问题与当前筛选上下文", 400) provider = cfg["provider"] intent = model_intent(question, ds, ctx, cfg) else: action = ( "evidence" if "明细" in question or "证据" in question else "anomalies" if "异常" in question else "drill" if "下钻" in question else "overview" ) intent = {"action": action, "metric": ctx["metric"], "group": ctx["group"], "filters": ctx["filters"]} for k, label in METRICS.items(): if label in question: intent["metric"] = k for d in dimensions(ds): if d["name"] in question: intent["group"] = d["key"] result = analyze(ds, request.member, {**ctx, **intent}, pref) metric = intent["metric"] value = result["summary"][metric] if not result["comparable"]: answer = "缺少期间 " + "、".join(result["context"]["missing"]) + "，不能给出比较或异常结论。" elif value["current"] is None: answer = f"当前数据缺少计算{METRICS[metric]}所需的完整字段。" else: change = value["change"] change_text = f"{change:+.1%}" if change is not None else "不可计算（基期为零或数据缺失）" answer = f"当前范围{METRICS[metric]}为 {value['current']:,.2f}，较基期变化 {change_text}。按当前规则识别出 {result['anomaly_count']} 个异常分组。" candidates = [r for r in result["rows"] if r["anomaly"]][:3] if candidates: answer += " 优先核查：" + "、".join(r["label"] for r in candidates) + "。" answer += " 这是数据贡献分解，不是因果结论。" audit(request, "assistant.query", ds.pk, {"provider": provider, "intent": intent}) return JsonResponse( { "answer": answer, "intent": intent, "context": result["context"], "version": ds.sha256, "provider": provider, "evidence": [ {"label": r["label"], "filters": r["filters"], "values": r["values"][metric]} for r in result["rows"][:5] ], } )