Notion AI写作助手：产品架构与实现

import timeimport randomimport hashlibfrom dataclasses import dataclass, fieldfrom typing import List, Dict, Optional, Generatorfrom enum import Enum  class TaskType(Enum):    CONTINUE_WRITING = "continue_writing"    SUMMARIZE = "summarize"    TRANSLATE = "translate"    CHANGE_TONE = "change_tone"    GENERATE_OUTLINE = "generate_outline"  class ModelTier(Enum):    LIGHTWEIGHT = "lightweight"    STANDARD = "standard"    ADVANCED = "advanced"  @dataclassclass NotionBlock:    block_id: str    block_type: str    content: str    children: List[str] = field(default_factory=list)    metadata: Dict = field(default_factory=dict)  @dataclassclass PageContext:    page_id: str    title: str    blocks: List[NotionBlock] = field(default_factory=list)    linked_page_ids: List[str] = field(default_factory=list)    database_properties: Dict = field(default_factory=dict)  class PromptComposer:    """提示词组合器：负责将用户指令、上下文和系统约束组合成完整提示"""     SYSTEM_PROMPT = "你是Notion AI写作助手，一个嵌入在Notion文档编辑器中的智能助手。你的输出应当以Markdown格式呈现，与Notion的Block结构兼容。"     def __init__(self):        self.max_context_tokens = 4000     def compose(self, task_type: TaskType, user_input: str,                context: PageContext, focus_block_id: str) -> str:        context_text = self._extract_context(context, focus_block_id)        task_instruction = self._get_task_instruction(task_type, user_input)        prompt_parts = [            self.SYSTEM_PROMPT,            f"当前文档标题: {context.title}",            f"文档上下文:\n{context_text}",            f"用户指令: {task_instruction}",            "请直接输出内容，不要包含解释性文字。"        ]        return "\n\n".join(prompt_parts)     def _extract_context(self, page: PageContext, focus_id: str) -> str:        focus_idx = 0        for i, block in enumerate(page.blocks):            if block.block_id == focus_id:                focus_idx = i                break        start = max(0, focus_idx - 3)        end = min(len(page.blocks), focus_idx + 4)        context_blocks = page.blocks[start:end]        return "\n".join([b.content for b in context_blocks])     def _get_task_instruction(self, task_type: TaskType, user_input: str) -> str:        instructions = {            TaskType.CONTINUE_WRITING: f"请基于以下上下文继续写作: {user_input}",            TaskType.SUMMARIZE: f"请对以下内容进行精炼总结: {user_input}",            TaskType.TRANSLATE: f"请将以下内容翻译为目标语言: {user_input}",            TaskType.CHANGE_TONE: f"请将以下内容的语气调整为: {user_input}",            TaskType.GENERATE_OUTLINE: f"请根据以下内容生成结构化大纲: {user_input}",        }        return instructions.get(task_type, user_input)  class ModelRouter:    """多模型路由器：根据任务类型和复杂度选择合适的模型层级"""     ROUTING_RULES = {        TaskType.SUMMARIZE: ModelTier.LIGHTWEIGHT,        TaskType.TRANSLATE: ModelTier.STANDARD,        TaskType.CONTINUE_WRITING: ModelTier.ADVANCED,        TaskType.CHANGE_TONE: ModelTier.STANDARD,        TaskType.GENERATE_OUTLINE: ModelTier.ADVANCED,    }     def __init__(self):        self.cache: Dict[str, str] = {}        self.request_count = 0        self.quota_limit = 100     def select_model(self, task_type: TaskType, content_length: int) -> ModelTier:        base_tier = self.ROUTING_RULES.get(task_type, ModelTier.STANDARD)        if content_length > 3000:            return ModelTier.ADVANCED        if content_length < 200 and task_type != TaskType.CONTINUE_WRITING:            return ModelTier.LIGHTWEIGHT        return base_tier     def check_quota(self) -> bool:        return self.request_count < self.quota_limit     def increment_usage(self):        self.request_count += 1     def get_cache_key(self, prompt: str) -> str:        return hashlib.md5(prompt.encode()).hexdigest()     def try_cache(self, prompt: str) -> Optional[str]:        key = self.get_cache_key(prompt)        return self.cache.get(key)     def store_cache(self, prompt: str, response: str):        key = self.get_cache_key(prompt)        self.cache[key] = response  class StreamingOutput:    """流式输出管理器：模拟逐token的流式生成过程"""     def __init__(self):        self.is_interrupted = False     def stream_generate(self, full_text: str) -> Generator[str, None, None]:        tokens = full_text.split(" ")        for i, token in enumerate(tokens):            if self.is_interrupted:                break            suffix = " " if i < len(tokens) - 1 else ""            yield token + suffix            time.sleep(0.02)     def interrupt(self):        self.is_interrupted = True     def reset(self):        self.is_interrupted = False  class NotionAIAssistant:    """Notion AI写作助手核心类：整合提示组合、模型路由和流式输出"""     def __init__(self):        self.prompt_composer = PromptComposer()        self.model_router = ModelRouter()        self.streaming = StreamingOutput()        self.generation_log: List[Dict] = []     def process_request(self, task_type: TaskType, user_input: str,                        context: PageContext, focus_block_id: str) -> str:        if not self.model_router.check_quota():            return "[错误] AI使用配额已耗尽，请升级计划或等待配额重置。"         prompt = self.prompt_composer.compose(            task_type, user_input, context, focus_block_id        )        cached = self.model_router.try_cache(prompt)        if cached:            return f"[缓存命中] {cached}"         content_length = len(prompt)        model_tier = self.model_router.select_model(task_type, content_length)        self.model_router.increment_usage()         simulated_response = self._simulate_model_response(            task_type, user_input, model_tier        )        self.model_router.store_cache(prompt, simulated_response)         log_entry = {            "task_type": task_type.value,            "model_tier": model_tier.value,            "prompt_length": content_length,            "response_length": len(simulated_response),        }        self.generation_log.append(log_entry)         return simulated_response     def stream_request(self, task_type: TaskType, user_input: str,                       context: PageContext, focus_block_id: str) -> List[str]:        full_response = self.process_request(task_type, user_input,                                             context, focus_block_id)        if full_response.startswith("["):            return [full_response]         self.streaming.reset()        chunks = list(self.streaming.stream_generate(full_response))        return chunks     def _simulate_model_response(self, task_type: TaskType,                                 user_input: str, tier: ModelTier) -> str:        responses = {            TaskType.CONTINUE_WRITING: "此外，系统还需要考虑用户权限管理、数据加密存储以及API速率限制等非功能性需求，以确保平台的整体安全性和稳定性。",            TaskType.SUMMARIZE: "本文档主要描述了产品的核心功能模块、技术架构方案以及实施计划，涵盖了从需求分析到上线部署的完整生命周期。",            TaskType.TRANSLATE: "The system architecture follows a microservices pattern, with each service responsible for a specific domain function.",            TaskType.CHANGE_TONE: "我们非常高兴地宣布，经过团队的不懈努力，新产品终于迎来了令人期待的正式发布！",            TaskType.GENERATE_OUTLINE: "一、项目概述\n二、目标用户分析\n三、核心功能设计\n四、技术实现方案\n五、上线计划与里程碑",        }        return responses.get(task_type, "AI已处理您的请求。")  def main():    assistant = NotionAIAssistant()    sample_blocks = [        NotionBlock("b1", "heading", "# 产品需求文档 v2.0"),        NotionBlock("b2", "paragraph", "本文档描述了新一代协作平台的核心需求。"),        NotionBlock("b3", "paragraph", "目标用户为中小型团队的 project manager 和 developer。"),        NotionBlock("b4", "heading", "## 功能性需求"),        NotionBlock("b5", "paragraph", "系统应支持实时多人协作编辑，延迟不超过200ms。"),        NotionBlock("b6", "paragraph", "系统需要提供细粒度的权限控制，包括查看、编辑、管理三级权限。"),        NotionBlock("b7", "paragraph", "系统应支持第三方集成，包括Slack、GitHub和Jira。"),    ]    page_ctx = PageContext(        page_id="page_001",        title="产品需求文档 v2.0",        blocks=sample_blocks,        linked_page_ids=["page_002", "page_003"],        database_properties={"status": "进行中", "priority": "高"}    )     print("=" * 60)    print("Notion AI 写作助手 - 模拟演示")    print("=" * 60)     print("\n--- 场景1: 继续写作 ---")    result = assistant.process_request(        TaskType.CONTINUE_WRITING,        "请继续补充非功能性需求",        page_ctx, "b6"    )    print(f"生成结果: {result}")     print("\n--- 场景2: 流式输出演示 ---")    chunks = assistant.stream_request(        TaskType.SUMMARIZE,        "总结当前文档",        page_ctx, "b1"    )    print("流式输出: ", end="")    for chunk in chunks:        print(chunk, end="", flush=True)    print()     print("\n--- 场景3: 缓存命中演示 ---")    result2 = assistant.process_request(        TaskType.SUMMARIZE,        "总结当前文档",        page_ctx, "b1"    )    print(f"缓存结果: {result2}")     print("\n--- 使用统计 ---")    for log in assistant.generation_log:        print(f"  任务={log['task_type']}, 模型={log['model_tier']}, "              f"提示长度={log['prompt_length']}, 响应长度={log['response_length']}")    print(f"总请求数: {assistant.model_router.request_count}")    print(f"缓存条目数: {len(assistant.model_router.cache)}")  if __name__ == "__main__":    main()