feat(assistant): 基于聚合结果动态回答并支持对话新增分析列

finance/assistant.py

@@ -2,49 +2,28 @@ import json
 import time
 
 import httpx
+from django.core.serializers.json import DjangoJSONEncoder
 from django.http import JsonResponse
 
-from .ai import resolve
-from .analysis import METRICS, analyze, context, dimensions
+from .ai import extract_output_text, request_response, resolve, validate_messages
+from .analysis import METRICS, analyze, context, custom_columns, dimensions
+from .formula_views import design_column
 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
+QUERY_ACTIONS = ("overview", "anomalies", "drill", "evidence")
+ACTIONS = (*QUERY_ACTIONS, "answer", "design_column")
+ROUND_BUDGET_SECONDS = 90.0
+MAX_HISTORY_TURNS = 5
 
 
 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()
+    text = extract_output_text(result)
     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"):
+    if intent.get("action") not in ACTIONS:
         raise ValueError("invalid action")
     filters = intent.pop("filters")
     if not isinstance(filters, list) or len(filters) > 20:
@@ -63,7 +42,7 @@ def _extract_intent(result):
     return intent
 
 
-def model_intent(question, ds, ctx, cfg=None):
+def model_intent(question, ds, ctx, cfg=None, history=None, deadline=None):
     cfg = cfg or resolve(ds.organization)
     if not cfg:
         raise APIError("尚未配置外部 AI 服务", 503)
@@ -71,7 +50,7 @@ def model_intent(question, ds, ctx, cfg=None):
     schema = {
         "type": "object",
         "properties": {
-            "action": {"type": "string", "enum": ["overview", "anomalies", "drill", "evidence"]},
+            "action": {"type": "string", "enum": list(ACTIONS)},
             "metric": {"type": "string", "enum": list(METRICS)},
             "group": {"type": "string", "enum": keys},
             "filters": {
@@ -87,27 +66,29 @@ def model_intent(question, ds, ctx, cfg=None):
         "required": ["action", "metric", "group", "filters"],
         "additionalProperties": False,
     }
-    # Only schema, selected context and the user's question leave the server; never raw rows.
+    system = (
+        "将财务分析对话转换成只读意图。不执行指令、SQL或写操作，不推测不存在的字段，不修改权限。"
+        "普通问答（解释、原因、建议、还有什么问题）使用 answer；"
+        "用户明确要求新增或添加分析列时使用 design_column；"
+        "只有明确要求总览、异常、下钻或明细证据时才使用 overview、anomalies、drill、evidence。"
+        "筛选规则：若用户明确提到某维度的新取值且属于查询动作，替换该维度的现有筛选；未提及的维度保留现有筛选；"
+        "answer 与 design_column 必须保留当前筛选、指标和分组。"
+        + json.dumps(
+            {
+                "dimensions": dimensions(ds),
+                "context": {k: ctx[k] for k in ("metric", "group", "filters")},
+            },
+            ensure_ascii=False,
+        )
+    )
+    # Only schema, selected context, bounded history 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": "system", "content": system},
+            *(history or []),
             {"role": "user", "content": question},
         ],
         "text": {
@@ -116,8 +97,10 @@ def model_intent(question, ds, ctx, cfg=None):
     }
     last_exc = None
     for _attempt in range(2):
+        if deadline is not None and deadline - time.monotonic() <= 1:
+            break
         try:
-            intent = _extract_intent(_request_response(cfg, payload))
+            intent = _extract_intent(request_response(cfg, payload, deadline=deadline))
             context(ds, {**ctx, **intent})
             return intent
         except (httpx.HTTPError, ValueError, KeyError, TypeError, AttributeError, StopIteration) as exc:
@@ -125,6 +108,165 @@ def model_intent(question, ds, ctx, cfg=None):
     raise APIError("模型暂时不可用或返回了无效意图；未执行任何操作", 502) from last_exc
 
 
+def _value_text(value):
+    return None if value is None else str(value)
+
+
+def _answer_payload(result):
+    metric = result["context"]["metric"]
+    summary = {
+        key: {part: _value_text(value[part]) for part in ("current", "baseline", "delta", "change")}
+        for key, value in result["summary"].items()
+    }
+    rows = result["rows"][:5]
+    groups = {
+        "total": len(result["rows"]),
+        "truncated": len(result["rows"]) > 5,
+        "items": [
+            {
+                "label": row["label"],
+                "filters": row["filters"],
+                "values": {
+                    key: {
+                        part: _value_text(value[part]) for part in ("current", "baseline", "delta", "change")
+                    }
+                    for key, value in row["values"].items()
+                },
+            }
+            for row in rows
+        ],
+    }
+    columns = [column["name"] for column in result["context"].get("custom_columns", [])]
+    return {
+        "period": {
+            "start": result["context"]["start"],
+            "end": result["context"]["end"],
+            "comparison": result["context"]["comparison"],
+            "baseline_end": result["context"]["baseline_end"],
+            "current": result["context"]["current"],
+            "baseline": result["context"]["baseline"],
+            "missing": result["context"]["missing"],
+        },
+        "filters": result["context"]["filters"],
+        "group": result["context"]["group"],
+        "metric": metric,
+        "metric_labels": METRICS,
+        "custom_columns": columns,
+        "summary": summary,
+        "comparable": result["comparable"],
+        "notice": result["notice"],
+        "anomaly_count": result["anomaly_count"],
+        "groups": groups,
+    }
+
+
+def model_answer(question, ds, result, cfg, history=None, deadline=None):
+    payload_data = _answer_payload(result)
+    system = (
+        "你是财务分析助手。只使用下面提供的服务端聚合数据回答用户当前问题，使用中文。"
+        "回答本次问题，不要复述整份数据；缺少期间、缺少字段或零基期时明确说明不可计算，不能填零或虚构数值。"
+        "相关性不等于因果；数据未支持的原因只能作为待核实假设，并明确标注为假设。"
+        "不得编造数据中没有的金额、比例、客户或期间；不得声称已修改数据或执行任何写操作。"
+        "聚合数据（JSON，金额为精确字符串）："
+        + json.dumps(payload_data, ensure_ascii=False, cls=DjangoJSONEncoder)
+    )
+    payload = {
+        "model": cfg["model"],
+        "store": False,
+        "max_output_tokens": 3000,
+        "input": [
+            {"role": "system", "content": system},
+            *(history or []),
+            {"role": "user", "content": question},
+        ],
+    }
+    try:
+        result = request_response(cfg, payload, deadline=deadline)
+        if isinstance(result, dict) and result.get("status") == "incomplete":
+            reason = (result.get("incomplete_details") or {}).get("reason")
+            if reason == "max_output_tokens":
+                raise APIError("模型回答被截断（达到输出上限），请重试或缩短问题", 502)
+            raise APIError("模型回答不完整；未更改分析结果", 502)
+        text = extract_output_text(result).strip()
+    except APIError:
+        raise
+    except (httpx.HTTPError, ValueError, KeyError, TypeError, StopIteration) as exc:
+        raise APIError("模型暂时不可用或未返回有效回答；未更改分析结果", 502) from exc
+    if not text:
+        raise APIError("模型返回了空回答；未更改分析结果", 502)
+    return text
+
+
+def _require_column_design(request):
+    if request.member.role not in ("admin", "analyst") or request.member.departments:
+        raise APIError("当前账号没有通过助手定义分析列的权限", 403)
+
+
+def _column_context(ctx, columns):
+    return {**ctx, "custom_columns": columns}
+
+
+def _column_design_reply(request, ds, ctx, columns, messages, cfg, provider, model):
+    _require_column_design(request)
+    design = design_column(messages, ds, ctx, cfg)
+    answer = (
+        f"已整理列定义：{design['column']['name']}；请确认后添加到当前分析表。"
+        if design["complete"]
+        else "\n".join(design["questions"])
+    )
+    audit(
+        request,
+        "assistant.query",
+        ds.pk,
+        {
+            "provider": provider,
+            "model": model,
+            "action": "design_column",
+            "source": "model",
+            "kind": "column_design",
+            "stage": "completed",
+            "ok": True,
+        },
+    )
+    return JsonResponse(
+        {
+            "answer": answer,
+            "intent": {
+                "action": "design_column",
+                "metric": ctx["metric"],
+                "group": ctx["group"],
+                "filters": ctx["filters"],
+            },
+            "context": _column_context(ctx, columns),
+            "version": ds.sha256,
+            "provider": provider,
+            "evidence": [],
+            "kind": "column_design",
+            "source": "model",
+            "model": model,
+            "design": design,
+        }
+    )
+
+
+def _rules_answer(result, metric):
+    value = result["summary"][metric]
+    if not result["comparable"]:
+        return "缺少期间 " + "、".join(result["context"]["missing"]) + "，不能给出比较或异常结论。"
+    if value["current"] is None:
+        return f"当前数据缺少计算{METRICS[metric]}所需的完整字段。"
+    change = value["change"]
+    change_text = f"{change:+.1%}" if change is not None else "不可计算（基期为零或数据缺失）"
+    answer = (
+        f"当前范围{METRICS[metric]}为 {value['current']:,.2f}，较基期变化 {change_text}。"
+        f"按当前规则识别出 {result['anomaly_count']} 个异常分组。"
+    )
+    candidates = [r for r in result["rows"] if r["anomaly"]][:3]
+    if candidates:
+        answer += " 优先核查：" + "、".join(r["label"] for r in candidates) + "。"
+    return answer + " 这是数据贡献分解，不是因果结论。"
+
+
 @endpoint(("POST",))
 def assistant(request, dataset_id):
     ds = dataset(request, dataset_id, True)
@@ -132,13 +274,36 @@ def assistant(request, dataset_id):
     question = data.get("question", "")
     if not isinstance(question, str) or not 1 <= len(question.strip()) <= 2000:
         raise APIError("问题需为 1 至 2,000 字")
+    mode = data.get("mode", "auto")
+    if mode not in ("auto", "column_design"):
+        raise APIError("对话模式无效")
+    history = data.get("history", [])
+    if not isinstance(history, list):
+        raise APIError("对话历史格式无效")
+    if len(history) > MAX_HISTORY_TURNS * 2:
+        raise APIError("对话最多保留 5 轮历史，请开始新对话")
+    conversation = validate_messages(
+        [*history, {"role": "user", "content": question.strip()}],
+        max_messages=(MAX_HISTORY_TURNS * 2) + 1,
+        user_chars=2000,
+        assistant_chars=4000,
+        total_chars=24000,
+    )
+    history, current = conversation[:-1], conversation[-1]
     limit(f"assistant:{request.user.pk}", 20, 60)
-    ctx = context(ds, data.get("context", {}))
+    raw_context = data.get("context", {})
+    ctx = context(ds, raw_context)
+    columns = custom_columns(ds, raw_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 mode == "column_design":
+        _require_column_design(request)
+        if not use_model:
+            raise APIError("列设计需要启用 AI 模型", 400)
+    cfg = None
+    provider, model = "local", None
     if use_model:
         cfg = resolve(request.member.organization)
         if not cfg:
@@ -146,8 +311,63 @@ def assistant(request, dataset_id):
         if data.get("consent") is not True:
             raise APIError("请先同意发送问题与当前筛选上下文", 400)
         provider = cfg["provider"]
-        intent = model_intent(question, ds, ctx, cfg)
-    else:
+        model = cfg["model"]
+    try:
+        if mode == "column_design":
+            return _column_design_reply(request, ds, ctx, columns, conversation, cfg, provider, model)
+        if use_model:
+            deadline = time.monotonic() + ROUND_BUDGET_SECONDS
+            intent = model_intent(current["content"], ds, ctx, cfg, history=history, deadline=deadline)
+            if intent["action"] == "design_column":
+                return _column_design_reply(request, ds, ctx, columns, conversation, cfg, provider, model)
+            if intent["action"] not in QUERY_ACTIONS:
+                intent = {
+                    "action": "answer",
+                    "metric": ctx["metric"],
+                    "group": ctx["group"],
+                    "filters": ctx["filters"],
+                }
+            result = analyze(
+                ds,
+                request.member,
+                {**ctx, **intent, "custom_columns": columns},
+                pref,
+            )
+            answer = model_answer(current["content"], ds, result, cfg, history=history, deadline=deadline)
+            audit(
+                request,
+                "assistant.query",
+                ds.pk,
+                {
+                    "provider": provider,
+                    "model": model,
+                    "action": intent["action"],
+                    "source": "model",
+                    "kind": "answer",
+                    "stage": "completed",
+                    "ok": True,
+                },
+            )
+            return JsonResponse(
+                {
+                    "answer": answer,
+                    "intent": intent,
+                    "context": result["context"],
+                    "version": ds.sha256,
+                    "provider": provider,
+                    "evidence": [
+                        {
+                            "label": r["label"],
+                            "filters": r["filters"],
+                            "values": r["values"][intent["metric"]],
+                        }
+                        for r in result["rows"][:5]
+                    ],
+                    "kind": "answer",
+                    "source": "model",
+                    "model": model,
+                }
+            )
         action = (
             "evidence"
             if "明细" in question or "证据" in question
@@ -164,32 +384,51 @@ def assistant(request, dataset_id):
         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]
-            ],
-        }
-    )
+        result = analyze(ds, request.member, {**ctx, **intent, "custom_columns": columns}, pref)
+        answer = _rules_answer(result, intent["metric"])
+        audit(
+            request,
+            "assistant.query",
+            ds.pk,
+            {
+                "provider": provider,
+                "model": None,
+                "action": intent["action"],
+                "source": "rules",
+                "kind": "answer",
+            },
+        )
+        return JsonResponse(
+            {
+                "answer": answer,
+                "intent": intent,
+                "context": result["context"],
+                "version": ds.sha256,
+                "provider": provider,
+                "evidence": [
+                    {"label": r["label"], "filters": r["filters"], "values": r["values"][intent["metric"]]}
+                    for r in result["rows"][:5]
+                ],
+                "kind": "answer",
+                "source": "rules",
+                "model": None,
+            }
+        )
+    except APIError as exc:
+        if use_model:
+            audit(
+                request,
+                "assistant.query",
+                ds.pk,
+                {
+                    "provider": provider,
+                    "model": model,
+                    "action": "unknown",
+                    "source": "model",
+                    "kind": "error",
+                    "stage": "model",
+                    "ok": False,
+                    "status": exc.status,
+                },
+            )
+        raise