初始化Milvus向量库连接

聊聊我这两年搞技术探索与实践踩过的坑

晚上十一点半，刚合上电脑，揉了揉发酸的眼睛。我住在上海张江这边，为了多睡半小时，特意在公司附近租了个老破小，每天骑个小电驴五分钟就能到工位。作为一个常年混迹在各大技术社区、靠评测各种AI编程工具混口饭吃的博主，我平时的生活基本就是被Cursor、Windsurf、Copilot这些工具包围着。说实话，我现在写业务代码，重度依赖ChatGPT和Claude辅助，有时候连个正则表达式都懒得自己写，直接让Claude给我生成。

但评测归评测，真到了公司里落地项目，那些AI工具顶多帮我提提效，真正要扛事儿的，还是得靠自己这些年攒下来的工程化底子。今天不聊那些花里胡哨的AI编程工具评测，想跟大家掏心窝子聊聊我最近在业务里搞技术探索与实践的一些真实经历。讲真，这阵子被业务逼着搞出来的这些玩意儿，踩的坑比我过去半年吃的盐都多。

事情得从上个月底说起。那天周五晚上，我正准备下班去楼下吃个隆江猪脚饭，产品经理老李突然在钉钉上疯狂@我。老李这人平时挺好说话，但一到年底冲KPI的时候就跟打了鸡血一样。他甩过来一个文档，说老板看了竞品，要求我们下周五必须上线一个“智能办公助手”的MVP（最小可行性产品）。

我点开文档一看，好家伙，需求写得那叫一个“综合”。不仅要能解析公司内部的几百份PDF和Word文档做知识库问答（也就是现在烂大街的AI办公场景），老李还异想天开地加了一条：支持语音输入，并且能实时把会议录音转成文字并提取摘要（AI音频场景）。

当时我真的想砸电脑。下周五上线？满打满算不到十天，这时间连需求评审都不够，更别提开发了。但吐槽归吐槽，活儿还是得干。我深吸一口气，打开Claude，先让它帮我梳理了一下技术架构，然后开始了我这半个月的地狱级技术探索。

先说AI办公这块。现在做个RAG（检索增强生成）好像特别简单，网上教程一搜一大把，LangChain调个API就完事了。但真到了企业级场景，那些玩具代码根本没法看。我们公司内部文档格式极其混乱，有扫描版的PDF，有带复杂表格的Word，还有各种PPT。

一开始我图省事，直接用LangChain自带的PDF解析器，结果跑出来的数据简直没法看，表格全乱了，段落截断也是家常便饭。召回率惨不忍睹，大模型回答的时候经常胡言乱语。老李测试的时候直接在工作群里阴阳怪气：“咱们这AI助手是不是智障啊？”

没办法，只能重构文档解析链路。我综合对比了市面上好几个开源方案，最后选了Marker来处理PDF，用Unstructured来处理其他格式。这俩玩意儿对表格和排版的还原度确实高，但代价就是解析速度慢得令人发指。为了解决这个问题，我把解析任务全部丢到了Celery里做异步处理，并引入了Redis做任务队列。

向量数据库的选型也让我纠结了半天。一开始用的Chroma，本地跑跑还行，一上K8s集群就各种莫名其妙的连接超时。后来咬咬牙上了Milvus。在K8s里部署Milvus可真是个技术活，它的组件太多了，etcd、MinIO、Pulsar，一堆StatefulSet。我当时为了调优它的索引参数，连着熬了两个通宵。

这里给大家分享一下我当时调优RAG召回率的一段核心Python代码，主要是加入了Rerank（重排序）机制，这招对提升准确率有奇效：

from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.vector_stores.milvus import MilvusVectorStore
from llama_index.postprocessor.flag_embedding_reranker import FlagEmbeddingReranker
from llama_index.core import Settings

vector_store = MilvusVectorStore(
    uri="http://milvus-proxy:19530",
    collection_name="company_knowledge_base",
    dim=1024,
    overwrite=False
)

# 配置Embedding模型，这里用的是BGE-M3
Settings.embed_model = "BAAI/bge-m3"

# 构建索引
storage_context = StorageContext.from_defaults(vector_store=vector_store)
index = VectorStoreIndex.from_documents(
    documents, 
    storage_context=storage_context
)

# 关键来了：引入FlagEmbeddingReranker进行二次排序
reranker = FlagEmbeddingReranker(
    model="BAAI/bge-reranker-v2-m3",
    top_n=5
)

# 构建查询引擎
query_engine = index.as_query_engine(
    similarity_top_k=20, # 先粗排召回20个
    node_postprocessors=[reranker] # 再用Rerank精排取前5个
)

# 执行查询
response = query_engine.query("公司最新的报销流程是什么？")
print(response)

加了Rerank之后，召回准确率直接从60%多飙升到了90%以上。老李再测试的时候，终于不骂街了，还夸我“技术牛逼”。我心里暗爽，但我知道，真正的硬骨头还在后面。

接下来就是那个让我差点崩溃的AI音频需求。老李要求开会的时候，能实时把语音转成文字，并且每隔5分钟自动生成一次会议摘要。

一开始我的思路很简单，直接部署一个Whisper ASR模型，写个WebSocket服务接收前端传来的音频流，实时推理。想法很美好，现实很骨感。Whisper这玩意儿是个显存刺客，我用了large-v3模型，单卡A10（24G显存）跑起来显存直接飙到20G以上。更致命的是，当并发会议超过3个的时候，推理延迟高达十几秒，而且服务经常因为OOM（Out of Memory）直接挂掉。

上周五晚上，也就是距离deadline还有三天的时候，线上压测直接翻车。我看着Grafana监控面板上Pod重启次数蹭蹭往上涨，当时真的想提桶跑路了。

冷静下来排查，发现根本原因有两个：一是Whisper在处理长音频流时，没有及时释放中间变量的显存，导致显存泄漏；二是音频流直接打进推理服务，没有做缓冲，瞬间的并发峰值把GPU算力争抢光了。

为了解决这个问题，我重新设计了音频处理架构。前端不再直接传长音频流，而是每隔10秒切一次片，通过Kafka发送消息。后端起一个消费者服务，用FFmpeg对音频进行降噪和重采样，然后再喂给Whisper。同时，我把Whisper的推理服务改成了批处理模式（Batching），攒够几个音频片段再一起推理，极大提高了GPU利用率。

这是当时我重写的一段音频流消费和切片的核心逻辑：

import ffmpeg
import numpy as np
import whisper
from kafka import KafkaConsumer
import json
import torch

# 加载Whisper模型，指定设备
model = whisper.load_model("large-v3", device="cuda")

# 初始化Kafka消费者
consumer = KafkaConsumer(
    'audio_stream_topic',
    bootstrap_servers=['kafka-broker-1:9092', 'kafka-broker-2:9092'],
    group_id='audio_processing_group',
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

def process_audio_chunk(audio_bytes, sample_rate=16000):
    # 使用ffmpeg进行音频解码和重采样，避免直接处理原始pcm导致的格式问题
    try:
        out, _ = (
            ffmpeg.input('pipe:', format='s16le', acodec='pcm_s16le', ar=sample_rate, ac=1)
            .output('pipe:', format='wav', ar=16000, ac=1)
            .run(capture_stdout=True, capture_stderr=True, input=audio_bytes)
        )
        # 将wav字节流转换为numpy数组
        audio_np = whisper.load_audio(out)
        audio_np = whisper.pad_or_trim(audio_np)
        # 转换为梅尔频谱
        mel = whisper.log_mel_spectrogram(audio_np).to(model.device)
        # 推理
        options = whisper.DecodingOptions(language="zh", fp16=torch.cuda.is_available())
        result = whisper.decode(model, mel, options)
        return result.text
    except Exception as e:
        print(f"Audio processing error: {e}")
        return ""

# 批量处理逻辑，提高GPU吞吐
batch_buffer = []
BATCH_SIZE = 4

for message in consumer:
    audio_data = message.value
    batch_buffer.append(audio_data['chunk'])
    
    if len(batch_buffer) >= BATCH_SIZE:
        texts = []
        for chunk in batch_buffer:
            text = process_audio_chunk(chunk)
            texts.append(text)
        
        # 将拼接后的文本发送给大模型做摘要（这里省略调用LLM的代码）
        # send_to_llm_for_summary("".join(texts))
        
        # 清空缓冲区，释放内存
        batch_buffer.clear()
        # 强制清理GPU缓存，防止显存泄漏
        torch.cuda.empty_cache()

改完这套逻辑后，再次压测，并发10个会议，延迟稳定在2秒以内，显存占用死死压在18G，再也没有OOM过。看着监控面板上平稳的曲线，我长舒了一口气，赶紧去楼下买了瓶冰可乐压压惊。

最后，作为咱们云原生和K8s的老本行，我必须得说说这套AI服务在K8s里的部署和兜底策略。搞AI应用，算力资源的管理是重中之重。我们集群里混部了CPU节点和GPU节点，怎么让Pod精准调度到GPU节点上，并且合理分配显存，是个大学问。

一开始运维小哥给我配的NVIDIA device plugin有点问题，导致Pod申请了GPU但实际用不了，排查了半天才发现是驱动版本和CUDA版本不匹配。后来我亲自上手，重写了Deployment的yaml，并且加入了HPA（Horizontal Pod Autoscaler）基于自定义指标（GPU显存使用率）的自动扩缩容。

这里分享一段我当时写的HPA配置，供大家参考：

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: whisper-asr-hpa
  namespace: ai-services
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: whisper-asr-deployment
  minReplicas: 2
  maxReplicas: 10
  metrics:
  - type: Pods
    pods:
      metric:
        name: DCGM_FI_DEV_MEM_COPY_UTIL # NVIDIA DCGM导出的显存使用率指标
      target:
        type: AverageValue
        averageValue: "70" # 当平均显存使用率达到70%时触发扩容
  behavior:
    scaleUp:
      stabilizationWindowSeconds: 60 # 扩容冷却时间，防止频繁抖动
      policies:
      - type: Pods
        value: 2
        periodSeconds: 60 # 每分钟最多扩容2个Pod
    scaleDown:
      stabilizationWindowSeconds: 300 # 缩容冷却时间，给足5分钟观察期

除了调度，监控告警也是必不可少的。我用Prometheus抓取了NVIDIA DCGM Exporter的指标，在Grafana里专门画了一个AI服务监控大盘。把GPU利用率、显存占用、推理延迟、QPS全放在一起。我还配了Alertmanager，只要GPU显存使用率连续3分钟超过90%，或者推理P99延迟超过5秒，钉钉机器人就会疯狂@我。这套监控上线后，帮我提前拦截了好几次潜在的线上事故。

历经半个月的折磨，这个“智能办公助手”终于在周五晚上准时上线了。老李在周一的早会上狠狠地表扬了研发团队，还自掏腰包请我们喝了星巴克。虽然累得够呛，但看着自己亲手敲出来的代码在业务里跑起来，那种成就感真的是写多少篇评测文章都换不来的。

回过头来看这次技术探索与实践，我最大的心得体会就是：技术永远是为业务服务的。现在AI概念满天飞，各种大模型、Agent、RAG名词层出不穷，但作为工程师，我们不能为了用技术而用技术。在选型的时候，必须“综合”考虑业务场景、团队技术栈、维护成本和上线时间。

AI办公和AI音频听起来很高大上，但剥开外衣，底层依然是文档解析、向量检索、流式处理、资源调度这些传统的工程问题。AI工具（比如我天天用的Claude）确实能帮我们快速生成代码、排查Bug，极大地提升了我们的开发效率，但它替代不了我们对系统架构的思考，替代不了我们对线上问题的敏锐嗅觉，更替代不了我们为了一个OOM熬红双眼的排查过程。

技术探索的路上没有捷径，踩过的每一个坑，看过的每一个报错日志，最终都会变成我们简历上最硬核的底气。

好了，不说了，老李又在群里发需求了，说是要加个AI生成PPT的功能。我得赶紧打开Cursor，让Claude帮我先搭个脚手架了。各位同行，咱们下期评测文章见，祝大家写的代码永不报错，发的版本永不回滚！