教程：如何用AutoRAG + Milvus避免RAG 与Agent 中出现串租问题

python3 -m venv .venvsource .venv/bin/activatepip install -U pippip install "autorag>=0.3" "pymilvus>=2.4.0" "openai" "pandas" "pyarrow"export OPENAI_API_KEY=sk-... #自行准备OpenAI_API_KEYexport MILVUS_URI=http://127.0.0.1:19530   # Milvus Standalone 服务地址export MILVUS_TOKEN="root:Milvus"

# 下载docker-compose.ymlwget https://github.com/milvus-io/milvus/releases/download/v2.6.8/milvus-standalone-docker-compose.yml -O docker-compose.yml# 启动Milvus（检查端口映射：19530:19530）docker-compose up -d# 验证服务启动docker ps | grep milvus# 应该看到3个容器：milvus-standalone, milvus-etcd, milvus-minio

import osimport pandas as pdos.makedirs("./data", exist_ok=True)corpus = pd.DataFrame([    {"doc_id": "a-1", "contents": "A租户的报销规则：差旅上限为内部标准。", "metadata": {"tenant_id": "tenant_a"}, "tenant_id": "tenant_a"},    {"doc_id": "a-2", "contents": "A租户合同模板要求法务审批。", "metadata": {"tenant_id": "tenant_a"}, "tenant_id": "tenant_a"},    {"doc_id": "b-1", "contents": "B租户的报销规则：海外差旅需要二级审批。", "metadata": {"tenant_id": "tenant_b"}, "tenant_id": "tenant_b"},    {"doc_id": "b-2", "contents": "B租户合同模板要求采购会签。", "metadata": {"tenant_id": "tenant_b"}, "tenant_id": "tenant_b"},])qa = pd.DataFrame([    {        "qid":"q1",        "query":          "报销规则里差旅审批要求是什么？",        "retrieval_gt":   [["a-1"]],           # List[List[str]]：期望命中的 doc_id 集合        "generation_gt":  ["A租户内部标准。"],  # List[str]：可接受的参考答案        "tenant_id":      "tenant_a",    },    {        "qid":            "q2",        "query":          "报销规则里差旅审批要求是什么？",   # 与 q1 文字完全相同的查询        "retrieval_gt":   [["b-1"]],        "generation_gt":  ["B租户海外差旅需二级审批。"],        "tenant_id":      "tenant_b",    },])corpus.to_parquet("./data/corpus.parquet", index=False)qa.to_parquet("./data/qa.parquet",index=False)

import osfrom pymilvus import MilvusClient, DataTypeclient = MilvusClient(    uri=os.getenv("MILVUS_URI", "http://127.0.0.1:19530"),    token=os.getenv("MILVUS_TOKEN", ""),)COLLECTION = "kb_multi_tenant_pk"if client.has_collection(COLLECTION):    client.drop_collection(COLLECTION)schema = client.create_schema(auto_id=False, enable_dynamic_field=False)schema.add_field("pk",DataType.VARCHAR, is_primary=True,max_length=64)schema.add_field("tenant_id", DataType.VARCHAR, is_partition_key=True, max_length=64)schema.add_field("doc_id",    DataType.VARCHAR, max_length=64)schema.add_field("contents",  DataType.VARCHAR, max_length=2048)# text-embedding-3-small 默认输出1536维度schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=1536)idx = client.prepare_index_params()idx.add_index(field_name="embedding", index_type="AUTOINDEX", metric_type="COSINE")client.create_collection(    collection_name=COLLECTION,    schema=schema,    index_params=idx,    num_partitions=16,   # Partition Key 模式下的物理分区数，默认 16，最大 4096)print(f"✅ Collection '{COLLECTION}' created，Partition Key → tenant_id")

import osimport pandas as pdfrom openai import OpenAIfrom pymilvus import MilvusClientopenai_client = OpenAI()client = MilvusClient(    uri=os.getenv("MILVUS_URI", "http://127.0.0.1:19530"),    token=os.getenv("MILVUS_TOKEN", ""),)COLLECTION = "kb_multi_tenant_pk"def embed(texts: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:    resp = openai_client.embeddings.create(input=texts, model=model)    return [item.embedding for item in resp.data]corpus_df= pd.read_parquet("./data/corpus.parquet")embeddings = embed(corpus_df["contents"].tolist())rows = [    {        "pk":        row["doc_id"],        "tenant_id": row["tenant_id"],# Partition Key 字段，决定物理路由        "doc_id":    row["doc_id"],        "contents":  row["contents"],        "embedding": emb,    }    for (_, row), emb in zip(corpus_df.iterrows(), embeddings)]client.insert(collection_name=COLLECTION, data=rows)client.flush(collection_name=COLLECTION)print(f"✅ Inserted {len(rows)} documents into Milvus")

import osmilvus_uri   = os.getenv("MILVUS_URI", "http://127.0.0.1:19530")milvus_token = os.getenv("MILVUS_TOKEN", "")collection_name = os.getenv("AUTORAG_COLLECTION", "kb_autorag_eval")config = f"""vectordb:  - name: milvus_tenant_store    db_type: milvus    embedding_model: openai_embed_3_small    collection_name: {collection_name}    uri: {milvus_uri}    token: {milvus_token}node_lines:  - node_line_name: retrieve_node_line    nodes:      - node_type: semantic_retrieval        strategy:          metrics: [retrieval_recall, retrieval_precision, retrieval_f1]        top_k: 5        modules:          - module_type: vectordb            vectordb: milvus_tenant_store"""os.makedirs("./config", exist_ok=True)with open("./config/autorag_milvus_tenant.yaml", "w") as f:    f.write(config.strip())print("✅ Config written to ./config/autorag_milvus_tenant.yaml")

# 按租户拆分评测集，保证评测数据不跨租户污染python3 -'EOF'import pandas as pdqa= pd.read_parquet('./data/qa.parquet')corpus = pd.read_parquet('./data/corpus.parquet')for tid in ["tenant_a", "tenant_b"]:    qa[qa["tenant_id"]== tid].to_parquet(f"./data/qa_{tid}.parquet",     index=False)    corpus[corpus["tenant_id"] == tid].to_parquet(f"./data/corpus_{tid}.parquet", index=False)EOF# 分别对两个租户执行评测，结果落到各自的 benchmark 目录# 注意：每个租户使用独立 collection，避免评测数据相互污染for TENANT in tenant_a tenant_b; do    AUTORAG_COLLECTION=kb_autorag_eval_${TENANT} python3 step5_write_config.py    autorag evaluate \    --config           ./config/autorag_milvus_tenant.yaml \    --qa_data_path     ./data/qa_${TENANT}.parquet \    --corpus_data_path ./data/corpus_${TENANT}.parquet \    --project_dir      ./benchmark/${TENANT}done

import osfrom openai import OpenAIfrom pymilvus import MilvusClientCOLLECTION = "kb_multi_tenant_pk"client = MilvusClient(    uri=os.getenv("MILVUS_URI", "http://127.0.0.1:19530"),    token=os.getenv("MILVUS_TOKEN", ""),)openai_client = OpenAI()def embed(texts: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:    resp = openai_client.embeddings.create(input=texts, model=model)    return [item.embedding for item in resp.data]query = "报销规则里差旅审批要求是什么？"query_vector = embed([query])[0]# ✅ 带 tenant 条件查询for tid in ["tenant_a", "tenant_b"]:    results = client.search(        collection_name=COLLECTION,        data=[query_vector],        filter=f'tenant_id == "{tid}"',        limit=5,        output_fields=["doc_id", "tenant_id", "contents"],    )    print(f"=== 查询租户: {tid} ===")    for hit in results[0]:        e = hit["entity"]        print(f"  doc={e['doc_id']}  tenant={e['tenant_id']}  score={hit['distance']:.4f}")        print(f"  → {e['contents'][:40]}...")# ❌ 无过滤，语义相似度跨租户返回print("=== ⚠️  无 tenant 过滤（危险示范）===")results_nf = client.search(    collection_name=COLLECTION,    data=[query_vector],    limit=5,    output_fields=["doc_id", "tenant_id", "contents"],)for hit in results_nf[0]:    e = hit["entity"]    print(f"  doc={e['doc_id']}  tenant={e['tenant_id']}  score={hit['distance']:.4f}")

=== 查询租户: tenant_a ===  doc=a-1  tenant=tenant_a  score=0.6015  → A租户的报销规则：差旅上限为内部标准。...  doc=a-2  tenant=tenant_a  score=0.3933  → A租户合同模板要求法务审批。...=== 查询租户: tenant_b ===  doc=b-1  tenant=tenant_b  score=0.6914  → B租户的报销规则：海外差旅需要二级审批。...  doc=b-2  tenant=tenant_b  score=0.2637  → B租户合同模板要求采购会签。...=== ⚠️  无 tenant 过滤（危险示范）===  doc=b-1  tenant=tenant_b  score=0.6914   ← 两个租户的文档混排  doc=a-1  tenant=tenant_a  score=0.6015  doc=a-2  tenant=tenant_a  score=0.3933  doc=b-2  tenant=tenant_b  score=0.2637

def validate_and_insert(doc: dict):    if not doc.get("tenant_id"):        raise ValueError(            f"doc_id={doc.get('doc_id')} 缺少 tenant_id，拒绝入库。"            "不允许事后补填——无tenant_id 的向量进入集合后无法补救。"        )    client.insert(collection_name=COLLECTION, data=[doc])

#❌ 错误：相信客户端传进来的值，可以被伪造tenant_id   = request.params.get("tenant_id")filter_expr = f'tenant_id == "{tenant_id}"'# ✅ 正确：从服务端验证过的Token 中提取，不可伪造tenant_id   = auth_token.claims["tenant_id"]filter_expr = f'tenant_id == "{tenant_id}"'results = client.search(    collection_name=COLLECTION,    data=[query_vector],    filter=filter_expr,        # 过滤条件由系统注入，不经过客户端    limit=top_k,    output_fields=["doc_id", "contents"],)