CodePilot - 技术方案¶

┌─────────────────────────────────────────────────────────────┐
│                    CodePilot 系统架构                        │
└─────────────────────────────────────────────────────────────┘

                         用户层
┌─────────────────────────────────────────────────────────────┐
│  VSCode 插件  │  Web IDE  │   CLI 工具   │   API          │
└────────┬─────────────┬────────────┬─────────────┬──────────┘
         │             │            │             │
         └─────────────┴─────┬──────┴─────────────┘
                             │
                      ┌──────▼──────┐
                      │  API 网关     │
                      │  FastAPI    │
                      └──────┬──────┘
                             │
              ┌──────────────┼──────────────┐
              │              │              │
       ┌──────▼──────┐ ┌─────▼────┐ ┌──────▼──────┐
       │ 代码生成引擎 │ │ RAG 引擎   │ │  静态分析   │
       │  (LLM)      │ │(向量检索) │ │  (规则)     │
       └──────┬──────┘ └─────┬────┘ └──────┬──────┘
              │              │              │
       ┌──────▼──────┐ ┌─────▼────┐ ┌──────▼──────┐
       │ Fine-tuned  │ │ 向量数据库 │ │  规则引擎   │
       │ CodeLlama   │ │ Qdrant   │ │             │
       └──────┬──────┘ └──────────┘ └─────────────┘
              │
       ┌──────▼──────┐
       │  训练数据   │
       │  Pipeline   │
       └──────┬──────┘
              │
       ┌──────▼──────┐
       │  语料库     │
       │  (GitHub 等) │
       └─────────────┘

class CodeGenerator:
    def __init__(self, model_path: str):
        self.model = load_model(model_path)
        self.tokenizer = load_tokenizer(model_path)

    def generate(self, 
                 prompt: str, 
                 context: dict,
                 max_tokens: int = 512) -> str:
        """
        生成嵌入式代码

        Args:
            prompt: 用户提示 (如"生成 STM32 GPIO 初始化代码")
            context: 上下文 (芯片型号、项目结构等)
            max_tokens: 最大生成长度

        Returns:
            生成的代码
        """
        # 构建完整的 prompt
        full_prompt = self._build_prompt(prompt, context)

        # 调用模型
        response = self.model.generate(
            full_prompt,
            max_new_tokens=max_tokens,
            temperature=0.7,
            top_p=0.95
        )

        # 后处理
        code = self._post_process(response)

        return code

    def _build_prompt(self, prompt: str, context: dict) -> str:
        """构建包含上下文的完整 prompt"""
        template = """你是一个嵌入式系统专家。请根据以下信息生成代码：

芯片：{chip}
框架：{framework}
要求：{requirements}

用户请求：{prompt}

请生成专业、可编译的嵌入式 C 代码：
"""
        return template.format(
            chip=context.get('chip', 'STM32F4'),
            framework=context.get('framework', 'HAL'),
            requirements=context.get('requirements', ''),
            prompt=prompt
        )

    def _post_process(self, code: str) -> str:
        """后处理：去除 markdown、格式化等"""
        # 去除 ```c ... ```
        code = re.sub(r'```c\n?', '', code)
        code = re.sub(r'\n```', '', code)
        return code.strip()

用户上传 PDF → 解析 → 分块 → 向量化 → 存储
                    ↓
用户提问 → 检索相关块 → LLM 生成答案 → 返回

class RAGEngine:
    def __init__(self, vector_db_url: str):
        self.vector_db = QdrantClient(url=vector_db_url)
        self.embedder = BGEM3Embedder()

    def add_document(self, 
                     pdf_path: str, 
                     doc_id: str,
                     chunk_size: int = 512):
        """添加数据手册到知识库"""
        # PDF 解析
        text = self._parse_pdf(pdf_path)

        # 分块
        chunks = self._chunk_text(text, chunk_size)

        # 向量化
        embeddings = self.embedder.encode(chunks)

        # 存储
        self.vector_db.upsert(
            collection_name="datasheets",
            points=[
                {
                    "id": f"{doc_id}_{i}",
                    "vector": emb,
                    "payload": {
                        "text": chunk,
                        "doc_id": doc_id,
                        "page": i // 10  # 估算页码
                    }
                }
                for i, (chunk, emb) in enumerate(zip(chunks, embeddings))
            ]
        )

    def query(self, question: str, chip_model: str = None) -> str:
        """基于数据手册回答问题"""
        # 向量化问题
        query_emb = self.embedder.encode([question])[0]

        # 检索
        results = self.vector_db.search(
            collection_name="datasheets",
            query_vector=query_emb,
            limit=5,
            filter={"chip_model": chip_model} if chip_model else None
        )

        # 生成答案
        context = "\n".join([r.payload["text"] for r in results])
        answer = self._generate_answer(question, context)

        return answer

class RegisterConfigGenerator:
    def __init__(self, datasheet_db):
        self.db = datasheet_db

    def generate(self, 
                 chip_model: str,
                 peripheral: str,
                 config: dict) -> str:
        """
        生成寄存器配置代码

        Args:
            chip_model: 芯片型号 (如"STM32F407")
            peripheral: 外设 (如"GPIO", "UART")
            config: 配置参数

        Example:
            generate("STM32F407", "GPIO", {
                "pin": "PA5",
                "mode": "output",
                "speed": "high"
            })
        """
        # 从数据手册获取寄存器定义
        regs = self.db.get_registers(chip_model, peripheral)

        # 生成配置代码
        code = self._generate_config(regs, config)

        return code

    def _generate_config(self, regs, config) -> str:
        """生成具体配置代码"""
        lines = []
        for reg_name, reg_def in regs.items():
            value = self._calculate_register_value(reg_def, config)
            lines.append(f"    {reg_name} = 0x{value:08X};")

        return "\n".join(lines)

GitHub 嵌入式项目
       ↓
   爬虫抓取
       ↓
   质量过滤
   (语言/Star/更新)
       ↓
   去重
       ↓
   格式化
       ↓
   训练数据集

# docker-compose.dev.yml
version: '3.8'
services:
  api:
    build: ./api
    ports:
      - "8000:8000"
    volumes:
      - ./api:/app
    environment:
      - MODEL_PATH=/models
      - VECTOR_DB_URL=http://qdrant:6333

  qdrant:
    image: qdrant/qdrant
    ports:
      - "6333:6333"
    volumes:
      - qdrant_storage:/qdrant_storage

  redis:
    image: redis:alpine
    ports:
      - "6379:6379"

volumes:
  qdrant_storage:

用户 → Cloudflare CDN → Load Balancer
                          ↓
              ┌───────────┼───────────┐
              ↓           ↓           ↓
         API Server 1  API Server 2  API Server 3
              ↓           ↓           ↓
              └───────────┼───────────┘
                          ↓
                    ┌─────┴─────┐
                    ↓           ↓
              Qdrant Cluster  PostgreSQL

收入 = $19/用户/月
成本 = $1,020/月

盈亏平衡：1020 / 19 ≈ 54 个付费用户

目标：
- 第 3 月：10 用户 → 亏损 $830
- 第 6 月：100 用户 → 盈利 $880
- 第 12 月：500 用户 → 盈利 $8,480