当团队把大模型接入业务时，最常碰到的瓶颈不是算力，而是数据检索与状态持久化。Cosmos Conf 2026 传出的信号很明确：AI 正在重塑应用开发，而 Azure Cosmos DB 正在从传统的 NoSQL 数据库演化为 AI 原生架构的核心基础设施。团队不再只是把数据库当作存 JSON 的仓库，而是利用它的全局分布、低延迟读写和新增的向量检索能力，直接在数据层构建 RAG（检索增强生成）和智能体记忆。
向量检索成为数据层的标配
过去做 RAG，开发者通常需要单独部署一个向量数据库（比如 Pinecone 或 Milvus），再通过中间层把业务数据和向量数据对齐。这种双库架构带来的数据一致性问题和运维成本，在规模化时尤为明显。
Cosmos DB 的 NoSQL API 已经原生支持向量索引。这意味着你可以把业务文档和它的 Embedding 向量存进同一个 Item。查询时，不再需要先查向量库拿 ID，再回业务库捞详情——一条包含 VectorDistance 的 SQL 查询就能搞定语义检索与数据过滤的联合查询。
这种单库模式对 AI 应用至关重要：大模型在推理时往往需要结合元数据（如权限标签、时间戳、地域）来裁剪上下文，把向量和元数据放在一起，过滤逻辑可以直接下推到数据库引擎，避免把无关数据拉进应用层再做剔除。
长上下文与对话记忆的持久化
大模型的上下文窗口在扩大，但无限上下文依然昂贵且容易失焦。生产级的 AI Agent 通常采用"记忆外置"策略：只把当前任务相关的摘要或片段塞进 Prompt，完整的对话历史和用户画像存储在高速外部存储中。
Cosmos DB 的毫秒级读写和按会话 ID 分区的设计，天然适合做这种状态存储。每次 Agent 运行时，按 sessionId 拉取历史，推理结束后把新状态写回。配合 Change Feed，还能轻松实现异步的事件驱动架构——比如用户提问后，Agent 自动触发后台的知识库更新或通知推送。
实战：用 Python 搭建一个 Cosmos DB 驱动的 RAG 服务
下面是一个可以直接改造使用的 Python 示例，演示如何将业务文本连同向量一起写入 Cosmos DB，并执行带元数据过滤的向量检索。运行前需安装依赖：
pip install azure-cosmos openai
并确保你已配置环境变量 COSMOS_CONNECTION_STRING 和 OPENAI_API_KEY。
import os
from azure.cosmos import CosmosClient, PartitionKey
from openai import OpenAI
# 初始化客户端
cosmos_client = CosmosClient.from_connection_string(os.environ["COSMOS_CONNECTION_STRING"])
openai_client = OpenAI()
DATABASE_NAME = "AiAppDb"
CONTAINER_NAME = "KnowledgeBase"
# 创建数据库与容器（如果不存在）
database = cosmos_client.create_database_if_not_exists(DATABASE_NAME)
# 关键：分区键设为 /department，便于后续按部门过滤
container = database.create_container_if_not_exists(
CONTAINER_NAME,
PartitionKey(path="/department"),
offer_throughput=400
)
def ingest_document(text: str, doc_id: str, department: str):
"""将文本与其向量一起写入 Cosmos DB"""
# 调用 OpenAI 生成 Embedding
embedding = openai_client.embeddings.create(
input=text, model="text-embedding-3-small"
).data[0].embedding
# 构造文档：业务字段与向量共存
item = {
"id": doc_id,
"department": department, # 分区键与过滤字段
"content": text,
"embedding": embedding
}
container.upsert_item(item)
print(f"文档 {doc_id} 已写入")
def search_knowledge(query: str, department: str, top_k: int = 3):
"""执行带元数据过滤的向量检索"""
query_embedding = openai_client.embeddings.create(
input=query, model="text-embedding-3-small"
).data[0].embedding
# Cosmos DB NoSQL API 向量检索语法
# 先按部门过滤，再在结果中计算向量距离并排序
query_str = """
SELECT TOP @top_k c.id, c.content, c.department,
VectorDistance(c.embedding, @embedding) AS SimilarityScore
FROM c
WHERE c.department = @department
ORDER BY VectorDistance(c.embedding, @embedding)
"""
params = [
{"name": "@top_k", "value": top_k},
{"name": "@embedding", "value": query_embedding},
{"name": "@department", "value": department}
]
results = container.query_items(
query=query_str,
parameters=params,
enable_cross_partition_query=False # 单分区查询，RU 消耗更低
)
for item in results:
print(f"[{item['department']}] (Score: {item['SimilarityScore']:.4f}) {item['content']}")
# --- 运行示例 ---
# ingest_document("公司差旅报销需在出差结束后 5 个工作日内提交", "policy-001", "HR")
# ingest_document("生产环境部署必须经过安全扫描和灰度审批", "policy-002", "Engineering")
# search_knowledge("出差回来怎么报销？", "HR")
这段代码的核心在于 WHERE c.department = @department 与 VectorDistance 的联合使用。对于 AI 应用来说，检索从来不是纯粹的语义相似，而是"在特定业务约束下的语义相似"。把过滤条件下推到数据库，能大幅减少无效 Token 的消耗。
落地建议与边界考量
把 Cosmos DB 当作 AI 应用的数据底座，收益显而易见，但工程上仍有几个需要权衡的点：
- 分区键设计决定检索成本：向量检索如果在跨分区模式下运行，RU（请求单位）消耗会随数据量线性飙升。设计容器时，分区键必须对齐你最常用的过滤维度（如租户 ID、业务线），确保大多数 RAG 查询命中单分区。
- 向量维度与存储开销：Embedding 维度从 384 到 3072 不等。高维向量带来更精准的语义，但也直接推高存储成本和索引计算量。如果业务对精度要求有限，优先选择小维度模型（如
text-embedding-3-small），或在入库前做降维投影。 - 近似检索的精度边界：Cosmos DB 当前提供的是近似最近邻（ANN）索引，而非精确 KNN。在法律合规、医疗诊断等对召回率有严苛要求的场景，建议在 ANN 检索后，对 Top-K 结果再做一层业务规则校验或二次精确计算，防止关键文档被漏召回。
- Change Feed 驱动异步 Agent：不要只把 Cosmos DB 当作被动存储。利用它的 Change Feed，当新知识入库或用户状态变更时，可以立即触发 Azure Functions 或 Semantic Kernel Agent 进行后台推理与通知，构建真正的事件驱动型 AI 系统。
AI 原生应用不是给旧系统加一个 API 调用，而是从数据层开始重新组织检索、记忆与事件流。Cosmos DB 正在补齐这块拼图中最关键的一角。