mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:21:58 +00:00
起草按「能共用一条 SQL 的一组」进行,并抢救被截断的回复
实测:一条 SQL 枚举十几列 CASE WHEN 就把输出额度吃光,文档里十几条指标一条都写不出来, 界面只报「没有一条指标通过校验」。现在提示模型一次只起草同组指标(最多 6 条)、SQL 写短, 其余的进 uncovered;回复被截断时按完整片段抢救指标和 SQL,失败原因里写清模型实际交回了 什么形状,不再只说没有指标。
This commit is contained in:
@@ -661,9 +661,13 @@ async def _extract_in_background(draft_id: str) -> None:
|
||||
"sql": plan["sql"], "version": 1,
|
||||
"trial": {"run_id": "", "status": "running", "error": ""},
|
||||
"dropped": problems, "truncated": result["truncated"], "model_id": result["model_id"],
|
||||
"open_questions": plan["open_questions"], "uncovered": result.get("uncovered") or [],
|
||||
"salvaged": bool(result.get("salvaged")),
|
||||
}
|
||||
if not plan["sql"] or not plan["metrics"]:
|
||||
error = _problems_text(problems) or "模型没有给出可用的取数 SQL 和指标"
|
||||
if result.get("diagnostics"):
|
||||
error = f"{error}({result['diagnostics']})"
|
||||
draft.payload = {**base, "trial": {"run_id": "", "status": "failed", "error": error}}
|
||||
draft.model_id = result["model_id"] or None
|
||||
draft.error = error
|
||||
|
||||
@@ -579,8 +579,12 @@ def merge_definition(card_metrics: list[dict], draft_metrics: list[dict]) -> tup
|
||||
return merged, warnings
|
||||
|
||||
|
||||
#: 被截断的回复里抢回已经完整的指标对象(draftplan 生成取数草稿时也要用)
|
||||
salvage_metrics = _salvage_metrics
|
||||
|
||||
|
||||
__all__ = [
|
||||
"DEFINITION_FIELDS", "DraftUnavailable", "FORBIDDEN_FIELDS", "MAX_INPUT_CHARS", "METRIC_FIELDS",
|
||||
"MIN_QUOTE_CHARS", "STATUSES", "alignment", "extract", "merge_definition", "normalize_metrics",
|
||||
"parse_reply", "prompt", "verify_quote",
|
||||
"parse_reply", "prompt", "salvage_metrics", "verify_quote",
|
||||
]
|
||||
|
||||
@@ -23,6 +23,7 @@ build_graph),单行结果取第 0 行的某一列;写成别的样子,卡
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from collections.abc import Mapping
|
||||
from typing import Any
|
||||
@@ -59,6 +60,11 @@ _SYSTEM = """你是业务口径与取数模板的起草助手。读给定的业
|
||||
5. 每条指标的定义必须带 citations:引用逐字抄原文,chunk 写材料里方括号内的 id。核对不上、或原文里没有依据的定义
|
||||
不要写进指标,写进 open_questions。
|
||||
6. 拿不准的不要猜,写进 open_questions,用一句问得清的话。不要编造文档里没有的数字、单位、生效日期。
|
||||
7. **一次只起草能共用同一条 SQL 的一组指标,最多 6 条。** 文档里其余的指标不要硬凑进这一条 SQL,把名称写进
|
||||
uncovered(字符串数组),下一轮再起草。不同粒度、不同表、不同时间字段的指标本来就不该共用一条 SQL。
|
||||
8. **sql 要短。** 能用 GROUP BY 和聚合表达的,不要为每个属性枚举一长串 CASE WHEN 列;输出额度有限,SQL 写长了,
|
||||
后面的指标就写不出来了。
|
||||
9. 写不下时宁可少写几条指标、把剩下的写进 uncovered,**也不要让 JSON 半截断掉**。
|
||||
|
||||
输出格式:
|
||||
{
|
||||
@@ -85,7 +91,8 @@ _SYSTEM = """你是业务口径与取数模板的起草助手。读给定的业
|
||||
"open_questions": ["还没说清、需要业务确认的问题"]
|
||||
}
|
||||
],
|
||||
"open_questions": ["跨指标、以及整条 SQL 还没把握的问题"]
|
||||
"open_questions": ["跨指标、以及整条 SQL 还没把握的问题"],
|
||||
"uncovered": ["这次没有起草的指标名称(不同粒度、不同表、或写不下的)"]
|
||||
}
|
||||
"""
|
||||
|
||||
@@ -184,6 +191,39 @@ def _fit(docs: list[dict], chunks: dict[str, dict], guidance: str,
|
||||
return kept, tables, truncated
|
||||
|
||||
|
||||
def _salvage_value(text: str, key: str) -> Any:
|
||||
"""从被截断的回复里抢回一个完整的顶层字段("sql": "…" 这样的值)。值本身不完整就返回 None。"""
|
||||
at = text.find(f'"{key}"')
|
||||
if at < 0:
|
||||
return None
|
||||
colon = text.find(":", at + len(key) + 2)
|
||||
if colon < 0:
|
||||
return None
|
||||
start = colon + 1
|
||||
while start < len(text) and text[start] in " \t\r\n":
|
||||
start += 1
|
||||
try:
|
||||
value, _ = json.JSONDecoder().raw_decode(text, start)
|
||||
except (json.JSONDecodeError, RecursionError):
|
||||
return None
|
||||
return value
|
||||
|
||||
|
||||
def _salvage_reply(text: str) -> dict:
|
||||
"""JSON 被截断时,把还能完整读出来的部分抢回来。
|
||||
|
||||
指标逐条抢(docdraft 按平衡括号找完整的指标对象),sql 和入口字段按完整的 JSON 值抢。抢不回来的留空,
|
||||
由调用方照实说「这次起草不完整」,而不是把半截回复当成没有指标。
|
||||
"""
|
||||
return {
|
||||
"metrics": docdraft.salvage_metrics(text),
|
||||
"sql": _salvage_value(text, "sql") or "",
|
||||
"entry_fields": _salvage_value(text, "entry_fields") or [],
|
||||
"open_questions": _salvage_value(text, "open_questions") or [],
|
||||
"uncovered": _salvage_value(text, "uncovered") or [],
|
||||
}
|
||||
|
||||
|
||||
async def plan(session: Any, *, docs: list[dict], chunks: dict[str, dict], guidance: str = "",
|
||||
datasource: dict | None = None, tables_text: str = "", model: Any = None) -> dict:
|
||||
"""一次模型调用,返回 {entry_fields, sql, metrics, open_questions, model_id, truncated}(未校验)。
|
||||
@@ -205,15 +245,36 @@ async def plan(session: Any, *, docs: list[dict], chunks: dict[str, dict], guida
|
||||
|
||||
kept, tables, truncated = _fit(docs, chunks, guidance, datasource, tables_text)
|
||||
reply = await model.ainvoke(prompt(docs, kept, guidance, datasource, tables))
|
||||
data = docdraft.parse_reply(message_text(reply))
|
||||
text = message_text(reply)
|
||||
salvaged = False
|
||||
try:
|
||||
data = docdraft.parse_reply(text)
|
||||
except ValueError:
|
||||
# 回复被截断(SQL 写长了、指标太多):能抢回多少算多少,别整份当成「没有指标」
|
||||
data, salvaged = _salvage_reply(text), True
|
||||
metrics = data.get("metrics")
|
||||
questions = data.get("open_questions")
|
||||
uncovered = data.get("uncovered")
|
||||
metrics_list = metrics if isinstance(metrics, list) else []
|
||||
diagnostics = ""
|
||||
if salvaged:
|
||||
diagnostics = f"回复被截断,从完整片段里抢救出 {len(metrics_list)} 条指标"
|
||||
if not metrics_list:
|
||||
keys = "、".join(str(k) for k in data) or "无"
|
||||
shape = {"dict": "对象", "list": "数组", "str": "字符串", "NoneType": "空"}.get(
|
||||
type(metrics).__name__, type(metrics).__name__)
|
||||
diagnostics = (f"模型回复里没有可读的指标数组(metrics 是{shape};顶层键:{keys})"
|
||||
+ (f";{diagnostics}" if diagnostics else ""))
|
||||
return {
|
||||
"entry_fields": data.get("entry_fields") if isinstance(data.get("entry_fields"), list) else [],
|
||||
"sql": data.get("sql") if isinstance(data.get("sql"), str) else "",
|
||||
"metrics": metrics if isinstance(metrics, list) else [],
|
||||
"metrics": metrics_list,
|
||||
"open_questions": [q for q in questions if isinstance(q, str) and q.strip()]
|
||||
if isinstance(questions, list) else [],
|
||||
"uncovered": [q.strip() for q in uncovered if isinstance(q, str) and q.strip()]
|
||||
if isinstance(uncovered, list) else [],
|
||||
"salvaged": salvaged,
|
||||
"diagnostics": diagnostics,
|
||||
"model_id": model_id,
|
||||
"truncated": truncated,
|
||||
}
|
||||
|
||||
@@ -198,3 +198,30 @@ async def test_plan_refuses_when_no_draft_model_is_available(monkeypatch):
|
||||
monkeypatch.setattr(recipe_ai, "resolve_draft_model", resolve)
|
||||
with pytest.raises(docdraft.DraftUnavailable, match="起草模型"):
|
||||
await draftplan.plan(None, docs=[], chunks={})
|
||||
|
||||
|
||||
async def test_plan_salvages_metrics_when_the_reply_is_cut_off():
|
||||
"""SQL 把输出额度吃光、JSON 在指标数组里断掉:抢回完整的指标和 SQL,并照实说「回复被截断」。"""
|
||||
text = ('{"entry_fields": [], "sql": "SELECT SUM(amount) AS gmv FROM orders", "metrics": ['
|
||||
+ json.dumps(raw()["metrics"][0], ensure_ascii=False) + ', {"id": "m2", "name": "写到这里就断了')
|
||||
result = await draftplan.plan(None, docs=[], chunks={}, model=FakeModel(text))
|
||||
assert result["salvaged"] is True
|
||||
assert len(result["metrics"]) == 1 and result["metrics"][0]["name"] == "销售额"
|
||||
assert result["sql"].startswith("SELECT SUM(amount)")
|
||||
assert "截断" in result["diagnostics"]
|
||||
|
||||
|
||||
async def test_plan_explains_what_came_back_when_metrics_is_not_a_list():
|
||||
"""指标为空时不能只说「没有一条通过校验」:要说出模型到底交回了什么形状。"""
|
||||
model = FakeModel(json.dumps({"sql": "SELECT 1", "metrics": {}}, ensure_ascii=False))
|
||||
result = await draftplan.plan(None, docs=[], chunks={}, model=model)
|
||||
assert result["metrics"] == []
|
||||
assert "metrics 是对象" in result["diagnostics"]
|
||||
assert "sql" in result["diagnostics"]
|
||||
|
||||
|
||||
async def test_plan_passes_uncovered_through():
|
||||
"""一次只起草能共用一条 SQL 的一组,其余的写进 uncovered,下一轮再起草。"""
|
||||
model = FakeModel(json.dumps({**raw(), "uncovered": ["月活买家"]}, ensure_ascii=False))
|
||||
result = await draftplan.plan(None, docs=[], chunks={}, model=model)
|
||||
assert result["uncovered"] == ["月活买家"]
|
||||
|
||||
Reference in New Issue
Block a user