Jayway Liu
3c3799910d
- assistant 解析兼容：支持 in_progress 异步轮询（DeepSeek 偶发）、 去除 markdown 代码围栏、整体失败重试一次后才报 502 - 系统提示词增加筛选替换规则与示例：用户明确提到某维度新取值时替换 现有筛选而非叠加（实测修复“口腔+健康管理”空范围问题） - KPI 数字字号改为容器查询 clamp(13px,1.9cqw,21px)，窄宽度下 完整显示金额而非省略号截断 - 新增 in_progress 轮询 + 重试的单测（test_assistant_api） 验证：pytest 82 passed；DeepSeek 真实调用 5/5 成功（异常/证据/ 筛选替换三种意图）；Playwright 回归通过
196 lines
8.2 KiB
Python
196 lines
8.2 KiB
Python
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]
|
|
],
|
|
}
|
|
)
|