人类已进入主动适应AI的阶段

[会议音频] 
    ↓ (实时转录，低延迟)
[钉钉 A1] → 初版纪要（速度快，80% 准确）
    ↓ (异步处理，高推理)
[悟空 Agent] → 纠错 + 补全 + 结构化（精度高，95%+ 准确）
    ↓ (人工确认)
[最终纪要] → 归档 + 待办分发

from dataclasses import dataclass
from typing import Optional

@dataclass
class MeetingCorrectionPrompt:
    """悟空纠错层 Prompt —— 人类为 AI 设计的验证协议"""
    project_context: str           # 项目背景、技术栈、关键术语表
    participants: list[str]        # 参会人及其角色
    a1_draft: str                  # A1 生成的初版纪要
    known_decisions: list[str]     # 会前已知的决策点
    
    def render(self) -> str:
        return f"""## Role
你是会议纪要纠错专家。你的任务是对钉钉 A1 生成的初版纪要进行精确校正。

## Project Context
{self.project_context}

## Participants
{chr(10).join(f"- {p}" for p in self.participants)}

## A1 Draft
{self.a1_draft}

## Correction Rules
1. **术语校正**：对照项目上下文修正技术术语。保留原始缩写，不要过度展开。
2. **意图分类**：将每条内容标记为 [讨论] / [决策] / [待办]。
   - 只有明确说"我们决定""就这么做"的才算 [决策]
   - "考虑一下""后续看看"属于 [讨论]
3. **行动项提取**：识别所有待办，包括隐性指派。
   - 格式：[待办] 任务描述 @负责人 (截止日期)
4. **不要添加 A1 未提及的内容**。你的职责是纠错，不是创作。

## Output Format
按以下结构输出：
### 校正后的纪要
（逐条列出，标注 [讨论]/[决策]/[待办]）

### 变更说明
（列出你做了哪些校正及原因）
"""

# generated by hugo AI

[日常收集阶段]
钉钉聊天记录 ─┐
钉钉文档       ├→ [钉钉收藏/标签] → 原始素材池（未处理）
网页剪藏       ┘

[编译阶段]（每日/每周触发）
原始素材池
    ↓ (提取、去重、关联)
[悟空 Agent] → 结构化知识库（Markdown + 向量索引）
    ↓ (生成元数据)
[知识图谱] → 标签、关联、时间线

[查询阶段]
自然语言提问
    ↓
[悟空 RAG] → 基于编译后知识库的回答（可追溯引用）

from dataclasses import dataclass
from enum import Enum

class ContentType(Enum):
    DECISION = "decision"      # 决策记录
    INSIGHT = "insight"        # 洞察/经验
    REFERENCE = "reference"    # 参考资料
    PROBLEM = "problem"        # 问题与解法

@dataclass
class CollectionRule:
    """人类定义的收集规则 —— 价值判断仍由人完成"""
    content_type: ContentType
    trigger_keywords: list[str]
    source: str  # 钉钉群/文档/网页
    retention_days: int
    
    def should_collect(self, text: str) -> bool:
        return any(kw in text for kw in self.trigger_keywords)

# 示例规则
RULES = [
    CollectionRule(
        content_type=ContentType.DECISION,
        trigger_keywords=["决定", "确定", "方案", "结论"],
        source="钉钉项目群",
        retention_days=365
    ),
    CollectionRule(
        content_type=ContentType.PROBLEM,
        trigger_keywords=["报错", "解决", "排查", "原因"],
        source="钉钉技术群",
        retention_days=180
    ),
]

# generated by hugo AI

@dataclass
class KnowledgeCompilationPrompt:
    """悟空知识编译 Prompt —— 将碎片信息编译为结构化知识"""
    raw_materials: list[dict]  # 原始素材：{source, content, timestamp, author}
    existing_knowledge: str    # 已有知识库（用于去重与关联）
    
    def render(self) -> str:
        materials_str = "\n\n".join(
            f"### 素材 {i+1}\n来源: {m['source']}\n时间: {m['timestamp']}\n"
            f"作者: {m['author']}\n内容:\n{m['content']}"
            for i, m in enumerate(self.raw_materials)
        )
        
        return f"""## Role
你是知识编译器。你的任务是将碎片化信息编译为结构化、可追溯的知识条目。

## Raw Materials
{materials_str}

## Existing Knowledge Base
{self.existing_knowledge}

## Compilation Rules
1. **不丢失细节**：保留原始数据、代码片段、关键引用。不要过度概括。
2. **可追溯**：每条知识必须标注来源（素材编号）。
3. **去重与合并**：如果与已有知识重复，标注为 [更新] 或 [补充]。
4. **结构化输出**：每条知识按以下格式：

### [类型] 标题
- **来源**：素材 X
- **时间**：YYYY-MM-DD
- **上下文**：（简要说明背景）
- **内容**：（保留原始细节）
- **关联**：（指向相关知识条目）
- **标签**：#tag1 #tag2

5. **不添加原始素材中不存在的内容**。如果信息不完整，标注 [待补充]。

## Output
按上述格式输出编译结果。
"""

# generated by hugo AI

用户："上次我们讨论 gRPC 超时问题的结论是什么？"

悟空 RAG:
→ 检索编译后的知识条目
→ 返回：
  [决策] gRPC 超时配置方案
  来源：素材 3（钉钉技术群 2026-05-12）
  结论：统一使用 30s 默认超时，关键路径配置 60s
  关联：[问题] gRPC 超时导致级联故障（素材 2）

收集数据 → 人工标注 → 微调训练 → 评估上线
周期：4-8 周
成本：标注 + 算力 + 工程

from dataclasses import dataclass
from enum import Enum
from datetime import datetime

class FeedbackType(Enum):
    ACCEPTED_DIRECTLY = "accepted"       # 直接采纳
    ACCEPTED_WITH_EDITS = "edited"       # 修改后采纳
    REGENERATED = "regenerated"          # 重新生成
    IGNORED = "ignored"                  # 忽略
    LIKED = "liked"                      # 点赞
    REPORTED = "reported"                # 举报错误

@dataclass
class MeetingFeedbackSignal:
    """从钉钉会议纪要工作流中采集的反馈信号"""
    meeting_id: str
    output_id: str              # AI 输出版本 ID
    feedback_type: FeedbackType
    edit_distance: float        # 修改幅度（0-1，0 为无修改）
    user_role: str              # 反馈者角色（主持人/参会人）
    timestamp: datetime
    context: dict               # 上下文：会议类型、时长、参会人数等
    
    def to_reward(self) -> float:
        """将反馈行为转化为奖励值"""
        base_rewards = {
            FeedbackType.ACCEPTED_DIRECTLY: 1.0,
            FeedbackType.ACCEPTED_WITH_EDITS: 0.5,
            FeedbackType.REGENERATED: -0.5,
            FeedbackType.IGNORED: -0.3,
            FeedbackType.LIKED: 1.5,
            FeedbackType.REPORTED: -1.0,
        }
        reward = base_rewards[self.feedback_type]
        
        # 编辑距离惩罚：修改越多，奖励越低
        if self.feedback_type == FeedbackType.ACCEPTED_WITH_EDITS:
            reward *= (1.0 - self.edit_distance)
        
        # 角色权重：主持人的反馈权重更高
        role_weight = 1.5 if self.user_role == "主持人" else 1.0
        
        return reward * role_weight

# generated by hugo AI

@dataclass
class KnowledgeQueryFeedback:
    """从悟空知识查询中采集的反馈信号"""
    query_id: str
    response_id: str
    user_action: str            # "click_source" / "follow_up" / "copy" / "ignore"
    time_spent: float           # 用户在回答上停留的时间（秒）
    follow_up_query: str | None # 是否有追问
    
    def to_reward(self) -> float:
        """基于用户行为的奖励函数"""
        reward = 0.0
        
        # 复制/引用回答 → 高质量信号
        if self.user_action == "copy":
            reward += 0.8
        
        # 点击查看原始来源 → 用户信任但需要验证（中性）
        if self.user_action == "click_source":
            reward += 0.2
        
        # 有追问 → 首次回答不完整
        if self.follow_up_query:
            reward -= 0.3
        
        # 停留时间过短 → 可能不相关
        if self.time_spent < 3.0:
            reward -= 0.2
        # 停留时间适中 → 认真阅读
        elif 5.0 < self.time_spent < 30.0:
            reward += 0.3
        
        return max(-1.0, min(1.5, reward))  # 截断到 [-1, 1.5]

# generated by hugo AI

from typing import Optional
import numpy as np

class RewardModel:
    """基于人类反馈的奖励模型"""
    
    def __init__(self, 
                 explicit_weight: float = 1.0,
                 implicit_weight: float = 0.6,
                 temporal_decay: float = 0.95):
        self.explicit_weight = explicit_weight    # 显式反馈权重（点赞/举报）
        self.implicit_weight = implicit_weight    # 隐式反馈权重（采纳/忽略）
        self.temporal_decay = temporal_decay      # 时间衰减因子
    
    def compute_reward(self, 
                       explicit_signal: Optional[float],
                       implicit_signal: Optional[float],
                       days_since_interaction: int = 0) -> float:
        """
        计算综合奖励值
        
        Args:
            explicit_signal: 显式反馈（点赞=+1.5, 举报=-1.0, None=无）
            implicit_signal: 隐式反馈（采纳=+1.0, 忽略=-0.3, None=无）
            days_since_interaction: 交互后经过的天数（用于延迟反馈）
        
        Returns:
            综合奖励值 [-1.5, 2.0]
        """
        reward = 0.0
        weight_sum = 0.0
        
        if explicit_signal is not None:
            reward += explicit_signal * self.explicit_weight
            weight_sum += self.explicit_weight
        
        if implicit_signal is not None:
            reward += implicit_signal * self.implicit_weight
            weight_sum += self.implicit_weight
        
        if weight_sum == 0:
            return 0.0  # 无反馈信号
        
        # 归一化
        normalized_reward = reward / weight_sum
        
        # 时间衰减：延迟反馈的权重递减
        temporal_weight = self.temporal_decay ** days_since_interaction
        
        return normalized_reward * temporal_weight
    
    def batch_compute_rewards(self, feedback_batch: list[dict]) -> np.ndarray:
        """批量计算奖励值，用于 DPO/GRPO 训练"""
        rewards = []
        for fb in feedback_batch:
            r = self.compute_reward(
                explicit_signal=fb.get("explicit"),
                implicit_signal=fb.get("implicit"),
                days_since_interaction=fb.get("days", 0)
            )
            rewards.append(r)
        return np.array(rewards)

# generated by hugo AI

[用户使用钉钉/悟空]
    ↓ (自然行为)
[反馈信号采集] → 隐式 + 显式反馈
    ↓ (奖励函数)
[奖励模型] → 每条 (prompt, response) 的奖励值
    ↓ (偏好对构建)
[DPO/GRPO 训练] → 模型参数更新
    ↓ (评估 + 灰度)
[新版模型上线]
    ↓
[用户继续使用] → 新的反馈信号...

@dataclass
class AdaptationCost:
    """适应成本 vs. 等待成本"""
    prompt_engineering_hours: float  # Prompt 工程投入
    workflow_redesign_hours: float   # 工作流重构投入
    training_hours: float            # 团队学习成本
    
    @property
    def total_adaptation_cost(self) -> float:
        return (self.prompt_engineering_hours + 
                self.workflow_redesign_hours + 
                self.training_hours)

@dataclass
class WaitingCost:
    model_improvement_months: float  # 等待模型提升的月数
    monthly_opportunity_cost: float  # 每月机会成本
    competitor_advantage_loss: float # 竞争劣势损失
    
    @property
    def total_waiting_cost(self) -> float:
        return (self.model_improvement_months * self.monthly_opportunity_cost +
                self.competitor_advantage_loss)

# 经验法则：适应成本通常是等待成本的 1/5 - 1/10
# generated by hugo AI

人类适应 AI（高质量输入 + 结构化工作流）
    ↓
AI 输出质量提升
    ↓
人类自然行为产生高质量反馈信号
    ↓
奖励模型 + 强化学习（DPO/GRPO）
    ↓
AI 能力进化 → 适应成本降低
    ↓
人类进一步适应...