┌──────────────────────────────────┐
│  1. 角色定义 (Role)              │
│  2. 任务描述 (Task)              │
│  3. 行为约束 (Constraints)       │
│  4. 输出格式 (Output Format)     │
│  5. 知识边界 (Knowledge Boundary)│
│  6. 示例 (Examples)              │
│  7. 异常处理 (Error Handling)    │
└──────────────────────────────────┘

# 角色
你是一个高级数据分析师，擅长从结构化数据中提取洞察。

# 任务
用户会提供数据描述或 CSV 数据，你需要：
1. 理解数据结构和业务含义
2. 进行统计分析
3. 生成可视化建议
4. 输出分析报告

# 约束
- 只基于用户提供的数据分析，不编造数据
- 如果数据不足，明确说明需要什么数据
- 不输出投资建议或确定性预测
- 统计方法需标注假设条件

# 输出格式
```json
{
  "summary": "一句话摘要",
  "statistics": {"mean": 0, "median": 0, "std": 0},
  "insights": ["洞察1", "洞察2"],
  "visualization": {"type": "bar", "reason": "原因"},
  "limitations": ["局限1"]
}

## 2. 角色定义的科学方法

### 2.1 角色三要素

```python
ROLE_TEMPLATE = """
# 身份
你是{profession}，拥有{experience}年经验。

# 专长
{expertise_list}

# 思维方式
{thinking_style}

# 沟通风格
{communication_style}
"""

# 示例
role = ROLE_TEMPLATE.format(
    profession="资深安全工程师",
    experience="15",
    expertise_list="- Web 安全\n- 逆向工程\n- 安全架构设计",
    thinking_style="先评估威胁模型，再设计方案，最后验证有效性",
    communication_style="简洁直接，用数据和事实说话，避免模糊表述"
)

# ❌ 过度设定
你是一个名叫Alice的AI助手，今年28岁，毕业于MIT计算机系，
喜欢喝拿铁，有一只叫Tensor的猫，性格内向但工作热情，
暗恋同事Bob但不敢表白...

# 这些信息对任务毫无帮助，浪费 Token 且分散注意力

CONSTRAINTS = """
# 硬约束（必须遵守）
- 只使用中文回答
- 输出必须是合法 JSON
- 单次回答不超过 500 字
- 不讨论政治、宗教敏感话题

# 软约束（尽量遵守）
- 优先使用专业术语
- 引用来源时标注可信度
- 如果不确定，明确说明

# 条件约束（满足条件时触发）
- 如果用户问代码问题 → 必须提供可运行代码
- 如果用户提供数据 → 必须先描述数据特征
- 如果回答涉及医学/法律 → 必须加免责声明
"""

# 方法1：JSON Schema 描述
OUTPUT_FORMAT = """
输出必须符合以下结构：
{
  "classification": "string (one of: A, B, C)",
  "confidence": "number (0-1)",
  "reasoning": "string (max 200 chars)",
  "evidence": ["string", "string"]
}
"""

# 方法2：XML 标签
OUTPUT_FORMAT = """
<result>
  <classification>分类结果</classification>
  <confidence>置信度</confidence>
  <reasoning>推理过程</reasoning>
</result>
"""

# 方法3：Markdown 模板
OUTPUT_FORMAT = """
## 分类结果
{{classification}}

## 置信度
{{confidence}}/10

## 推理过程
{{reasoning}}
"""

class ConversationManager:
    def __init__(self, system_prompt):
        self.system_prompt = system_prompt
        self.history = []
        self.state = {
            "topic": None,
            "user_intent": None,
            "facts_collected": {},
            "turn_count": 0
        }
    
    def update_state(self, user_input, assistant_output):
        """每轮对话后更新状态"""
        self.state["turn_count"] += 1
        
        # 话题检测
        new_topic = self.detect_topic(user_input)
        if new_topic != self.state["topic"]:
            self.state["topic"] = new_topic
            self.state["facts_collected"] = {}  # 话题切换，重置事实
    
    def build_prompt(self, user_input):
        """构造带状态追踪的 Prompt"""
        context = f"""
当前对话状态：
- 话题：{self.state['topic']}
- 已收集信息：{json.dumps(self.state['facts_collected'], ensure_ascii=False)}
- 对话轮次：{self.state['turn_count']}

用户输入：{user_input}
"""
        return context

def manage_context(messages, max_tokens=6000):
    """上下文窗口管理"""
    total = count_tokens(messages)
    
    if total <= max_tokens:
        return messages
    
    # 保留 system prompt + 最近 N 轮
    system = messages[0]
    recent = messages[-6:]  # 最近3轮
    
    # 中间部分摘要
    middle = messages[1:-6]
    summary = summarize_conversation(middle)
    
    return [system, 
            {"role": "system", "content": f"对话摘要：{summary}"},
            *recent]

# prompts/v2_analyst.yaml
version: "2.1.0"
parent: "2.0.0"
changelog:
  - "增加了 JSON 输出格式约束"
  - "收紧了长度限制到 500 字"
  - "增加了条件约束：数据为空时的处理"

system_prompt: |
  # 角色
  你是高级数据分析师...
  
  # 约束
  - 输出必须是合法 JSON
  ...

metadata:
  author: "prompt-engineering-team"
  created: "2026-06-25"
  tags: ["analyst", "json", "v2"]
  metrics:
    accuracy: 0.91
    latency_p95: 2.3s
    user_satisfaction: 4.2

import yaml
from pathlib import Path

class PromptRegistry:
    """Prompt 版本注册中心"""
    
    def __init__(self, registry_dir="prompts/"):
        self.dir = Path(registry_dir)
    
    def load(self, name, version="latest"):
        if version == "latest":
            versions = sorted(self.dir.glob(f"{name}_v*.yaml"))
            if not versions:
                raise FileNotFoundError(f"No prompts found for {name}")
            path = versions[-1]
        else:
            path = self.dir / f"{name}_v{version}.yaml"
        
        with open(path) as f:
            return yaml.safe_load(f)
    
    def diff(self, name, v1, v2):
        """比较两个版本的差异"""
        p1 = self.load(name, v1)["system_prompt"]
        p2 = self.load(name, v2)["system_prompt"]
        import difflib
        return list(difflib.unified_diff(
            p1.splitlines(), p2.splitlines(),
            lineterm="", n=2
        ))

import random
from dataclasses import dataclass
from typing import List

@dataclass
class PromptVariant:
    name: str
    system_prompt: str
    weight: float  # 流量分配比例

@dataclass
class TestResult:
    variant: str
    quality_score: float  # 1-5 评分
    latency_ms: float
    token_count: int
    user_feedback: str

class PromptABTest:
    def __init__(self, variants: List[PromptVariant]):
        # 归一化权重
        total = sum(v.weight for v in variants)
        self.variants = [
            PromptVariant(v.name, v.system_prompt, v.weight / total)
            for v in variants
        ]
        self.results: List[TestResult] = []
    
    def assign(self) -> PromptVariant:
        """流量分配"""
        r = random.random()
        cumulative = 0
        for v in self.variants:
            cumulative += v.weight
            if r <= cumulative:
                return v
        return self.variants[-1]
    
    def record(self, result: TestResult):
        self.results.append(result)
    
    def report(self):
        """生成 A/B 测试报告"""
        from statistics import mean, stdev
        report = {}
        for v in self.variants:
            v_results = [r for r in self.results if r.variant == v.name]
            if not v_results:
                continue
            report[v.name] = {
                "sample_size": len(v_results),
                "quality_mean": mean(r.quality_score for r in v_results),
                "quality_std": stdev(r.quality_score for r in v_results) if len(v_results) > 1 else 0,
                "latency_p50": sorted(r.latency_ms for r in v_results)[len(v_results)//2],
                "avg_tokens": mean(r.token_count for r in v_results),
            }
        return report

# 定义两个变体
variant_a = PromptVariant(
    name="v2.0_concise",
    system_prompt="你是数据分析师。简洁回答，不超过200字。",
    weight=0.5
)

variant_b = PromptVariant(
    name="v2.1_structured",
    system_prompt="你是数据分析师。按JSON格式回答，包含analysis、recommendations字段。",
    weight=0.5
)

# 运行测试
test = PromptABTest([variant_a, variant_b])

for query in test_queries:
    variant = test.assign()
    start = time.time()
    output = llm.generate(variant.system_prompt, query)
    latency = (time.time() - start) * 1000
    
    # 人工或自动评分
    score = quality_evaluator.score(output, query)
    
    test.record(TestResult(
        variant=variant.name,
        quality_score=score,
        latency_ms=latency,
        token_count=count_tokens(output),
        user_feedback=""
    ))

# 查看报告
print(json.dumps(test.report(), indent=2))

from scipy import stats

def is_significant(results_a, results_b, alpha=0.05):
    """判断 A/B 测试结果是否统计显著"""
    scores_a = [r.quality_score for r in results_a]
    scores_b = [r.quality_score for r in results_b]
    
    # 样本量检查
    if len(scores_a) < 30 or len(scores_b) < 30:
        return False, "样本量不足（至少需要30）"
    
    # 正态性检验
    _, p_normal = stats.shapiro(scores_a + scores_b)
    if p_normal < 0.05:
        # 非正态分布，用 Mann-Whitney U 检验
        stat, p_value = stats.mannwhitneyu(scores_a, scores_b)
    else:
        # 正态分布，用 t 检验
        stat, p_value = stats.ttest_ind(scores_a, scores_b)
    
    return p_value < alpha, f"p={p_value:.4f}"