{
  "name": "generate_minutes",
  "description": "生成会议纪要",
  "params_schema": { "topic_ids": "array<string>", "format": "enum[markdown,docx]" },
  "preconditions": ["meeting.active == true", "asr.transcript.available == true"],
  "effects": ["artifact.minutes.created"],
  "idempotency_key": "meeting_id+topic_hash",
  "resource_cost": { "cpu": "500m", "latency_p99_ms": 8000 },
  "side_effects": ["notification.webhook"],
  "compensation": "delete_minutes_artifact"
}

PENDING → RUNNING → SUCCESS | FAILED_RETRYABLE | FAILED_NON_RETRYABLE | COMPENSATING | COMPENSATED

# 伪代码：流式话题分段与摘要生成
class StreamingTopicSegmenter:
    def __init__(self, embed_model, llm_summarizer, threshold=0.65, min_seg_tokens=800):
        self.embed_model = embed_model          # 轻量级 Embedding (bge-small-zh)
        self.llm_summarizer = llm_summarizer    # 摘要专用小模型 (Qwen-1.5-4B-Chat)
        self.threshold = threshold              # 语义相似度阈值
        self.buffer = []                        # 当前段 Token 缓冲
        self.last_emb = None                    # 上一段尾部 Embedding
        self.segments = []                      # 已确认段落列表

    def feed(self, asr_token: str, timestamp: float, speaker_id: str):
        self.buffer.append((asr_token, timestamp, speaker_id))
        if len(self.buffer) < self.min_seg_tokens:
            return None
        
        # 1. 计算当前缓冲区尾部语义向量
        tail_text = " ".join([t for t, _, _ in self.buffer[-200:]])  # 取尾部 200 tokens
        curr_emb = self.embed_model.encode(tail_text)
        
        # 2. 语义漂移检测
        if self.last_emb is not None:
            sim = cosine_sim(curr_emb, self.last_emb)
            if sim < self.threshold:  # 话题发生漂移
                return self._flush_segment(curr_emb)
        
        # 3. 强制分段：说话人长独白/议程切换关键词
        if self._detect_hard_boundary(self.buffer):
            return self._flush_segment(curr_emb)
        return None

    def _flush_segment(self, curr_emb):
        full_text = " ".join([t for t, _, _ in self.buffer])
        # 结构化摘要 Prompt：强制输出 JSON，含 decisions, actions, entities, sentiment
        summary_json = self.llm_summarizer.extract_structured(full_text)
        seg = {
            "seg_id": uuid4(),
            "start_ts": self.buffer[0][1],
            "end_ts": self.buffer[-1][1],
            "speakers": list(set(s for _, _, s in self.buffer)),
            "summary": summary_json,
            "embedding": curr_emb,
            "raw_tokens": len(full_text)
        }
        self.segments.append(seg)
        self.last_emb = curr_emb
        self.buffer = []
        # 异步写入 Vector DB & Graph DB
        async_write_memory(seg)
        return seg

def build_context_package(query: str, meeting_id: str, token_budget: int = 3000) -> List[MemoryUnit]:
    # 1. 多路召回
    candidates = []
    candidates += vector_search(query, meeting_id, top_k=20, filter={"layer": "episodic"})
    candidates += graph_traverse(query, meeting_id, depth=2)  # 实体邻域扩展
    candidates += working_memory_fetch(meeting_id)            # 必含最近 3 轮
    
    # 2. 评分融合
    for c in candidates:
        c.score = (0.5 * c.semantic_sim + 
                   0.2 * c.recency_weight + 
                   0.2 * c.entity_overlap(query) + 
                   0.1 * c.business_priority)  # 决策类 > 讨论类 > 闲聊类
    
    # 3. MMR 去重选择
    selected = []
    remaining = sorted(candidates, key=lambda x: -x.score)
    while remaining and estimate_tokens(selected) < token_budget:
        best = remaining.pop(0)
        if all(mmr_sim(best, s) < 0.75 for s in selected):  # 语义去重阈值
            selected.append(best)
    
    # 4. 顺序重排：按时间正序，保证因果链条可读
    return sort_by_timestamp(selected)

{
  "role": "system",
  "content": {
    "identity": "MeetingControlAgent",
    "version": "v3.2.1",
    "capabilities": [
      {"tool": "generate_minutes", "params": {"topic_ids": "string[]", "format": "enum[md,docx]"}, "requires": ["meeting.active", "transcript.ready"]},
      {"tool": "send_notification", "params": {"channel": "enum[im,email]", "recipients": "string[]", "content": "string"}, "requires": ["user.permission.notify"]}
    ],
    "constraints": {
      "pii_handling": "mask_before_log",
      "external_network": "deny_all",
      "max_execution_time_sec": 120,
      "max_tool_calls_per_turn": 8
    },
    "reasoning_protocol": "ChainOfThought:Hidden -> Plan:JSON -> Action:FunctionCall"
  }
}

// Go 实现：工具调用前置校验中间件
func (e *Executor) ValidateAndEnrich(ctx context.Context, call *FunctionCall) error {
    // 1. Schema 校验
    schema := e.registry.GetSchema(call.Name)
    if err := jsonschema.Validate(schema, call.Arguments); err != nil {
        return ErrInvalidParams.Wrap(err)
    }

    // 2. 权限校验 (ABAC)
    subject := authz.Subject{UserID: ctx.UserID, Role: ctx.Role, MeetingID: ctx.MeetingID}
    resource := authz.Resource{Tool: call.Name, Params: call.Arguments}
    if !e.pdp.Allow(subject, "execute", resource) {
        audit.LogRejected(ctx, call, "permission_denied")
        return ErrForbidden
    }

    // 3. 幂等键生成与冲突检测
    idempotencyKey := fmt.Sprintf("%s:%s:%s", ctx.MeetingID, call.Name, hashArgs(call.Arguments))
    if exists, _ := e.redis.SetNX(ctx, "idempotency:"+idempotencyKey, "1", 24*time.Hour).Result(); !exists {
        return ErrDuplicateCall
    }

    // 4. 资源配额预扣 (令牌桶)
    cost := schema.EstimateCost(call.Arguments)
    if !e.quota.TryConsume(ctx.MeetingID, cost) {
        return ErrQuotaExceeded
    }

    // 5. 副作用标记注入上下文，供补偿事务使用
    call.Metadata["compensation"] = schema.CompensationAction
    call.Metadata["side_effects"] = schema.SideEffects
    return nil
}

class ModelRouter:
    def __init__(self):
        self.cascade_chain = [
            ("local_tiny", TinyLLM(), {"max_latency_ms": 200, "confidence_thresh": 0.9}),
            ("local_small", SmallLLM(), {"max_latency_ms": 800, "confidence_thresh": 0.85}),
            ("local_moe", MoELLM(), {"max_latency_ms": 2500, "confidence_thresh": 0.8}),
            ("cloud_large", CloudAPILLM(), {"max_latency_ms": 10000, "confidence_thresh": 0.0}), # 兜底
        ]
        self.speculative_drafter = TinyLLM()  # 用于推测解码加速大模型

    async def route(self, task: TaskContext) -> ModelResponse:
        # 1. 任务画像提取
        profile = self._profile_task(task)  # 估算复杂度、Token数、延迟预算
        
        # 2. 级联尝试
        for name, model, policy in self.cascade_chain:
            if not self._should_try(profile, policy):
                continue
            
            # 3. 推测解码加速 (仅大模型启用)
            if name in ["local_moe", "cloud_large"] and self.speculative_drafter:
                resp = await self._speculative_generate(model, self.speculative_drafter, task)
            else:
                resp = await model.generate(task.prompt, task.constraints)
            
            # 4. 置信度自评
            confidence = await self._self_critique(resp, task)
            if confidence >= policy["confidence_thresh"]:
                return ModelResponse(model=name, content=resp, confidence=confidence)
            
            # 5. 低置信度：记录日志，升级下一级
            logger.warning(f"Model {name} low confidence {confidence:.2f} for task {task.id}, escalating.")
        
        raise RoutingExhaustedError("All models failed confidence threshold")

    async def _speculative_generate(self, target, drafter, task):
        # 标准 Speculative Decoding: Drafter 生成草稿 -> Target 并行验证
        # 此处简化：Drafter 仅生成前 50 tokens 预热，Target 接管
        draft = await drafter.generate(task.prompt, max_new_tokens=50)
        return await target.generate(task.prompt, prefix=draft, max_new_tokens=task.max_tokens)

def compress_prompt(original_prompt: str, budget: int, question: str) -> str:
    # 1. 识别不可压缩区块：系统指令、工具 Schema、当前轮用户指令
    protected = extract_protected_blocks(original_prompt)
    compressible = remove_blocks(original_prompt, protected)
    
    # 2. 计算 Token 重要性 (基于小模型 Attention/Perplexity)
    importance = compute_token_importance(compressible, question)
    
    # 3. 贪心保留高重要性 Token 至预算
    compressed = greedy_select(compressible, importance, budget - count_tokens(protected))
    
    return reassemble(protected, compressed)

# chaos-experiment/meeting-agent-resilience.yaml
apiVersion: chaos-mesh.org/v1alpha1
kind: Schedule
metadata:
  name: weekly-agent-stress
spec:
  schedule: "0 2 * * 1" # 每周一凌晨 2 点
  type: "workflow"
  workflowSpec:
    entry: "main"
    templates:
      - name: "main"
        templateType: "serial"
        children:
          - "inject-asr-latency"
          - "verify-degradation"
          - "inject-im-throttle"
          - "verify-saga-compensation"
          - "recover-all"
      
      - name: "inject-asr-latency"
        templateType: "networkchaos"
        action: "delay"
        mode: "one"
        selector:
          namespaces: ["meeting-prod"]
          labelSelectors: "app=asr-gateway"
        delay:
          latency: "2s"
          correlation: "50"
        duration: "300s"
      
      - name: "verify-degradation"
        templateType: "probe"
        probe:
          type: "http"
          httpProbe:
            url: "http://agent-api/health/degradation-status"
            criteria: "status == 'ASR_FALLBACK_ACTIVE' && p99_latency < 5000"
        pause: "60s" # 观察窗口
      
      - name: "inject-im-throttle"
        templateType: "podchaos"
        action: "kill"
        mode: "fixed-percent"
        value: "30"
        selector:
          namespaces: ["infra-prod"]
          labelSelectors: "app=im-gateway"
        duration: "120s"
      
      - name: "verify-saga-compensation"
        templateType: "probe"
        probe:
          type: "sql"
          sqlProbe:
            dsn: "clickhouse://analytics"
            query: "SELECT count() FROM execution_traces WHERE meeting_id IN (SELECT meeting_id FROM active_meetings) AND status='COMPENSATED' AND tool='send_notification'"
            criteria: "result > 0" # 至少有一个补偿成功案例

graph LR
A[会议纪要/录音] --> B(LLM 抽取器: Schema-Constrained)
B --> C{冲突检测器}
C -- 新知识 --> D[写入 Graph DB]
C -- 冲突(如负责人变更) --> E[生成变更工单 -> 人工确认 -> 版本化更新]
D --> F[向量化同步至 Vector DB]
F --> G[下游 RAG 检索增强]

// 统一执行轨迹 Schema (Protobuf 3)
message ExecutionTrace {
  string trace_id = 1;
  string meeting_id = 2;
  string user_id = 3;
  int64 start_time_ms = 4;
  int64 end_time_ms = 5;
  
  // 意图消歧链路
  message IntentResolution {
    string raw_utterance = 1;
    repeated CandidateIntent candidates = 2;
    string final_intent_id = 3;
    float confidence = 4;
    bool clarification_triggered = 5;
    string clarification_question = 6;
  }
  IntentResolution intent_resolution = 10;

  // 规划图快照
  message PlanGraph {
    repeated Node nodes = 1;
    repeated Edge edges = 2;
    repeated string execution_layers = 3; // 拓扑层级序列
    map<string, ResourceEstimate> resource_budget = 4;
  }
  PlanGraph plan_graph = 11;

  // 节点执行详情
  message NodeExecution {
    string node_id = 1;
    string tool_name = 2;
    map<string, string> input_args = 3; // 脱敏后
    map<string, string> output_summary = 4;
    int64 latency_ms = 5;
    int32 retry_count = 6;
    ExecutionStatus status = 7;
    string error_code = 8;
    string compensation_action = 9;
    bool compensation_executed = 10;
  }
  repeated NodeExecution node_executions = 12;

  // 资源消耗
  message ResourceUsage {
    double model_tokens_in = 1;
    double model_tokens_out = 2;
    double cpu_core_seconds = 3;
    double gpu_seconds = 4;
    int64 vector_db_read_bytes = 5;
    int64 graph_db_query_count = 6;
    double estimated_cost_cny = 7;
  }
  ResourceUsage resource_usage = 13;

  // 合规标记
  string data_classification = 20; // L1/L2/L3
  bool pii_detected = 21;
  repeated string pii_types = 22;
}