架构设计

向量数据库：RAG时代的基石 所有RAG系统都需要向量数据库。2026年市场已经从"有没有"过渡到"哪个好"——选型不当会带来性能瓶颈、扩展困难、功能不足等问题。 三大主流方案 Milvus：大规模分布式 架构特点 Milvus采用存算分离架构： Coordinator：元数据管理和调度 Worker Node：查询节点和数据节点分离 对象存储：数据持久化（S3/MinIO） 消息队列：变更数据捕获（Kafka/Pulsar） 核心优势 水平扩展：支持十亿级向量 混合检索：向量+标量过滤 多索引：HNSW、IVF、DiskANN、GPU索引 云原生：K8s原生部署 适用场景 大规模生产环境（亿级向量） 需要高可用和水平扩展 团队有K8s运维能力 注意事项 部署复杂度高（微服务架构） 小规模场景（<100万向量）过重 内存占用较大 Qdrant：高性能轻量级 架构特点 Qdrant用Rust编写，单二进制部署： Collection：数据集合 Payload：向量关联的元数据 量化支持：标量量化(SQ8)、乘积量化(PQ)、二进制量化 磁盘索引：支持大于内存的数据集 核心优势 简单部署：单二进制+Docker 高性能：Rust实现，无GC暂停 丰富过滤：Payload过滤能力强大 量化压缩：内存占用极低 适用场景 中小规模（<1亿向量） 快速原型开发 对延迟敏感的场景 运维资源有限的团队 注意事项 分布式能力不如Milvus 不支持多租户隔离 生态插件较少 Weaviate：AI原生设计 架构特点 Weaviate定位为"AI原生数据库"： GraphQL API：内置API层 模块化向量化：内置多种embedding模型 对象存储+向量索引：一体化设计 多模态支持：图像/文本/视频嵌入 核心优势 开箱即用：内置embedding模型 GraphQL接口：前端友好 多模态：原生支持多种数据类型 混合检索：BM25+向量融合 适用场景 快速构建AI应用 需要多模态检索 前端团队主导的项目 不想单独管理embedding流程 注意事项 性能不如Qdrant极致 大规模部署经验较少 Go实现，性能依赖GC调优 性能对比 写入性能 数据库 单线程写入 批量写入(10K) 索引构建速度 Milvus 5K/s 50K/s 中等 Qdrant 8K/s 80K/s 快 Weaviate 3K/s 30K/s 中等 查询性能 数据库 p50延迟 p99延迟 并发QPS Milvus 5ms 20ms 10K+ Qdrant 2ms 8ms 5K+ Weaviate 8ms 30ms 3K+ 注：测试条件为1M向量，768维，HNSW索引，单节点。实际性能受数据规模、硬件、配置影响。 ...

为什么需要边缘AI？ 云计算解决了大规模AI部署问题，但很多场景无法依赖云端： 隐私要求：医疗、金融数据不能上云 延迟要求：自动驾驶、实时翻译需要毫秒级响应 离线场景：野外、飞机模式、网络不稳定 成本控制：大量设备同时运行，云推理成本不可控 边缘AI的目标：在手机、车载芯片、IoT设备上本地运行AI模型。 模型压缩四件套 1. 量化 将模型从FP32压缩到INT8/INT4： 精度 模型大小(7B) 精度损失 推理速度 FP16 14GB 0% 基准 INT8 7GB <1% 1.5-2x INT4 3.5GB 1-3% 2-3x 2-bit 1.75GB 5-10% 3-4x 移动端推荐INT4量化，使用GGUF格式（llama.cpp生态）。常用方案： GPTQ：基于二阶信息的权重量化 AWQ：基于激活分布的权重量化 GGUF Q4_K_M：移动端最常用格式 2. 蒸馏 将大模型知识"蒸馏"到小模型： Teacher模型(70B) → 生成回答/中间表示 → Student模型(1B)学习 任务特定蒸馏：只蒸馏目标任务的能力，而非通用能力。效果通常比直接训练小模型好20-30%。 3. 剪枝 移除不重要的参数： 结构化剪枝：移除整个attention head或MLP中间维度，可实际加速 非结构化剪枝：移除稀疏参数，需特殊硬件支持 移动端推荐结构化剪枝，因为标准硬件就能获得加速。 4. 架构搜索（NAS） 针对特定硬件搜索最优模型结构： 确定目标硬件的延迟/功耗预算 搜索满足约束的最高精度模型 MobileNet、EfficientNet就是NAS的经典成果 移动端推理引擎 llama.cpp C++实现，无依赖 支持GGUF格式 iOS/Android均可运行 支持CPU推理和GPU加速 典型使用: 1. 将模型量化为GGUF格式 2. 编译llama.cpp for Android/iOS 3. 通过JNI或C API调用 4. 管理上下文窗口和内存 MLC-LLM 基于TVM编译器 自动优化模型到目标硬件 支持GPU加速（Metal/Vulkan） 跨平台：iOS/Android/Web ONNX Runtime Mobile 微软出品，支持ONNX格式 Android/iOS支持完整 包体积极小（<10MB运行时） 支持NNAPI/CoreML后端加速 平台原生方案 iOS：CoreML + MLX Android：ML Kit + NNAPI 华为：CANN + NPU 端侧部署实战 iOS部署 // 使用MLX框架运行LLM import MLX import MLXLMCommon let model = try await loadModel(name: "qwen-2-1.5b-4bit") let result = try await model.generate( prompt: "你好", maxTokens: 256 ) 关键配置： ...

数字人的本质 数字人是多模态AI的集大成者——它需要看起来像人、动起来像人、说起来像人。技术上需要融合3D图形、语音合成、NLP、计算机视觉等多个领域。 技术栈分层 第一层：外观创建 3D建模 传统管线：Maya/Blender手工建模→UV展开→纹理绘制→绑定骨骼 AI生成：单张照片生成3D模型（如PifuHD、DreamGaussian） 数字分身：3D扫描+AI增强，快速生成高保真数字人 2D数字人 2D数字人（如HeyGen、D-ID）不需要3D模型，直接在2D图像上做面部动画： 输入：一张正面照片 + 音频 输出：说话的视频 技术：基于扩散模型的面部动画生成 2D方案成本低、速度快，但角度受限。 第二层：语音合成 数字人的"声音"是核心体验。TTS技术已非常成熟： 流式TTS 延迟要求：首音延迟<300ms，支持流式输入 情感控制：通过ssml或prompt控制语气（开心、悲伤、严肃） 多语言：支持中英日韩等多语言自然切换 声音克隆 少量样本（3-10秒）克隆目标音色 使用说话人编码（speaker encoder）技术 需注意Deepfake伦理问题 代表方案 商用：Azure TTS、ElevenLabs、火山引擎 开源：ChatTTS、GPT-SoVITS、CosyVoice 第三层：唇形同步 唇形同步是数字人的技术难点——嘴型需要精确匹配语音内容。 3D方案：BlendShape驱动 预定义一组口型（ARKit的52个BlendShape） 音素到口型映射（ viseme mapping） 音频特征→音素识别→口型映射→BlendShape权重 2D方案：面部驱动 音频特征提取（mel-spectrogram） 基于扩散模型或GAN生成面部帧 时序一致性处理（光流+时间一致性损失） Wav2Lip类方法 核心思想是用音频直接驱动任意视频中的嘴唇运动： 输入：视频 + 新音频 输出：嘴型匹配新音频的视频 局限：仅嘴部区域，其他部位不动 第四层：动作驱动 身体动作 动作捕捉：光学/惯性动捕，精度高但成本高 视频驱动：单摄像头提取3D姿态，驱动数字人 AI生成：文本→动作序列（如动作扩散模型） 表情驱动 面部关键点检测（68/468点） FACS（面部动作编码系统）映射 实时跟踪摄像头数据→表情重定向到数字人 眼神控制 眼神是数字人"活起来"的关键： 注视追踪：让数字人看着对话者 扫视（saccade）：模拟自然眼动 眨眼频率：6-15次/分钟的自然频率 第五层：大脑驱动 数字人的"大脑"由LLM驱动： 用户输入 → ASR(语音转文字) → LLM(生成回复) → TTS(文字转语音) → 面部/动作驱动 关键要求： ...

代码生成的三个时代 第一代：智能补全（2021-2023） GitHub Copilot开创了AI代码补全时代。核心模式是在编辑器中实时预测下一段代码，从单行补全到多行生成。技术上是将光标前后的代码作为上下文送入LLM，生成候选代码。 局限性很明显： 缺乏项目级上下文，只看当前文件 无法理解运行时行为 生成代码质量依赖使用者审查能力 第二代：对话式编程（2023-2025） Cursor、Windsurf等AI IDE引入了对话模式。开发者可以用自然语言描述需求，AI生成完整函数甚至文件。关键改进： 上下文扩展到整个项目（通过embedding检索相关文件） 支持多轮对话迭代修改 自动应用diff编辑到文件 第三代：自主编程Agent（2025-2026） 最新一代代码Agent不再需要人类逐步指导，而是： 理解完整需求描述 自主规划实现步骤 创建/修改多个文件 运行测试验证结果 自主调试修复错误 自主编程Agent的架构 核心组件 [需求理解] → [任务规划] → [代码实现] → [测试执行] → [调试修复] → [完成] ↑ | └──────────────────反馈循环──────────────────────────────┘ 需求理解模块 将自然语言需求转化为结构化的技术规格： 提取功能需求列表 识别技术约束（语言、框架、数据库） 识别非功能需求（性能、安全） 输出：结构化的规格文档 任务规划模块 将规格分解为可执行的子任务序列： 文件创建/修改计划 依赖关系图 每步的验证标准 关键挑战是规划粒度——太粗导致单步过于复杂，太细则浪费token。实践中采用自适应粒度：简单模块整块生成，复杂模块逐步拆解。 代码实现模块 实际生成代码的引擎。关键技术： 1. 上下文构建 不是简单把整个代码库塞进context，而是智能构建： 上下文 = 系统prompt + 项目结构概览(目录树) + 相关文件内容(RAG检索) + 接口定义(类型签名) + 当前任务描述 2. 增量编辑 ...

推理瓶颈在哪里？ 大模型推理的核心瓶颈是自回归生成的串行性：每生成一个token都需要将之前所有token的KV对参与注意力计算。随着序列变长，计算量和显存占用呈线性甚至二次增长。 推理优化的目标可以归纳为三个维度： 降低延迟（首token延迟TTFT + 每token延迟TPOT） 提升吞吐（每秒生成token总数） 减少显存（KV Cache占用） KV Cache：标准但不够 KV Cache是现代推理引擎的标配。在生成阶段，避免重复计算历史token的Key和Value向量，而是缓存复用。但KV Cache的显存占用非常大：以Llama-70B为例，单条请求的KV Cache在4K序列长度下约占2GB显存。 优化方向： PagedAttention（vLLM）：将KV Cache分为固定大小的block，通过页表管理，显著减少碎片 量化KV Cache：将FP16的KV对压缩为INT8甚至INT4，显存减半甚至1/4 稀疏KV Cache：基于注意力分数剪枝，丢弃不重要的KV对 连续批处理（Continuous Batching） 传统static batching需要等待批次内所有请求完成才能释放资源，短请求被长请求拖累。连续批处理（也叫iteration-level batching）在每个forward pass级别动态加入新请求、移除已完成请求： 时间步1: [req_A(token5), req_B(token3), req_C(token1)] 时间步2: [req_A(token6), req_B(token4), req_D(token1)] # C完成，D加入 vLLM和TGI都实现了这一机制，实测吞吐量可提升3-5倍。 投机解码（Speculative Decoding） 投机解码利用一个小模型（draft model）快速生成候选token序列，再由大模型一次性验证。如果小模型的猜测正确率足够高，就能以接近小模型的速度输出大模型质量的token： 小模型生成N个候选token（并行，很快） 大模型对这N个token做一次forward（并行验证） 接受正确的前缀，拒绝处重新采样 Medusa在此基础上扩展，在单次forward中同时预测多个位置的token（多头预测），进一步提升加速比。 量化推理 训练时用FP16/BF16，推理时降低精度： 精度 方法 性能损失 INT8 W8A8 / SmoothQuant <1% INT4 W4A16 / GPTQ / AWQ 1-3% 2-bit BitNet / 1.58bit 训练时需特殊设计 AWQ和GPTQ是目前最流行的4bit权重量化方案，配合Marlin内核可达极高吞吐。需注意，4bit量化在70B以上模型上表现好，但7B级别可能有明显精度下降。 ...

为什么Agent需要记忆？ 传统大语言模型的上下文窗口本质上是"短期工作记忆"，一旦对话超出窗口长度，早期信息就会丢失。而真正的智能体需要在跨会话、跨任务的长周期运作中保持上下文连贯性——这就要求我们构建结构化的记忆系统。 记忆系统的三层架构 1. 工作记忆（Working Memory） 工作记忆对应Agent当前正在处理的上下文，通常映射为LLM的context window。工程实现上需要注意以下几点： 动态裁剪策略：不是简单截断，而是基于注意力权重或摘要压缩来保留关键信息 滑动窗口+摘要：将溢出的历史对话通过小模型生成摘要，注入到新窗口头部 多模态工作区：除文本外，缓存当前任务相关的图片、表格、代码片段 一个典型实现是使用Redis作为热数据存储，配合embedding检索，将相关历史片段动态注入context。 2. 情景记忆（Episodic Memory） 情景记忆记录Agent过去的交互经历，包括时间、场景、行为和结果。实现方式通常是向量数据库存储： 记忆条目结构: { "id": "ep_001", "timestamp": "2026-07-16T10:00:00Z", "task": "数据分析任务", "actions": ["读取CSV", "生成图表", "撰写报告"], "outcome": "成功生成季度销售分析报告", "embedding": [0.12, -0.34, ...], "metadata": {"user_id": "u123", "session_id": "s456"} } 检索时，将当前任务描述编码为向量，在数据库中做kNN查询，召回最相关的N条历史经历。 3. 语义记忆（Semantic Memory） 语义记忆是Agent的"知识库"，存储事实、概念和规则。与RAG系统类似，但更强调知识的结构化组织和增量更新： 本体图谱：使用知识图谱存储实体关系 文档库：向量索引+全文检索双路召回 技能库：Agent掌握的工具调用模式和Prompt模板 记忆的写入与遗忘 记忆系统最关键的不是存储，而是遗忘机制。人脑会自动遗忘无关信息以保持认知效率，Agent同样需要： 衰减权重：每条记忆有一个decay score，随时间递减，长期未被检索的记忆降权 合并压缩：相似记忆条目通过聚类合并，提取共性模式 主动遗忘：低质量记忆（如失败且无参考价值的尝试）直接删除 实践中的策略是设置三层保留周期： 7天内：完整保留所有细节 7-30天：保留摘要和关键决策点 30天以上：仅保留高价值模式和教训 工程实现要点 存储选型 记忆层 推荐存储 理由 工作记忆 Redis / 本地内存 低延迟，频繁读写 情景记忆 Milvus / Pinecone 向量检索为主 语义记忆 Neo4j + 向量库 图查询+向量混合 检索融合 最终注入prompt的记忆来自三层的融合结果。推荐的重排策略： ...

从单兵作战到团队协作 单个Agent的能力受限于单一模型的上下文窗口和推理能力。当任务复杂到需要多种专业能力时，多Agent协作成为必然选择。但如何编排多个Agent高效协作，是一个充满设计权衡的工程问题。 编排模式分类 1. 中心化编排（Orchestrator模式） 一个中心编排器分发任务给多个专业Agent： class Orchestrator: def __init__(self): self.agents = { "researcher": ResearchAgent(), "writer": WriterAgent(), "reviewer": ReviewerAgent(), "fact_checker": FactCheckAgent(), } self.task_decomposer = TaskDecomposer() def execute(self, task): # 1. 任务分解 subtasks = self.task_decomposer.decompose(task) # 2. 分配给合适的Agent results = {} for subtask in subtasks: agent = self._select_agent(subtask) result = agent.execute(subtask, context=results) results[subtask.id] = result # 3. 综合结果 final = self._synthesize(results) return final def _select_agent(self, subtask): """根据子任务类型选择Agent""" if subtask.type == "research": return self.agents["researcher"] elif subtask.type == "writing": return self.agents["writer"] elif subtask.type == "fact_check": return self.agents["fact_checker"] # ... 优势： 控制流清晰，易于调试 可以精确控制执行顺序 中心节点维护全局状态 劣势： 中心节点是性能瓶颈 所有通信经过中心，延迟高 中心节点故障则全系统故障 2. 去中心化编排（P2P模式） Agent之间直接通信，无中心节点： class P2PAgent: def __init__(self, name, capabilities): self.name = name self.capabilities = capabilities self.peers = {} # 已知的其他Agent self.message_queue = asyncio.Queue() async def run(self): """Agent主循环""" while True: message = await self.message_queue.get() if message.type == "task": # 处理任务 if self._can_handle(message.task): result = await self._handle(message.task) await self._send(message.sender, "result", result) else: # 转发给合适的peer peer = self._find_capable_peer(message.task) await self._send(peer, "task", message.task) elif message.type == "result": self._process_result(message) async def _send(self, peer_name, msg_type, content): """直接发送消息给peer""" peer = self.peers[peer_name] await peer.message_queue.put({ "type": msg_type, "content": content, "sender": self.name }) 优势： ...

从工具到平台：AI的操作系统化 传统操作系统管理硬件资源、提供系统调用、运行应用程序。AI操作系统（AI-OS）将这一范式升级：用大模型作为"内核"，用自然语言作为"系统调用"，用Agent作为"应用程序"。这不是科幻——多个团队已在构建雏形。 AI-OS的架构愿景 分层架构 ┌──────────────────────────────────────────┐ │ 用户交互层 │ │ (自然语言 / 多模态 / 脑机接口) │ ├──────────────────────────────────────────┤ │ Agent运行时层 │ │ (Agent调度 / 隔离 / 权限 / 通信) │ ├──────────────────────────────────────────┤ │ LLM内核层 │ │ (推理 / 规划 / 记忆 / 对齐) │ ├──────────────────────────────────────────┤ │ 工具与API层 │ │ (MCP Server / 函数调用 / 外部服务) │ ├──────────────────────────────────────────┤ │ 数据与知识层 │ │ (向量DB / 知识图谱 / 文件系统) │ ├──────────────────────────────────────────┤ │ 硬件抽象层 │ │ (GPU/NPU调度 / 存储 / 网络) │ └──────────────────────────────────────────┘ LLM内核 传统OS内核负责进程调度、内存管理、文件系统。AI-OS的LLM内核负责： ...

规划：Agent的核心能力 AI Agent与聊天机器人的本质区别在于规划能力——将复杂目标分解为可执行步骤并动态调整的能力。从简单的ReAct循环到复杂的树搜索规划，Agent规划算法经历了快速演进。 ReAct：推理与行动的交错 基本循环 ReAct（Reasoning + Acting）是Agent规划的基石，核心思想是交错推理和行动： Thought: 用户要查北京天气，我需要调用天气API Action: search_weather(city="北京") Observation: 北京今天晴，25°C Thought: 获取到天气信息，可以回答用户了 Answer: 北京今天天气晴朗，气温25°C，适合外出活动。 实现细节 class ReActAgent: def __init__(self, llm, tools, max_steps=10): self.llm = llm self.tools = tools self.max_steps = max_steps def run(self, task): trajectory = [] for step in range(self.max_steps): # 生成下一步思考和行为 prompt = self._build_prompt(task, trajectory) response = self.llm.generate(prompt) thought, action = self._parse(response) trajectory.append({"thought": thought, "action": action}) if action["type"] == "finish": return action["answer"] # 执行工具 if action["type"] == "tool": result = self.tools[action["name"]](**action["params"]) trajectory.append({"observation": result}) return "达到最大步数限制" def _build_prompt(self, task, trajectory): prompt = f"""任务：{task} 可用工具：{list(self.tools.keys())} 历史： """ for item in trajectory: if "thought" in item: prompt += f"Thought: {item['thought']}\n" prompt += f"Action: {item['action']}\n" elif "observation" in item: prompt += f"Observation: {item['observation']}\n" prompt += "Thought:" return prompt ReAct的局限 无记忆反馈：失败的经验不会影响后续尝试 线性思维：无法回溯到之前的选择点尝试其他路径 容易陷入循环：在复杂任务中反复尝试相同的失败方案 Reflexion：从失败中学习 核心改进 Reflexion在ReAct基础上增加了自我反思机制。当任务执行失败时，Agent会生成反思并存储到长期记忆中： ...

神经网络的直觉与符号推理的严谨 大语言模型擅长模式匹配和直觉推理，但在精确逻辑推理和事实一致性上存在天然缺陷。知识图谱作为结构化的符号知识表示，恰好互补了LLM的短板。两者的融合——神经符号AI——正在成为构建可靠AI系统的重要方向。 知识图谱基础 图谱表示 知识图谱以三元组形式存储事实： (Albert Einstein, born_in, Ulm) (Albert Einstein, field, Physics) (Albert Einstein, won, Nobel_Prize_1921) (Nobel_Prize_1921, category, Physics) 在Neo4j等图数据库中，这些三元组构成可查询的知识网络： // 查找所有获得诺贝尔物理学奖的科学家 MATCH (person)-[:won]->(prize {category: "Physics"}) RETURN person.name, prize.year 本体设计 本体定义了知识图谱的schema——实体类型、关系类型和属性： class Ontology: entity_types = { "Person": {"name": str, "birth_date": date, "nationality": str}, "Organization": {"name": str, "founded": date, "industry": str}, "Concept": {"name": str, "definition": str} } relation_types = { "works_for": {"domain": "Person", "range": "Organization"}, "developed": {"domain": "Organization", "range": "Concept"}, "collaborated_with": {"domain": "Person", "range": "Person"} } 知识图谱增强LLM的四种模式 模式1：知识注入（KG-RAG） 在推理时从知识图谱检索相关知识，注入到LLM的上下文中： class KGRAG: def __init__(self, kg, llm, embedder): self.kg = kg # 知识图谱 self.llm = llm self.embedder = embedder def query(self, question): # 1. 实体链接 entities = self._extract_entities(question) # 2. 子图检索 subgraph = self._retrieve_subgraph(entities, hops=2) # 3. 路径排序 paths = self._rank_paths(question, subgraph) # 4. 文本化 context = self._serialize_paths(paths) # 5. LLM生成 prompt = f"""基于以下知识图谱信息回答问题： 知识： {context} 问题：{question} """ return self.llm.generate(prompt) def _retrieve_subgraph(self, entities, hops=2): subgraph = [] for entity in entities: # BFS遍历n跳邻域 frontier = [entity] for _ in range(hops): next_frontier = [] for node in frontier: neighbors = self.kg.get_neighbors(node) for neighbor, relation in neighbors: subgraph.append((node, relation, neighbor)) next_frontier.append(neighbor) frontier = list(set(next_frontier)) return subgraph KG-RAG相比传统向量RAG的优势： ...