435 lines
17 KiB
Python
435 lines
17 KiB
Python
import json
|
||
import time
|
||
|
||
import httpx
|
||
from django.core.serializers.json import DjangoJSONEncoder
|
||
from django.http import JsonResponse
|
||
|
||
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
|
||
|
||
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):
|
||
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 ACTIONS:
|
||
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, history=None, deadline=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": list(ACTIONS)},
|
||
"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,
|
||
}
|
||
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": system},
|
||
*(history or []),
|
||
{"role": "user", "content": question},
|
||
],
|
||
"text": {
|
||
"format": {"type": "json_schema", "name": "finance_intent", "strict": True, "schema": schema}
|
||
},
|
||
}
|
||
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, deadline=deadline))
|
||
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
|
||
|
||
|
||
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)
|
||
data = body(request)
|
||
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)
|
||
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 模式必须为布尔值")
|
||
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:
|
||
raise APIError("尚未配置外部 AI 服务；请管理员在「AI 模型」中配置", 503)
|
||
if data.get("consent") is not True:
|
||
raise APIError("请先同意发送问题与当前筛选上下文", 400)
|
||
provider = cfg["provider"]
|
||
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
|
||
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, "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
|