入口只留每次都会变的东西，固定条件写死在 SQL 里

实测同一份文档：入口从 7 个（start/end/factory_code/object_id/line_code/gc_code/scx_code）
降到 2 个（start/end）。

提示词加了四条：同一维度只做一个入口（factory_code 和 gc_code 是同一件事）、编码和主键不当
必填入口、跨表编码在 SQL 里 JOIN 映射、可选维度写成空值不过滤；入口上限 4 个，超了照实报
出来但不悄悄删（删了 SQL 还引用着）。起草页多一栏「固定条件」：工厂、产线、属性组填这里，
写死在 SQL 里，不变成每次都要填的参数。

backend/app/api/ledger.py

@@ -813,6 +813,7 @@ async def _extract_in_background(draft_id: str) -> None:
         datasource = {"name": source.name, "type": source.kind, "description": source.description}
 
         guidance = str(origin.get("guidance") or "")
+        fixed_filters = str(origin.get("fixed_filters") or "")
         alignment: dict[str, list[Any]] = {"new": [], "duplicate": [], "conflict": []}
         # 每组一组指标：{sql, entry_fields, metrics}，循环结束时拼成一张图
         blocks: list[dict[str, Any]] = []
@@ -835,7 +836,8 @@ async def _extract_in_background(draft_id: str) -> None:
         try:
             for rnd in range(1, MAX_ROUNDS + 1):
                 result = await draftplan.plan(session, docs=entries, chunks=chunks, guidance=guidance,
-                                              datasource=datasource, tables_text=tables_text, covered=covered)
+                                              datasource=datasource, tables_text=tables_text, covered=covered,
+                                              fixed=fixed_filters)
                 plan, round_problems = draftplan.normalize_plan(result, chunks, allowed_tables=allowed_tables)
                 fresh = _fresh_metrics(plan["metrics"], covered)
                 if plan["sql"] and fresh:
@@ -951,12 +953,17 @@ def _problems_text(problems: list[dict[str, Any]]) -> str:
 
 
 class DraftExtractIn(BaseModel):
-    """从哪几份文档起草、起草时给模型看的附加要求。数据源必须给：它的表结构和目录是 SQL 的唯一依据。"""
+    """从哪几份文档起草、起草时给模型看的附加要求。数据源必须给：它的表结构和目录是 SQL 的唯一依据。
+
+    fixed_filters 是「每次跑都不变的条件」（工厂、产线、属性组）：写死在 SQL 里，不做成入口字段。
+    入口只留「每次跑都可能不同」的东西（时间范围，最多再一个业务维度），否则每问一次数都要填一遍编码。
+    """
 
     documents: list[str] = Field(min_length=1, max_length=20)
     collection: str | None = Field(default=None, max_length=100)
     datasource_id: str | None = Field(default=None, max_length=32)
     guidance: str = Field(default="", max_length=2000)
+    fixed_filters: str = Field(default="", max_length=1000)
 
 
 @router.post("/drafts/extract", status_code=201)
@@ -989,7 +996,8 @@ async def draft_extract(payload: DraftExtractIn, background: BackgroundTasks,
     draft = CaliberDraft(title=_doc_title(docs),
                          collection=payload.collection or (docs[0].collection or "default"),
                          datasource_id=payload.datasource_id or None, status="drafting", payload={},
-                         origin={"documents": doc_ids, "guidance": payload.guidance.strip()},
+                         origin={"documents": doc_ids, "guidance": payload.guidance.strip(),
+                                 "fixed_filters": payload.fixed_filters.strip()},
                          created_by=current_principal().username)
     session.add(draft)
     await session.commit()

backend/app/ledger/draftplan.py

@@ -46,7 +46,10 @@ ENTRY_TYPES = ("string", "number", "integer", "boolean")
 #: 指标里固定出现的字段（顺序照接口契约）
 METRIC_FIELDS = ("id", "name", "synonyms", "unit", "decimals", "expression", "definition", "open_questions")
 
-#: SQL 模板里的入口占位符：{{ input.x }}（引号内外都认）
+#: 入口字段最多几个（时间范围占两个）。再多说明把「每次都不变」的条件也做成了参数，
+#: 人每跑一次都要填一遍，口径就没法用了
+MAX_ENTRY_FIELDS = 4
+#: 输入里能出现的入口占位符：{{ input.x }}（引号内外都认）
 _INPUT = re.compile(r"\{\{\s*input\.([A-Za-z_][\w-]*)\s*\}\}")
 #: 口径卡的取值表达式，列名可以用单引号或双引号
 _EXPRESSION = re.compile(r"^cell\(\s*nodes\.fetch\s*,\s*0\s*,\s*['\"]([^'\"]+)['\"]\s*\)$")
@@ -73,6 +76,15 @@ _SYSTEM = """你是业务口径与取数模板的起草助手。读给定的业
 8. **sql 要短。** 能用 GROUP BY 和聚合表达的，不要为每个属性枚举一长串 CASE WHEN 列；输出额度有限，SQL 写长了，
    后面的指标就写不出来了。
 9. 写不下时宁可少写几条指标、把剩下的写进 uncovered，**也不要让 JSON 半截断掉**。
+10. **同一件事只做一个入口。** 工厂、产线这类维度全图只给一个字段，不要为不同的表各造一个
+    （factory_code / gc_code / scx_code 说的是同一件事）；跨表的编码差异在 sql 里用 JOIN 或子查询映射，
+    不要让填参数的人重复填等价编码。
+11. **编码和主键不当必填入口。** object_id、line_code 这类字段只有业务上确实每次按它跑时才做入口，
+    而且必须是可选（留空不过滤）；人填不出来才能跑的入口，等于这份口径跑不了。
+12. **可选维度要写成「空值不过滤」**：`AND ('{{ input.line_code }}' = '' OR t.line_code = '{{ input.line_code }}')`。
+    不这么写，留空会把结果筛成空。
+13. **入口最多 4 个**，时间范围占两个。再多就把不常变的条件写死在 sql 里（见第 14 条）。
+14. **不常变的条件写进 sql，不做入口**：固定筛选段里给出的取值（工厂、产线、属性组、状态）直接写成常量。
 
 输出格式：
 {
@@ -135,7 +147,7 @@ def _chunk_block(chunk_id: str, chunk: Any) -> str:
 
 
 def _human(docs: list[dict], chunks: Mapping[str, dict], guidance: str, datasource: dict | None,
-           tables_text: str, covered: Sequence[str] = ()) -> str:
+           tables_text: str, covered: Sequence[str] = (), fixed: str = "") -> str:
     parts = ["文档清单：", *(_doc_line(doc) for doc in docs), "",
              "逐段原文（引用时 chunk 写方括号里的 id，quote 逐字抄）："]
     parts.extend(_chunk_block(chunk_id, chunks[chunk_id]) for chunk_id in _order(chunks))
@@ -151,17 +163,20 @@ def _human(docs: list[dict], chunks: Mapping[str, dict], guidance: str, datasour
     names = [name.strip() for name in covered if isinstance(name, str) and name.strip()]
     parts.append("、".join(names) if names else "（无）")
     parts.append("")
+    parts.append("固定条件（这些写死在 sql 里，**不要**做成入口字段）：{}".format(
+        fixed.strip() if fixed and fixed.strip() else "（无）"))
+    parts.append("")
     parts.append("用户补充要求：{}".format(guidance.strip() if guidance and guidance.strip() else "（无）"))
     return "\n".join(parts)
 
 
 def prompt(docs: list[dict], chunks: dict[str, dict], guidance: str = "", datasource: dict | None = None,
-           tables_text: str = "", covered: Sequence[str] = ()) -> list[Any]:
-    """构造 system + human：文档原文 + 数据目录 + 已经起草过的指标。目录是 sql 的唯一依据，和原文一样原样交给模型。"""
+           tables_text: str = "", covered: Sequence[str] = (), fixed: str = "") -> list[Any]:
+    """构造 system + human：文档原文 + 数据目录 + 已经起草过的指标 + 固定条件。目录是 sql 的唯一依据。"""
     from langchain_core.messages import HumanMessage, SystemMessage
 
     return [SystemMessage(content=_SYSTEM),
-            HumanMessage(content=_human(docs, chunks, guidance, datasource, tables_text, covered))]
+            HumanMessage(content=_human(docs, chunks, guidance, datasource, tables_text, covered, fixed))]
 
 
 def _chars(messages: list[Any]) -> int:
@@ -170,7 +185,7 @@ def _chars(messages: list[Any]) -> int:
 
 def _fit(docs: list[dict], chunks: dict[str, dict], guidance: str,
          datasource: dict | None, tables_text: str,
-         covered: Sequence[str] = ()) -> tuple[dict[str, dict], str, bool, bool]:
+         covered: Sequence[str] = (), fixed: str = "") -> tuple[dict[str, dict], str, bool, bool]:
     """把材料压到 MAX_INPUT_CHARS 以内：先截表目录尾部，再从后往前按 chunk 边界丢，最后截仅剩的那一段。
 
     表目录先截：SQL 必须写在目录里的表上，目录比文档的细枝末节更硬；被截掉的是表详情，表名索引还在最前面。
@@ -187,7 +202,7 @@ def _fit(docs: list[dict], chunks: dict[str, dict], guidance: str,
         catalog_truncated = True
 
     def size() -> int:
-        return _chars(prompt(docs, kept, guidance, datasource, tables, covered))
+        return _chars(prompt(docs, kept, guidance, datasource, tables, covered, fixed))
 
     order = _order(kept)
     while size() > MAX_INPUT_CHARS and len(order) > 1:
@@ -243,11 +258,12 @@ def _salvage_reply(text: str) -> dict:
 
 async def plan(session: Any, *, docs: list[dict], chunks: dict[str, dict], guidance: str = "",
                datasource: dict | None = None, tables_text: str = "", model: Any = None,
-               covered: Sequence[str] = ()) -> dict:
+               covered: Sequence[str] = (), fixed: str = "") -> dict:
     """一次模型调用，返回 {entry_fields, sql, metrics, open_questions, model_id, doc_truncated,
     catalog_truncated, truncated}（未校验）。
 
-    covered 是前几轮已经起草过的指标名，写进提示词让模型不要重复；调用方按 uncovered 续跑，直到 covered 不再增长。
+    covered 是前几轮已经起草过的指标名，写进提示词让模型不要重复；fixed 是「每次都不变的条件」（工厂、产线、
+    属性组），提示词要求它写死在 sql 里、不做成入口字段。调用方按 uncovered 续跑，直到 covered 不再增长。
     truncated 仍是两者的「或」，老前端只认这一个键。不落库。model 可注入（测试用假对象：有 async ainvoke(messages)，
     回复有 .content）；不给就按设置解析，模型不可用时抛 docdraft.DraftUnavailable。SQL 的对错、引用能不能核对，交给
     normalize_plan。
@@ -265,8 +281,8 @@ async def plan(session: Any, *, docs: list[dict], chunks: dict[str, dict], guida
         model_id = named if isinstance(named, str) else ""
 
     kept, tables, doc_truncated, catalog_truncated = _fit(docs, chunks, guidance, datasource, tables_text,
-                                                          covered)
-    reply = await model.ainvoke(prompt(docs, kept, guidance, datasource, tables, covered))
+                                                          covered, fixed)
+    reply = await model.ainvoke(prompt(docs, kept, guidance, datasource, tables, covered, fixed))
     text = message_text(reply)
     salvaged = False
     try:
@@ -435,6 +451,11 @@ def normalize_plan(raw: Any, chunks: Mapping[str, dict], *, allowed_tables: set[
     body = raw if isinstance(raw, Mapping) else {}
     problems: list[dict] = []
     entry_fields = _entry_fields(body.get("entry_fields"), problems)
+    if len(entry_fields) > MAX_ENTRY_FIELDS:
+        # 不自动删：入口删了 SQL 还引用着，跑不通。照实说清楚，让人把不常变的条件写成固定条件重起一份
+        problems.append({"name": "入口字段",
+                         "reason": f"入口有 {len(entry_fields)} 个（上限 {MAX_ENTRY_FIELDS}）：同一维度只该有一个，"
+                                   f"不常变的条件应当写死在 sql 里（用「固定条件」重起一份）"})
 
     sql = body.get("sql")
     sql_text = sql.strip() if isinstance(sql, str) else ""

backend/tests/test_ledger_docdraft_api.py

@@ -736,3 +736,25 @@ async def test_running_the_trial_with_inputs_typed_on_the_page(client, documents
     res = await client.post(f"/api/ledger/drafts/{body['draft_id']}/trial", json={"inputs": {"start": "2026-01-01"}})
     assert res.status_code == 200, res.text
     assert seen["start"] == "2026-01-01"
+
+
+async def test_extract_hands_the_fixed_filters_to_every_round(client, documents, datasource, monkeypatch):
+    """固定条件（工厂、产线、属性组）要交给模型写死在 SQL 里，不做成入口字段。"""
+    from app.ledger import draftplan
+
+    doc = await documents("dd-fixed")
+    source_id = await datasource()
+    fake_model(monkeypatch, plan_reply(f"{doc}-2"))
+    seen: dict[str, object] = {}
+    real_plan = draftplan.plan
+
+    async def spy(session, **kwargs):
+        seen.update(kwargs)
+        return await real_plan(session, **kwargs)
+
+    monkeypatch.setattr(draftplan, "plan", spy)
+    body = await client.post("/api/ledger/drafts/extract", json={
+        "documents": [doc], "datasource_id": source_id, "fixed_filters": "line_code = 'FOAM01'"})
+    made = (await client.get(f"/api/ledger/drafts/{body.json()['draft_id']}")).json()
+    assert made["workflow_id"]
+    assert seen["fixed"] == "line_code = 'FOAM01'"

backend/tests/test_ledger_draftplan.py

@@ -346,3 +346,21 @@ async def test_plan_reports_document_and_catalog_truncation_separately():
     assert result["doc_truncated"] is False
     assert result["catalog_truncated"] is False
     assert result["truncated"] is False
+
+
+async def test_plan_tells_the_model_which_conditions_are_fixed():
+    """固定条件要原样交到模型手里，并明确说「写死在 SQL 里、不要做成入口」。"""
+    model = FakeModel(json.dumps(raw(), ensure_ascii=False))
+    await draftplan.plan(None, docs=[], chunks={}, model=model, fixed="line_code = 'FOAM01'")
+    human = model.messages[-1].content
+    assert "固定条件" in human and "FOAM01" in human
+    assert "不要" in human and "入口" in human
+
+
+def test_too_many_entry_fields_is_reported_not_silently_trimmed():
+    """入口超过 4 个说明把不常变的条件也做成了参数：照实报出来，但不悄悄删（删了 SQL 还引用着）。"""
+    fields = [{"name": "start", "type": "string", "required": True, "default": "2026-09-01"}]
+    fields += [{"name": f"dim{i}", "type": "string", "required": False} for i in range(4)]
+    plan, problems = draftplan.normalize_plan(raw(entry_fields=fields), CHUNKS, allowed_tables={"orders"})
+    assert len(plan["entry_fields"]) == 5
+    assert any(p["name"] == "入口字段" and "上限" in p["reason"] for p in problems)

frontend/src/api/caliberLedger.ts

@@ -64,8 +64,10 @@ export const caliberLedger = {
    */
   runDraftTrial: (draftId: string, inputs: Record<string, unknown>) =>
     post<DraftTrialOut>(`${draftAt(draftId)}/trial`, { inputs }),
-  /** 选文档发起起草：documents 是要用的知识库文档 id，guidance 是给模型看的附加要求 */
-  extractDraft: (body: { documents: string[]; collection?: string; datasource_id?: string; guidance?: string }) =>
+  /** 选文档发起起草：documents 是要用的知识库文档 id，guidance 是给模型看的附加要求，
+   *  fixed_filters 是每次跑都不变的条件（工厂、产线、属性组），服务端要求写死在 SQL 里、不做成入口字段 */
+  extractDraft: (body: { documents: string[]; collection?: string; datasource_id?: string; guidance?: string;
+    fixed_filters?: string }) =>
     post<{ draft_id: string; status: string }>('/ledger/drafts/extract', body),
   /**
    * 改标题、整体换掉指标、或把答清的问题消掉（resolved 是问题原文）。服务端据此重算状态：还有问题 → questions_open，

frontend/src/lib/termsLedger.ts

@@ -269,6 +269,10 @@ export const LEDGER_UI = {
     datasourceHint: '取数 SQL 对照这个库的真实表和列写，先探查过表结构才能起草',
     guidance: '起草要求（可选）',
     guidancePlaceholder: '例如：只整理和月度结算有关的指标',
+    fixedFilters: '固定条件（可选）',
+    fixedFiltersHint: '每次运行都不变的条件写死在 SQL 里，不变成入口字段：工厂、产线、属性组这类编码都放这里。'
+      + '入口只留时间范围，最多再一个业务维度',
+    fixedFiltersPlaceholder: "例如：line_code = 'FOAM01'、attribute_group = 'OEE'",
     submit: '开始起草',
     submitting: '正在发起…',
     needDoc: '请至少选一份文档',

frontend/src/pages/ledger/CaliberDrafts.tsx

@@ -188,6 +188,8 @@ export function CaliberDrafts() {
   const [pickedDocs, setPickedDocs] = useState<string[]>([])
   const [datasourceId, setDatasourceId] = useState('')
   const [guidance, setGuidance] = useState('')
+  /** 每次跑都不变的条件（工厂、产线、属性组）：要求写死在 SQL 里，不做成入口字段 */
+  const [fixedFilters, setFixedFilters] = useState('')
   const [submitting, setSubmitting] = useState(false)
 
   const [pickedQuestions, setPickedQuestions] = useState<string[]>([])
@@ -404,10 +406,12 @@ export function CaliberDrafts() {
         documents: pickedDocs,
         ...(datasourceId ? { datasource_id: datasourceId } : {}),
         ...(guidance.trim() ? { guidance: guidance.trim() } : {}),
+        ...(fixedFilters.trim() ? { fixed_filters: fixedFilters.trim() } : {}),
       })
       toast.ok(D.started)
       setPickedDocs([])
       setGuidance('')
+      setFixedFilters('')
       setDatasourceId('')
       setFilter('')
       loadList()
@@ -978,6 +982,13 @@ export function CaliberDrafts() {
                   <textarea className="field min-h-[64px] w-full text-xs" rows={3} value={guidance}
                             placeholder={D.guidancePlaceholder} onChange={(e) => setGuidance(e.target.value)} />
                 </label>
+                <label className="block text-2xs text-faint">
+                  <span className="mb-0.5 block font-semibold text-dim">{D.fixedFilters}</span>
+                  <textarea className="field min-h-[52px] w-full text-xs" rows={2} value={fixedFilters}
+                            placeholder={D.fixedFiltersPlaceholder} data-draft-fixed-filters=""
+                            onChange={(e) => setFixedFilters(e.target.value)} />
+                  <span className="mt-0.5 block">{D.fixedFiltersHint}</span>
+                </label>
                 <button type="button" className="btn btn-primary btn-sm" disabled={submitting || !datasourceId}
                         onClick={() => void submit()} data-draft-submit="">
                   {submitting ? <Spinner size={12} /> : <FileText size={12} aria-hidden />} {submitting ? D.submitting : D.submit}