AI工具链7月选型总结：LangChain、向量数据库与LLMOps工具生态

AI工具链7月选型总结：LangChain、向量数据库与LLMOps工具生态

一、选型场景：从实验到生产

7月团队面临一个关键节点：AI功能从"内部实验"走向"生产交付"。之前的原型阶段可以用Jupyter Notebook+手动脚本完成。但在生产环境中需要一套完整的工具链：从模型调用到数据管理、从评估到监控。

我在7月系统地评估了AI工具链的三个核心环节：框架层（LangChain/LlamaIndex）、存储层（向量数据库）、运维层（LLMOps）。

本文是评估结果的月度总结。

二、框架层：LangChain vs LlamaIndex vs 原生

评估标准：开发效率、生产稳定性、学习成本、生态成熟度。

LangChain的取舍：

LangChain仍然是使用最广泛的LLM框架。但7月的使用体验证实了社区的一个批评：LangChain过度抽象。

# LangChain的抽象层次（简化概念模型）
from langchain_core.runnables import RunnablePassthrough
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 链式调用 – LangChain的核心模式
def build_qa_chain(llm, retriever):
"""构建问答链"""
template = """基于以下上下文回答问题：

上下文：{context}

问题：{question}

回答："""

prompt = ChatPromptTemplate.from_template(template)

chain = (
{"context": retriever, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)

return chain

LangChain的优势：

• 生态丰富：600+集成，覆盖几乎所有模型和工具。
• 抽象完善：Chains、Agents、Tools等模式成熟。
• 社区活跃：遇到问题容易找到解决方案。

LangChain的问题：

• 版本升级频繁且不兼容（0.x→1.x改动巨大）。
• 过度抽象导致调试困难（一个错误可能穿透5层抽象）。
• 学习曲线陡峭。

LlamaIndex的定位：

如果LangChain是"通用框架"，LlamaIndex就是"数据密集型AI应用的瑞士军刀"。它的核心优势在数据处理管道上。

# LlamaIndex的数据处理管道
from llama_index.core import (
VectorStoreIndex, SimpleDirectoryReader,
Settings, StorageContext
)
from llama_index.embeddings.openai import OpenAIEmbedding

# 数据摄入 → 索引构建 → 查询
class DocumentPipeline:
"""文档处理管道"""

def __init__(self):
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

def index_documents(self, doc_dir: str):
"""从目录构建索引"""
documents = SimpleDirectoryReader(doc_dir).load_data()
index = VectorStoreIndex.from_documents(
documents,
show_progress=True,
# 自定义分块策略
transformations=[
SentenceSplitter(chunk_size=512, chunk_overlap=50),
]
)
return index

def query(self, index, question: str):
"""查询索引"""
query_engine = index.as_query_engine(
response_mode="tree_summarize",
similarity_top_k=5
)
return query_engine.query(question)

选型结论：

核心原则：能用原生SDK解决的不上框架。框架引入的抽象成本需要有足够的复杂度来支付。

三、存储层：向量数据库选型

向量数据库是RAG系统的核心基础设施。

Milvus：性能王者，但运维成本高。需要独立部署K8s集群，内存占用基础就要8GB+。

Chroma：轻量级嵌入式方案，适合原型和低负载。

pgvector：PostgreSQL扩展，零运维增量。

— pgvector的使用（生产级示例）
CREATE EXTENSION vector;

— 创建带向量列的表
CREATE TABLE documents (
id SERIAL PRIMARY KEY,
content TEXT,
embedding vector(1536), — OpenAI text-embedding-3-small维度
metadata JSONB,
created_at TIMESTAMPTZ DEFAULT NOW()
);

— HNSW索引（生产推荐）
CREATE INDEX ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);

— 语义搜索
SELECT id, content,
1 – (embedding <=> query_embedding) AS similarity
FROM documents
WHERE 1 – (embedding <=> query_embedding) > 0.7
ORDER BY embedding <=> query_embedding
LIMIT 10;

选型建议（按数据规模）：

个人选择：团队选择了pgvector + PostgreSQL。理由：

• 已有PostgreSQL，零运维增量。
• 数据量在100万以下，pgvector完全满足。
• 向量数据和非向量数据在同一事务内查询，数据一致性有保证。

四、运维层：LLMOps工具评估

生产环境需要的三个核心运维能力：LLM调用监控、成本追踪、质量评估。

LangSmith（可观测性）：

# LangSmith追踪集成
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "production-qa-bot"

from langsmith import Client

client = Client()

# 查询上周的性能数据
runs = client.list_runs(
project_name="production-qa-bot",
start_time=datetime.now() – timedelta(days=7),
error=True # 只看失败的
)

# 按延迟分布分析
stats = client.read_project_stats(
project_ids=["production-qa-bot"]
)

LangSmith的优势：与LangChain深度集成，零侵入追踪。问题：定价不透明，月费用随调用量线性增长。

DeepEval（评估框架）：

from deepeval import evaluate
from deepeval.metrics import (
AnswerRelevancyMetric,
FaithfulnessMetric,
ContextualRecallMetric
)
from deepeval.test_case import LLMTestCase

# 定义评估用例
test_cases = [
LLMTestCase(
input="退款政策是什么？",
actual_output="您可以在购买后30天内申请全额退款。",
expected_output="30天内可申请全额退款",
retrieval_context=["退款政策:30天全额退款","需保留原包装"]
),
# … 更多用例
]

# 多维度评估
metrics = [
AnswerRelevancyMetric(threshold=0.7),
FaithfulnessMetric(threshold=0.8),
ContextualRecallMetric(threshold=0.7),
]

results = evaluate(test_cases, metrics)
print(f"整体通过率: {results.score}")

成本监控体系：

class LLMCostTracker:
"""LLM调用成本追踪器"""

# 各模型定价 (per 1K tokens)
PRICING = {
'gpt-4o': {'input': 0.0025, 'output': 0.01},
'gpt-4o-mini': {'input': 0.00015, 'output': 0.0006},
'claude-3.5-sonnet': {'input': 0.003, 'output': 0.015},
}

def __init__(self):
self.daily_cost = defaultdict(float)
self.total_tokens = defaultdict(lambda: {'input': 0, 'output': 0})

def track(self, model: str, input_tokens: int, output_tokens: int):
"""记录一次调用的成本"""
price = self.PRICING.get(model, {})
cost = (
input_tokens / 1000 * price.get('input', 0) +
output_tokens / 1000 * price.get('output', 0)
)

today = datetime.now().strftime('%Y-%m-%d')
self.daily_cost[today] += cost
self.total_tokens[model]['input'] += input_tokens
self.total_tokens[model]['output'] += output_tokens

if self.daily_cost[today] > 50: # 日成本超过$50告警
self._alert(f"Daily LLM cost exceeded: ${self.daily_cost[today]:.2f}")

五、总结

核心技术提炼：