mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:11:59 +00:00
口径详情给模型看定义说明
文档起草过来的口径,定义是它唯一带着的业务边界(适用范围、不包含什么)。此前 look 只给 指标式和单位,模型只能凭名字猜这份口径答的是不是这个问题。定义压成一句、限 240 字, 概览一行不加。
This commit is contained in:
@@ -33,6 +33,11 @@ OVERVIEW_LINE_PROMPT = "- {workflow} v{version}:{metrics}"
|
||||
OVERVIEW_MORE_PROMPT = "- …另有 {n} 个工作流"
|
||||
DETAIL_HEAD_PROMPT = "· {name}(指标 id {metric}{unit}):出自「{workflow}」v{version} 的口径卡「{card}」({caliber})"
|
||||
DETAIL_EXPR_PROMPT = " 指标式:{expression}"
|
||||
#: 定义说明只在详情里给:它是这张口径的业务边界(适用范围、不包含什么),选口径时要看的就是这一句。
|
||||
#: 概览一行放不下,也不该让它挤掉指标名
|
||||
DETAIL_DEFINITION_PROMPT = " 定义:{text}"
|
||||
#: 定义说明写进查看结果的上限:口述给人看的边界通常一两句,超出的部分在口径账本里看全文
|
||||
DEFINITION_MAX = 240
|
||||
DETAIL_REUSE_PROMPT = (" 复用写法(「看哪些数」这一项的 reuse 照抄,不要改):{reuse}\n"
|
||||
" 这张卡读:{inputs}。实例化时那个库的取数节点就叫这个名字、交出这几列,复用才算得出来")
|
||||
DETAIL_INPUT_PROMPT = "{node}({tool})的列 {columns}"
|
||||
@@ -77,6 +82,8 @@ class CaliberView:
|
||||
expression: str
|
||||
inputs: tuple[tuple[str, str, tuple[str, ...]], ...]
|
||||
reason: str | None
|
||||
#: 指标卡上的定义说明(说明、适用范围、不包含什么),压成一句;没有定义时为空串
|
||||
definition: str = ""
|
||||
|
||||
@property
|
||||
def label(self) -> str:
|
||||
@@ -159,6 +166,29 @@ def _metric_paths(metric: Any) -> list[str] | None:
|
||||
return paths
|
||||
|
||||
|
||||
def definition_text(definition: Any) -> str:
|
||||
"""指标卡上的定义说明 → 一句给模型看的文字:说明、适用范围、不包含什么。
|
||||
|
||||
只取这三项:负责人和生效日期是治理信息,模型选口径时用不上,写进去只会挤占额度。
|
||||
"""
|
||||
if not isinstance(definition, dict):
|
||||
return ""
|
||||
parts: list[str] = []
|
||||
text = definition.get("text")
|
||||
if isinstance(text, str) and text.strip():
|
||||
parts.append(text.strip())
|
||||
scope = definition.get("scope")
|
||||
if isinstance(scope, str) and scope.strip():
|
||||
parts.append(f"适用范围:{scope.strip()}")
|
||||
exclusions = definition.get("exclusions")
|
||||
if isinstance(exclusions, list):
|
||||
kept = [str(x).strip() for x in exclusions if str(x).strip()]
|
||||
if kept:
|
||||
parts.append("不包含:" + "、".join(kept))
|
||||
merged = ";".join(parts)
|
||||
return merged[:DEFINITION_MAX] + ("…" if len(merged) > DEFINITION_MAX else "")
|
||||
|
||||
|
||||
def views_of(workflow_id: str, workflow_name: str, version: Any) -> list[CaliberView]:
|
||||
"""一个已发布版本里每张口径卡(源头的)的每个指标。"""
|
||||
from app.apps.tree import node_map
|
||||
@@ -179,7 +209,7 @@ def views_of(workflow_id: str, workflow_name: str, version: Any) -> list[Caliber
|
||||
level=str(getattr(version, "level", None) or ""), node_id=nid, card=node["title"], caliber=caliber,
|
||||
metric_id=str(metric["id"]), name=str(metric.get("name") or metric["id"]),
|
||||
unit=str(metric.get("unit") or ""), expression=str(metric.get("expression") or ""), inputs=packed,
|
||||
reason=reason))
|
||||
reason=reason, definition=definition_text(metric.get("definition"))))
|
||||
return out
|
||||
|
||||
|
||||
@@ -314,6 +344,8 @@ class CaliberKind:
|
||||
workflow=item.workflow_name, version=item.version, card=item.card,
|
||||
caliber=item.caliber),
|
||||
DETAIL_EXPR_PROMPT.format(expression=item.expression)]
|
||||
if item.definition:
|
||||
block.append(DETAIL_DEFINITION_PROMPT.format(text=item.definition))
|
||||
option = self.option_of(item)
|
||||
if option is not None:
|
||||
block += _compose_lines(item, option, seen)
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
"""已发布口径的定义说明要能被选口径的模型看到(app/look/calibers.definition_text 与 detail)。
|
||||
|
||||
定义是文档起草那条路唯一带过来的东西:模型看不到它,就只能凭指标名猜这份口径答的是什么问题,
|
||||
文档里写的「不包含退款」也就白写了。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from app.look.calibers import DEFINITION_MAX, CaliberKind, CaliberView, definition_text
|
||||
|
||||
|
||||
def _view(definition: str = "", **over: object) -> CaliberView:
|
||||
base = dict(workflow_id="w1", workflow_name="周销售", version=1, level="governed", node_id="card",
|
||||
card="口径卡", caliber="周销售 @ v1", metric_id="gmv", name="周销售额", unit="元",
|
||||
expression="cell(nodes.fetch, 0, 'gmv')",
|
||||
inputs=(("fetch", "db_query__shop", ("gmv",)),), reason=None)
|
||||
base.update(over)
|
||||
return CaliberView(definition=definition, **base) # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_definition_text_keeps_three_parts() -> None:
|
||||
text = definition_text({"text": "本周已支付订单金额合计", "scope": "全渠道", "exclusions": ["退款", "测试账号"],
|
||||
"owner": "财务部", "effective_from": "2026-01-01"})
|
||||
assert text == "本周已支付订单金额合计;适用范围:全渠道;不包含:退款、测试账号"
|
||||
# 负责人和生效日期是治理信息,不占模型的额度
|
||||
assert "财务部" not in text and "2026-01-01" not in text
|
||||
|
||||
|
||||
def test_definition_text_is_capped_and_empty_when_missing() -> None:
|
||||
long = definition_text({"text": "长" * (DEFINITION_MAX + 50)})
|
||||
assert len(long) == DEFINITION_MAX + 1 and long.endswith("…")
|
||||
assert definition_text(None) == ""
|
||||
assert definition_text({}) == ""
|
||||
assert definition_text({"owner": "财务部"}) == ""
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_detail_shows_definition_only_when_present() -> None:
|
||||
with_def = _view("本周已支付订单金额合计;不包含:退款")
|
||||
plain = _view()
|
||||
text, record = await CaliberKind([with_def, plain]).detail([(with_def, "named")], budget=10_000)
|
||||
assert "定义:本周已支付订单金额合计;不包含:退款" in text
|
||||
assert record[0]["kind"] == "caliber"
|
||||
|
||||
text, _ = await CaliberKind([plain]).detail([(plain, "named")], budget=10_000)
|
||||
assert "定义:" not in text
|
||||
Reference in New Issue
Block a user