Vectara 2026: 准确率超 90% 的 RAG-as-a-Service 平台 — API 集成与基准测试

# 注册后，前往控制台获取凭证：
# - Customer ID
# - Corpus ID  
# - API Key

# 存储为环境变量
export VECTARA_CUSTOMER_ID="your-customer-id"
export VECTARA_CORPUS_ID="your-corpus-id"
export VECTARA_API_KEY="zwt-your-api-key"

# 安装官方 Vectara Python 客户端
pip install vectara

# 或直接通过 REST API 使用 requests
pip install requests

from vectara import VectaraClient

# 初始化客户端
client = VectaraClient(
    customer_id="your-customer-id",
    api_key="zwt-your-api-key"
)

# 创建语料库（文档集合）
corpus = client.create_corpus(
    name="product-documentation",
    description="Technical docs for our API platform"
)

# 索引文档
document = {
    "documentId": "api-guide-v2",
    "title": "API Integration Guide v2.0",
    "metadataJson": json.dumps({"version": "2.0", "category": "technical"}),
    "parts": [
        {
            "text": "The Vectara Query API accepts JSON payloads with three required fields: query, corpusKey, and numResults.",
            "metadataJson": json.dumps({"section": "authentication"})
        },
        {
            "text": "Authentication uses OAuth 2.0 client credentials flow. Obtain your client ID and secret from the Vectara Console.",
            "metadataJson": json.dumps({"section": "authentication"})
        }
    ]
}

client.index_document(corpus_id=corpus.corpus_id, document=document)
print(f"Document indexed to corpus {corpus.corpus_id}")

# RAG 查询
response = client.query(
    corpus_id="your-corpus-id",
    query="How do I authenticate with the Query API?",
    num_results=5,
    generate=True,  # 启用生成式摘要
    generation_config={
        "max_tokens": 256,
        "temperature": 0.0,  # 事实性响应
        "citation_style": "numeric"  # 包含来源引用
    }
)

print("Answer:", response.summary)
print("\nSources:")
for idx, result in enumerate(response.search_results, 1):
    print(f"[{idx}] {result.text[:100]}... (score: {result.score:.3f})")

import os
from pathlib import Path

# 批量上传目录中所有 PDF
pdf_dir = Path("./documentation")
for pdf_file in pdf_dir.glob("*.pdf"):
    with open(pdf_file, "rb") as f:
        client.upload_file(
            corpus_id="your-corpus-id",
            file_content=f.read(),
            file_name=pdf_file.name,
            metadata={"source": "docs", "format": "pdf"}
        )
    print(f"Uploaded: {pdf_file.name}")

print("Batch upload complete!")

# 查询端点
curl -X POST "https://api.vectara.io/v1/query" \
  -H "x-api-key: ${VECTARA_API_KEY}" \
  -H "customer-id: ${VECTARA_CUSTOMER_ID}" \
  -H "Content-Type: application/json" \
  -d '{
    "query": [
      {
        "query": "What are the pricing tiers?",
        "numResults": 10,
        "corpusKey": [{"customerId": "'"${VECTARA_CUSTOMER_ID}"'", "corpusId": "'"${VECTARA_CORPUS_ID}"'"}],
        "summary": [{"maxSummarizedResults": 5, "responseLang": "eng"}]
      }
    ]
  }'

import { VectaraClient } from "@vectara/sdk";

const client = new VectaraClient({
  apiKey: process.env.VECTARA_API_KEY!,
  customerId: process.env.VECTARA_CUSTOMER_ID!,
});

async function askQuestion(query: string) {
  const response = await client.query({
    corpusId: process.env.VECTARA_CORPUS_ID!,
    query,
    numResults: 5,
    generate: true,
  });

  return {
    answer: response.summary,
    sources: response.searchResults.map((r) => ({
      text: r.text,
      score: r.score,
      documentId: r.documentId,
    })),
  };
}

// Express.js 端点
app.post("/api/rag", async (req, res) => {
  const result = await askQuestion(req.body.question);
  res.json(result);
});

# 按元数据字段过滤
response = client.query(
    corpus_id="your-corpus-id",
    query="API rate limits",
    num_results=10,
    metadata_filter="doc.version >= '2.0' AND doc.category = 'technical'",
    generate=True
)

# 带日期范围的复杂过滤
response = client.query(
    corpus_id="your-corpus-id",
    query="Recent security updates",
    metadata_filter="doc.date >= '2026-01-01' AND doc.type = 'security-bulletin'",
    generate=True
)

# 用英语查询西班牙语文档
response = client.query(
    corpus_id="your-corpus-id",
    query="What are the safety guidelines?",
    response_lang="eng",  # 响应语言
    # 文档可以是西班牙语、德语、日语等
    # Boomerang 自动跨语言检索
)

# 用中文查询
response = client.query(
    corpus_id="your-corpus-id",
    query="如何集成API？",
    response_lang="zho"
)

import json

# SSE 流式传输用于聊天应用
response = client.query(
    corpus_id="your-corpus-id",
    query="Explain our refund policy",
    generate=True,
    stream=True  # 启用 Server-Sent Events
)

# 处理流式块
for chunk in response:
    if chunk.type == "search_result":
        print(f"Source: {chunk.document_id}")
    elif chunk.type == "generation":
        print(chunk.text, end="", flush=True)  # 流式输出 token

# 配置混合搜索权重
response = client.query(
    corpus_id="your-corpus-id",
    query="authentication errors",
    num_results=10,
    search_config={
        "lexical_interpolation": 0.3,  # 30% 关键词，70% 语义
        "reranker": {
            "type": "mmr",  # 最大边际相关性
            "diversity_bias": 0.2
        }
    },
    generate=True
)

# MMR 重排序获取多样化结果
response = client.query(
    corpus_id="your-corpus-id",
    query="cloud deployment options",
    search_config={
        "reranker": {
            "type": "mmr",
            "diversity_bias": 0.3  # 越高 = 来源越多样化
        }
    }
)

# 自定义评分权重
response = client.query(
    corpus_id="your-corpus-id",
    query="security best practices",
    search_config={
        "reranker": {
            "type": "slingshot",  # Vectara 的神经重排序器
            "cutoff": 0.7  # 最低相关度分数
        }
    }
)

# 更新特定文档
document_update = {
    "documentId": "api-guide-v2",
    "title": "API Integration Guide v2.1",
    "metadataJson": json.dumps({"version": "2.1", "category": "technical"}),
    "parts": [
        {
            "text": "Updated: The Query API now supports batch requests up to 100 queries per call.",
            "metadataJson": json.dumps({"section": "batch-operations"})
        }
    ]
}

# 重新索引以原子方式替换文档
client.index_document(
    corpus_id="your-corpus-id",
    document=document_update
)

response = client.query(
    query="authentication timeout",
    corpus_keys=[
        {"corpusId": "product-docs", "weight": 0.6},
        {"corpusId": "support-tickets", "weight": 0.3},
        {"corpusId": "engineering-wiki", "weight": 0.1}
    ],
    generate=True
)

# 存储对话历史
conversation = []

def chat_turn(user_query: str) -> str:
    global conversation
    
    response = client.query(
        corpus_id="your-corpus-id",
        query=user_query,
        generate=True,
        chat_config={
            "store": True,
            "conversationId": "conv_001",
            "turns": conversation[-5:]  # 最近 5 轮用于上下文
        }
    )
    
    conversation.append({"role": "user", "text": user_query})
    conversation.append({"role": "assistant", "text": response.summary})
    
    return response.summary

# 获取语料库统计信息
stats = client.get_corpus_stats(corpus_id="your-corpus-id")
print(f"Documents: {stats.num_docs}")
print(f"Total parts: {stats.num_parts}")
print(f"Avg document size: {stats.avg_doc_size} bytes")

# 查询分析
analytics = client.get_query_analytics(
    corpus_id="your-corpus-id",
    start_date="2026-04-01",
    end_date="2026-05-19"
)
print(f"Total queries: {analytics.total_queries}")
print(f"Avg latency: {analytics.avg_latency_ms}ms")
print(f"Hallucination rate: {analytics.hallucination_rate}%")