把领域专家的知识编码进系统：Spotify 数据助手背后的上下文层

# context_layer.py — 最小上下文层定义

from dataclasses import dataclass, field
from datetime import date
from typing import Optional

@dataclass
class DataEntity:
    """一个数据实体，携带语义描述而非纯技术元数据"""
    id: str
    name: str
    description: str          # 面向业务的语义描述
    authoritative: bool       # 是否为该指标的权威来源
    owner_team: str
    known_caveats: list[str] = field(default_factory=list)  # 已知偏差或注意事项
    decisions: list["Decision"] = field(default_factory=list)

@dataclass
class Decision:
    """口径变更或架构决策的记录"""
    date: date
    summary: str
    reason: str
    changed_fields: list[str] = field(default_factory=list)

@dataclass
class Relationship:
    """实体间关系：依赖、引用、冲突等"""
    source_id: str
    target_id: str
    relation_type: str  # "depends_on" | "referenced_by" | "conflicts_with"
    note: Optional[str] = None


# ---- 填入领域专家的知识 ----

nordic_retention = DataEntity(
    id="nordic_daily_retention",
    name="北欧市场日级留存率",
    description="北欧五国用户的 30 天滚动留存率，按注册日分组",
    authoritative=True,
    owner_team="growth-analytics",
    known_caveats=[
        "2023-06-01 前的数据使用 7 天口径，不可直接比较",
        "芬兰用户量小，单日波动大，建议看 7 天均值",
    ],
    decisions=[
        Decision(
            date=date(2023, 6, 1),
            summary="留存口径从 7 天改为 30 天",
            reason="业务方需要更长窗口评估订阅转化",
            changed_fields=["retention_window"],
        ),
    ],
)

global_retention = DataEntity(
    id="global_daily_retention",
    name="全球日级留存率",
    description="全量用户的 30 天留存，不含北欧单独口径",
    authoritative=True,
    owner_team="core-analytics",
    known_caveats=["与 nordic_daily_retention 口径不同，不可直接相减"],
)

entities = {e.id: e for e in [nordic_retention, global_retention]}

relationships = [
    Relationship(
        source_id="nordic_daily_retention",
        target_id="global_daily_retention",
        relation_type="conflicts_with",
        note="口径差异：北欧含订阅转化延迟，全球不含",
    ),
]

# assemble_context.py — 把问题相关的上下文拼成 prompt 片段

def find_relevant_entities(question: str, entities: dict) -> list[DataEntity]:
    """简易关键词匹配；生产环境可用向量检索替代"""
    keywords = {
        "北欧": ["nordic_daily_retention"],
        "留存": ["nordic_daily_retention", "global_daily_retention"],
        "全球": ["global_daily_retention"],
    }
    matched_ids = set()
    for kw, ids in keywords.items():
        if kw in question:
            matched_ids.update(ids)
    return [entities[id] for id in matched_ids if id in entities]


def find_related(entity_id: str, relationships: list[Relationship]) -> list[Relationship]:
    return [r for r in relationships if r.source_id == entity_id or r.target_id == entity_id]


def build_context_block(question: str, entities: dict, relationships: list[Relationship]) -> str:
    """组装一段结构化上下文，插入 LLM prompt"""
    relevant = find_relevant_entities(question, entities)
    if not relevant:
        return "未找到相关数据实体，请仅基于通用知识回答，并声明不确定性。"

    lines = ["[上下文层 — 领域专家编码信息]\n"]
    for entity in relevant:
        lines.append(f"实体: {entity.name} (id={entity.id})")
        lines.append(f"描述: {entity.description}")
        lines.append(f"权威来源: {'是' if entity.authoritative else '否'}")
        lines.append(f"负责团队: {entity.owner_team}")
        if entity.known_caveats:
            lines.append("已知注意事项:")
            for c in entity.known_caveats:
                lines.append(f"  - {c}")
        if entity.decisions:
            lines.append("决策日志:")
            for d in entity.decisions:
                lines.append(f"  - [{d.date}] {d.summary} (原因: {d.reason})")

        rels = find_related(entity.id, relationships)
        if rels:
            lines.append("关联关系:")
            for r in rels:
                lines.append(f"  - {r.relation_type}: {r.target_id if r.source_id == entity.id else r.source_id} — {r.note}")

    lines.append("\n[指令] 基于以上上下文回答问题。如果上下文中有注意事项或口径差异，必须在回答中明确提及。")
    return "\n".join(lines)


# ---- 测试 ----

question = "上周北欧留存率为什么突然跌了？"
context = build_context_block(question, entities, relationships)
print(context)

[上下文层 — 领域专家编码信息]

实体: 北欧市场日级留存率 (id=nordic_daily_retention)
描述: 北欧五国用户的 30 天滚动留存率，按注册日分组
权威来源: 是
负责团队: growth-analytics
已知注意事项:
  - 2023-06-01 前的数据使用 7 天口径，不可直接比较
  - 芬兰用户量小，单日波动大，建议看 7 天均值
决策日志:
  - [2023-06-01] 留存口径从 7 天改为 30 天 (原因: 业务方需要更长窗口评估订阅转化)
关联关系:
  - conflicts_with: global_daily_retention — 口径差异：北欧含订阅转化延迟，全球不含

[指令] 基于以上上下文回答问题。如果上下文中有注意事项或口径差异，必须在回答中明确提及。

# call_llm.py — 把上下文层拼入实际 LLM 调用

import os
from openai import OpenAI

client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

system_prompt = """你是 Spotify 内部数据助手。
回答数据相关问题时，必须优先参考上下文层提供的领域专家信息。
如果上下文层没有覆盖到的问题，明确说明你的回答是推测而非确认。"""

question = "上周北欧留存率为什么突然跌了？"
context_block = build_context_block(question, entities, relationships)

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": f"{context_block}\n\n用户问题: {question}"},
    ],
    temperature=0.2,
)

print(response.choices[0].message.content)