用JetBrains Junie打造你的AI搜索工具:从入门到实战优化

UI还原大师
2026-07-14 17:55
阅读 673

作者按:最近掘金上好多小伙伴私信问我,AI搜索到底怎么玩?JetBrains Junie又是什么神仙工具?我当初刚接触这些的时候也是一脸懵,踩了不少坑。今天这篇教程,我把自己这段时间的探索和实践优化经验,用最通俗的方式整理出来,希望能帮到同样零基础的你。


一、开篇:我们今天要聊什么

在开始动手之前,我们先搞清楚几个核心概念,别急着写代码,理解清楚再上手,效率会高很多。

1.1 什么是AI搜索

传统的搜索,比如你在百度输入一个关键词,它返回的是包含这个关键词的网页列表。但AI搜索不一样,它能理解你的意图,然后从海量信息中提炼、整合、推理,给你一个更精准、更有逻辑的答案。

举个简单的例子:

  • 传统搜索:你搜"Python怎么读取Excel",它会给你一堆链接,你得自己点进去看。
  • AI搜索:你搜同样的问题,它会直接告诉你"可以用openpyxl库,代码如下……",甚至还能根据你的上下文给出优化建议。

AI搜索的核心能力包括:

能力 说明 举例
语义理解 理解用户真正想问什么 把"咋读表格"理解为"读取Excel文件"
信息整合 从多个来源汇总信息 综合文档、代码库、网页给出答案
上下文推理 结合对话历史给出连贯回答 记住你前面说用的是Python 3.10
代码生成 直接输出可运行的代码 给出完整的读取Excel脚本

1.2 什么是JetBrains Junie

JetBrains Junie 是 JetBrains 推出的一款 AI 智能体(AI Agent),它可以直接集成到你的开发环境中(比如 IntelliJ IDEA、PyCharm、WebStorm 等)。它不是一个简单的代码补全工具,而是一个能理解你的项目上下文、执行多步骤任务、甚至自动修复 Bug 的 AI 助手。

简单来说,Junie 就像一个坐在你旁边的资深同事,你告诉它"帮我把这个接口的返回值改成驼峰命名",它就能自动分析代码、找到相关文件、修改代码、甚至跑测试验证。

1.3 为什么要把AI搜索和Junie结合起来

我当初学的时候,最大的困惑就是:AI搜索工具那么多,为什么还要和Junie扯上关系?

答案是:Junie 可以作为 AI 搜索的"执行层"

想象一下这个场景:

  1. 你通过 AI 搜索找到了一个开源的搜索框架文档
  2. 你想在项目里实践这个框架
  3. 你告诉 Junie:"根据这份文档,帮我在项目里搭建一个基础的搜索服务"
  4. Junie 自动分析你的项目结构,生成代码,配置依赖,甚至写好单元测试

这就是技术探索与实践优化的闭环:AI搜索帮你找到知识,Junie帮你把知识落地到代码里。


二、环境准备

好,概念搞清楚了,我们开始搭环境。我会尽量把每一步都写清楚,你跟着做就行。

2.1 基础环境要求

工具 版本要求 说明
Python 3.10+ 我们用它来写AI搜索的后端服务
JetBrains IDE 2024.3+ Junie 需要较新版本才支持
pip 最新版 Python包管理器
Git 任意版本 用于版本控制

2.2 安装Python环境

如果你还没装Python,按下面的步骤来:

Windows 用户:

# 1. 去 python.org 下载 Python 3.10+ 安装包
# 2. 安装时记得勾选 "Add Python to PATH"
# 3. 打开命令行验证
python --version
# 应该输出 Python 3.10.x 或更高

# 4. 升级 pip
python -m pip install --upgrade pip

macOS 用户:

# 用 Homebrew 安装
brew install python@3.10

# 验证
python3 --version

Linux 用户:

# Ubuntu/Debian
sudo apt update
sudo apt install python3.10 python3.10-venv python3-pip

# 验证
python3 --version

2.3 创建项目并配置虚拟环境

我当初学的时候,最大的教训就是一定要用虚拟环境。不然各种依赖冲突会让你怀疑人生。

# 1. 创建项目目录
mkdir ai-search-demo
cd ai-search-demo

# 2. 创建虚拟环境
python -m venv venv

# 3. 激活虚拟环境
# Windows:
venv\Scripts\activate
# macOS/Linux:
source venv/bin/activate

# 4. 激活后,命令行前面会出现 (venv) 标识
# 这时候安装的所有包都只在这个项目里生效

2.4 安装核心依赖

# 安装我们需要的核心库
pip install fastapi uvicorn openai chromadb sentence-transformers

# 各库的作用:
# fastapi     - 快速搭建API服务
# uvicorn     - ASGI服务器,用来跑FastAPI
# openai      - 调用大模型API(也可以用其他替代)
# chromadb    - 向量数据库,存储搜索索引
# sentence-transformers - 文本向量化

2.5 配置JetBrains Junie

这一步很关键,很多新手卡在这里。

第一步:确认IDE版本

打开你的 JetBrains IDE(IntelliJ IDEA / PyCharm / WebStorm 等),进入 Help → About,确认版本号 >= 2024.3。

第二步:安装Junie插件

1. 打开 Settings/Preferences
2. 进入 Plugins → Marketplace
3. 搜索 "Junie"
4. 点击 Install,安装完成后重启 IDE

第三步:配置Junie

1. 重启后,在右侧工具栏找到 Junie 图标
2. 点击打开 Junie 面板
3. 首次使用需要登录 JetBrains 账号
4. 在 Settings → Tools → Junie 中配置:
   - 选择你的默认工作模式(Code / Chat / Task)
   - 设置上下文范围(建议选 "Project")
   - 配置允许 Junie 执行的操作权限

第四步:验证Junie是否正常工作

在你的项目里随便打开一个文件,然后在 Junie 面板里输入:

帮我分析一下当前项目的目录结构

如果 Junie 能正确列出你的项目文件,说明配置成功了。

新手常见问题:Junie面板打不开怎么办?

  • 检查IDE版本是否 >= 2024.3
  • 检查网络是否能访问 JetBrains 服务
  • 尝试 File → Invalidate Caches / Restart
  • 如果还不行,去 JetBrains 官网下载最新的 IDE 版本

三、核心概念

在写代码之前,我们需要理解几个关键概念。我用最通俗的方式给你讲明白。

3.1 向量与嵌入(Embedding)

传统搜索是"关键词匹配"——你搜"苹果",它找包含"苹果"两个字的内容。

AI搜索是"语义匹配"——你搜"苹果",它知道你可能在问水果,也可能在问手机,然后根据上下文给你不同的答案。

这背后的核心技术就是向量嵌入(Embedding)

文本 → [嵌入模型] → 向量(一组数字)

例如:
"今天天气真好" → [0.12, -0.34, 0.56, ..., 0.78]  (假设768维)
"今日阳光明媚" → [0.11, -0.33, 0.55, ..., 0.77]  (语义相近,向量也相近)
"股票大跌"     → [-0.89, 0.45, -0.12, ..., 0.03]  (语义不同,向量差距大)

两个向量的距离越近,说明两段文本的语义越相似。这就是AI搜索能理解"言外之意"的原因。

3.2 RAG(检索增强生成)

RAG 是当前 AI 搜索最主流的架构,全称是 Retrieval-Augmented Generation

它的工作流程可以用下面的步骤来描述:

用户提问
    ↓
[步骤1] 将问题转为向量
    ↓
[步骤2] 在向量数据库中检索最相似的文档片段
    ↓
[步骤3] 把检索到的文档片段 + 用户问题,一起发给大模型
    ↓
[步骤4] 大模型基于这些上下文,生成最终答案
    ↓
返回给用户

为什么要用 RAG?因为大模型的知识有截止日期,而且它不知道你私有项目里的代码。通过 RAG,你可以让大模型基于最新的、特定的数据来回答问题。

3.3 AI Agent 与 Junie 的角色

在传统的 AI 应用里,模型只是"一问一答"。而 AI Agent 能自主规划、执行多步骤任务。

Junie 就是一个 AI Agent。它的能力模型可以这样理解:

                    ┌─────────────┐
                    │   用户指令   │
                    └──────┬──────┘
                           ↓
                    ┌─────────────┐
                    │  理解意图    │
                    │  拆解任务    │
                    └──────┬──────┘
                           ↓
              ┌────────────┼────────────┐
              ↓            ↓            ↓
       ┌──────────┐ ┌──────────┐ ┌──────────┐
       │ 读取文件  │ │ 修改代码  │ │ 执行命令  │
       └──────────┘ └──────────┘ └──────────┘
              ↓            ↓            ↓
              └────────────┼────────────┘
                           ↓
                    ┌─────────────┐
                    │  验证结果    │
                    │  反馈调整    │
                    └─────────────┘

在我们的实践中,Junie 会扮演"执行者"的角色——AI搜索负责找到知识和方案,Junie 负责把方案落地到你的项目代码里。


四、实战项目:搭建一个AI搜索服务

好了,概念都清楚了,我们开始写代码。这个项目会分几个阶段,循序渐进。

4.1 项目结构

我们先规划好项目的文件结构:

ai-search-demo/
├── venv/                   # 虚拟环境
├── app/
│   ├── __init__.py
│   ├── main.py             # FastAPI 入口
│   ├── indexer.py           # 文档索引模块
│   ├── searcher.py          # 搜索模块
│   ├── models.py            # 数据模型
│   └── config.py            # 配置文件
├── data/
│   └── docs/               # 存放待索引的文档
├── tests/
│   └── test_search.py      # 测试文件
├── requirements.txt         # 依赖清单
└── README.md

4.2 第一步:配置文件与数据模型

创建 app/config.py

# app/config.py
"""
配置文件:集中管理所有配置项
"""

# 向量数据库存储路径
CHROMA_PERSIST_DIR = "./chroma_db"

# 向量数据库集合名称
COLLECTION_NAME = "ai_search_docs"

# 嵌入模型名称(使用 sentence-transformers 的本地模型)
EMBEDDING_MODEL = "all-MiniLM-L6-v2"

# 搜索返回的最大结果数
TOP_K = 5

# FastAPI 服务配置
HOST = "0.0.0.0"
PORT = 8000

创建 app/models.py

# app/models.py
"""
数据模型:定义请求和响应的数据结构
"""
from pydantic import BaseModel
from typing import List, Optional

class SearchRequest(BaseModel):
    """搜索请求模型"""
    query: str                          # 用户查询文本
    top_k: Optional[int] = 5            # 返回结果数量,默认5

class SearchResult(BaseModel):
    """单条搜索结果"""
    content: str                        # 匹配的文档内容
    score: float                        # 相似度得分
    source: Optional[str] = None        # 来源文件

class SearchResponse(BaseModel):
    """搜索响应模型"""
    query: str                          # 原始查询
    results: List[SearchResult]         # 搜索结果列表
    total: int                          # 结果总数

4.3 第二步:文档索引模块

创建 app/indexer.py

# app/indexer.py
"""
索引模块:负责将文档切分、向量化并存入向量数据库
"""
import os
import chromadb
from sentence_transformers import SentenceTransformer
from app.config import CHROMA_PERSIST_DIR, COLLECTION_NAME, EMBEDDING_MODEL

class DocumentIndexer:
    """文档索引器"""

    def __init__(self):
        # 初始化嵌入模型(首次运行会自动下载模型)
        print("正在加载嵌入模型,首次可能需要几分钟...")
        self.model = SentenceTransformer(EMBEDDING_MODEL)
        print("嵌入模型加载完成!")

        # 初始化向量数据库客户端
        self.client = chromadb.PersistentClient(path=CHROMA_PERSIST_DIR)
        self.collection = self.client.get_or_create_collection(
            name=COLLECTION_NAME,
            metadata={"hnsw:space": "cosine"}  # 使用余弦相似度
        )

    def chunk_text(self, text: str, chunk_size: int = 200, overlap: int = 50) -> list:
        """
        将长文本切分成小块
        
        参数:
            text: 原始文本
            chunk_size: 每块的最大字符数
            overlap: 相邻块之间的重叠字符数
        
        返回:
            文本块列表
        """
        chunks = []
        start = 0
        while start < len(text):
            end = start + chunk_size
            chunk = text[start:end]
            if chunk.strip():  # 跳过空块
                chunks.append(chunk.strip())
            start += chunk_size - overlap
        return chunks

    def index_file(self, file_path: str) -> int:
        """
        索引单个文件
        
        参数:
            file_path: 文件路径
        
        返回:
            成功索引的文本块数量
        """
        with open(file_path, "r", encoding="utf-8") as f:
            content = f.read()

        # 切分文本
        chunks = self.chunk_text(content)
        if not chunks:
            return 0

        # 生成向量
        embeddings = self.model.encode(chunks).tolist()

        # 生成唯一ID
        ids = [f"{file_path}_chunk_{i}" for i in range(len(chunks))]

        # 存入向量数据库
        self.collection.add(
            ids=ids,
            embeddings=embeddings,
            documents=chunks,
            metadatas=[{"source": file_path}] * len(chunks)
        )

        return len(chunks)

    def index_directory(self, dir_path: str) -> dict:
        """
        索引整个目录下的所有 .txt 和 .md 文件
        
        参数:
            dir_path: 目录路径
        
        返回:
            索引统计信息
        """
        stats = {"files_processed": 0, "total_chunks": 0, "errors": []}

        for root, _, files in os.walk(dir_path):
            for file_name in files:
                if file_name.endswith((".txt", ".md")):
                    file_path = os.path.join(root, file_name)
                    try:
                        chunk_count = self.index_file(file_path)
                        stats["files_processed"] += 1
                        stats["total_chunks"] += chunk_count
                        print(f"✓ 已索引: {file_path} ({chunk_count} 个文本块)")
                    except Exception as e:
                        stats["errors"].append(f"{file_path}: {str(e)}")
                        print(f"✗ 索引失败: {file_path} - {str(e)}")

        return stats

4.4 第三步:搜索模块

创建 app/searcher.py

# app/searcher.py
"""
搜索模块:负责接收查询、检索相似文档并返回结果
"""
import chromadb
from sentence_transformers import SentenceTransformer
from app.config import CHROMA_PERSIST_DIR, COLLECTION_NAME, EMBEDDING_MODEL, TOP_K
from app.models import SearchResult

class Searcher:
    """AI搜索引擎"""

    def __init__(self):
        # 加载嵌入模型
        self.model = SentenceTransformer(EMBEDDING_MODEL)

        # 连接向量数据库
        self.client = chromadb.PersistentClient(path=CHROMA_PERSIST_DIR)
        self.collection = self.client.get_collection(name=COLLECTION_NAME)

    def search(self, query: str, top_k: int = TOP_K) -> list:
        """
        执行语义搜索
        
        参数:
            query: 用户查询文本
            top_k: 返回结果数量
        
        返回:
            SearchResult 列表
        """
        # 将查询文本转为向量
        query_embedding = self.model.encode([query]).tolist()

        # 在向量数据库中检索
        results = self.collection.query(
            query_embeddings=query_embedding,
            n_results=top_k
        )

        # 组装返回结果
        search_results = []
        if results and results["documents"]:
            for i in range(len(results["documents"][0])):
                content = results["documents"][0][i]
                # chromadb 返回的距离越小越相似,转换为相似度得分
                distance = results["distances"][0][i]
                score = 1.0 - distance  # 余弦距离转相似度
                source = results["metadatas"][0][i].get("source", "unknown")

                search_results.append(
                    SearchResult(
                        content=content,
                        score=round(score, 4),
                        source=source
                    )
                )

        return search_results

    def get_collection_stats(self) -> dict:
        """获取索引库统计信息"""
        return {
            "total_documents": self.collection.count(),
            "collection_name": COLLECTION_NAME
        }

4.5 第四步:FastAPI 服务入口

创建 app/main.py

# app/main.py
"""
FastAPI 服务入口:提供 HTTP 接口
"""
from fastapi import FastAPI, HTTPException
from app.models import SearchRequest, SearchResponse
from app.indexer import DocumentIndexer
from app.searcher import Searcher
from app.config import HOST, PORT

# 创建 FastAPI 应用
app = FastAPI(
    title="AI Search Demo",
    description="一个基于向量检索的AI搜索服务",
    version="1.0.0"
)

# 初始化全局实例
indexer = DocumentIndexer()
searcher = Searcher()

@app.get("/")
async def root():
    """健康检查接口"""
    return {
        "status": "running",
        "service": "AI Search Demo",
        "docs": "/docs"
    }

@app.post("/search", response_model=SearchResponse)
async def search(request: SearchRequest):
    """
    搜索接口
    
    接收用户查询,返回语义最相关的文档片段
    """
    if not request.query.strip():
        raise HTTPException(status_code=400, detail="查询内容不能为空")

    results = searcher.search(
        query=request.query,
        top_k=request.top_k or TOP_K
    )

    return SearchResponse(
        query=request.query,
        results=results,
        total=len(results)
    )

@app.post("/index")
async def index_documents(dir_path: str = "./data/docs"):
    """
    索引接口
    
    扫描指定目录下的文档并建立索引
    """
    import os
    if not os.path.exists(dir_path):
        raise HTTPException(status_code=404, detail=f"目录不存在: {dir_path}")

    stats = indexer.index_directory(dir_path)
    return {
        "message": "索引完成",
        "stats": stats
    }

@app.get("/stats")
async def stats():
    """获取索引库统计信息"""
    return searcher.get_collection_stats()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host=HOST, port=PORT)

4.6 第五步:准备测试数据

创建一些测试文档,放在 data/docs/ 目录下:

创建 data/docs/python_basics.md

# Python 基础入门

Python 是一种高级编程语言,以简洁易读著称。它支持多种编程范式,
包括面向对象、函数式和过程式编程。

## 变量与数据类型

Python 中的变量不需要声明类型,赋值即创建。常见的数据类型包括:
- int:整数类型,如 42
- float:浮点数类型,如 3.14
- str:字符串类型,如 "hello"
- list:列表类型,如 [1, 2, 3]
- dict:字典类型,如 {"name": "Alice"}

## 控制流

Python 使用缩进来表示代码块,而不是花括号。
主要的控制流语句包括 if/elif/else、for 循环和 while 循环。

## 函数

使用 def 关键字定义函数。Python 支持默认参数、可变参数和关键字参数。
函数是一等公民,可以赋值给变量、作为参数传递。

创建 data/docs/fastapi_guide.md

# FastAPI 快速指南

FastAPI 是一个用于构建 API 的现代、快速(高性能)的 Web 框架,
使用 Python 3.7+ 并基于标准的 Python 类型提示。

## 特性

- 快速:与 NodeJS 和 Go 相当的性能
- 快速编码:将开发速度提高 200% 到 300%
- 更少的 Bug:减少约 40% 的人为错误
- 直观:编辑器支持完善,补全无处不在
- 简单:设计为易于使用和学习
- 短:最小化代码重复
- 健壮:生产可用的代码
- 基于标准:基于 OpenAPI 和 JSON Schema

## 安装

pip install fastapi
pip install uvicorn[standard]

## 第一个应用

from fastapi import FastAPI
app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

运行:uvicorn main:app --reload

创建 data/docs/ai_search_intro.md

# AI 搜索技术简介

AI 搜索是利用人工智能技术改进信息检索的过程。与传统基于关键词的搜索不同,
AI 搜索能够理解查询的语义和上下文。

## 核心技术

### 向量嵌入
将文本转换为高维向量空间中的点,语义相似的文本在向量空间中距离更近。
常用的嵌入模型包括 OpenAI 的 text-embedding-ada-002、
Sentence-BERT 等。

### 向量数据库
专门用于存储和检索向量的数据库。常见的有:
- ChromaDB:轻量级,适合原型开发
- Pinecone:云端托管,适合生产环境
- Milvus:开源,支持大规模数据
- Weaviate:支持混合搜索

### RAG(检索增强生成)
将检索系统与生成模型结合,先从知识库中检索相关文档,
再让大模型基于检索结果生成回答。这种方式可以让模型利用最新数据,
减少幻觉,提高回答的准确性。

## 应用场景

- 企业知识库搜索
- 代码库智能搜索
- 客服问答系统
- 学术研究文献检索

4.7 第六步:运行与测试

# 1. 启动服务
python -m app.main

# 你应该看到类似输出:
# INFO:     Uvicorn running on http://0.0.0.0:8000

打开浏览器访问 http://localhost:8000/docs,你会看到 FastAPI 自动生成的 API 文档页面。

测试索引功能:

在 API 文档页面找到 /index 接口,点击 "Try it out",然后 "Execute"。或者用命令行:

curl -X POST "http://localhost:8000/index?dir_path=./data/docs"

你应该看到类似输出:

{
  "message": "索引完成",
  "stats": {
    "files_processed": 3,
    "total_chunks": 18,
    "errors": []
  }
}

测试搜索功能:

curl -X POST "http://localhost:8000/search" \
  -H "Content-Type: application/json" \
  -d '{"query": "Python怎么定义函数", "top_k": 3}'

你应该能看到返回了 python_basics.md 中关于函数定义的内容,即使你的查询用的是中文,而文档是英文术语混合中文——这就是语义搜索的魅力。


五、用JetBrains Junie优化实践

项目跑起来了,但这只是起点。接下来我们看看怎么用 Junie 来优化这个项目。

5.1 让Junie帮你写单元测试

在 Junie 面板中输入:

请为 app/searcher.py 中的 Searcher 类编写单元测试,
使用 pytest 框架,mock 掉向量数据库的依赖。
测试文件放在 tests/test_search.py

Junie 会自动分析 Searcher 类的代码,然后生成类似这样的测试:

# tests/test_search.py
"""
Searcher 模块的单元测试
"""
import pytest
from unittest.mock import MagicMock, patch
from app.searcher import Searcher
from app.models import SearchResult

class TestSearcher:
    """Searcher 测试类"""

    @patch("app.searcher.SentenceTransformer")
    @patch("app.searcher.chromadb.PersistentClient")
    def setup_method(self, method, mock_client, mock_model_class):
        """每个测试方法前的初始化"""
        # 配置 mock
        self.mock_model = MagicMock()
        mock_model_class.return_value = self.mock_model

        self.mock_collection = MagicMock()
        self.mock_client_instance = MagicMock()
        self.mock_client_instance.get_collection.return_value = self.mock_collection
        mock_client.return_value = self.mock_client_instance

        self.searcher = Searcher()

    def test_search_returns_results(self):
        """测试搜索能正常返回结果"""
        # 准备 mock 数据
        self.mock_model.encode.return_value = [[0.1, 0.2, 0.3]]
        self.mock_collection.query.return_value = {
            "documents": [["Python是一种编程语言", "FastAPI是一个Web框架"]],
            "distances": [[0.15, 0.35]],
            "metadatas": [
                {"source": "python.md"},
                {"source": "fastapi.md"}
            ]
        }

        # 执行搜索
        results = self.searcher.search("什么是Python", top_k=2)

        # 验证结果
        assert len(results) == 2
        assert isinstance(results[0], SearchResult)
        assert results[0].source == "python.md"
        assert results[0].score > results[1].score  # 第一个结果相似度更高

    def test_search_empty_query(self):
        """测试空结果的情况"""
        self.mock_model.encode.return_value = [[0.1, 0.2, 0.3]]
        self.mock_collection.query.return_value = {
            "documents": [[]],
            "distances": [[]],
            "metadatas": [[]]
        }

        results = self.searcher.search("不存在的内容")
        assert len(results) == 0

    def test_get_collection_stats(self):
        """测试获取统计信息"""
        self.mock_collection.count.return_value = 42

        stats = self.searcher.get_collection_stats()
        assert stats["total_documents"] == 42

运行测试:

pip install pytest
pytest tests/ -v

5.2 让Junie帮你做性能优化

在 Junie 面板中输入:

分析 app/indexer.py 的性能瓶颈,
特别是 chunk_text 方法和 index_directory 方法,
给出优化建议并直接修改代码。
主要优化方向:
1. 大批量文本的切分效率
2. 多文件索引的并发处理
3. 向量编码的批处理优化

Junie 可能会给出这样的优化版本:

# app/indexer.py 优化版本(核心改动部分)

import concurrent.futures
from typing import List

class DocumentIndexer:
    # ... 前面的代码不变 ...

    def chunk_text(self, text: str, chunk_size: int = 200, overlap: int = 50) -> list:
        """优化版:使用更高效的字符串切片"""
        if not text or not text.strip():
            return []

        # 预分配列表,减少动态扩容开销
        estimated_chunks = (len(text) // (chunk_size - overlap)) + 1
        chunks = []

        # 使用 range 步进来替代 while 循环
        for start in range(0, len(text), chunk_size - overlap):
            chunk = text[start:start + chunk_size]
            if chunk.strip():
                chunks.append(chunk.strip())
            if start + chunk_size >= len(text):
                break

        return chunks

    def _encode_batch(self, texts: List[str], batch_size: int = 32) -> list:
        """批处理向量编码,提高GPU利用率"""
        all_embeddings = []
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]
            batch_embeddings = self.model.encode(batch).tolist()
            all_embeddings.extend(batch_embeddings)
        return all_embeddings

    def index_directory(self, dir_path: str, max_workers: int = 4) -> dict:
        """优化版:使用线程池并发处理文件读取"""
        stats = {"files_processed": 0, "total_chunks": 0, "errors": []}

        # 先收集所有需要处理的文件
        file_paths = []
        for root, _, files in os.walk(dir_path):
            for file_name in files:
                if file_name.endswith((".txt", ".md")):
                    file_paths.append(os.path.join(root, file_name))

        # 并发读取文件内容
        def read_file(file_path):
            with open(file_path, "r", encoding="utf-8") as f:
                return file_path, f.read()

        file_contents = []
        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {executor.submit(read_file, fp): fp for fp in file_paths}
            for future in concurrent.futures.as_completed(futures):
                try:
                    file_contents.append(future.result())
                except Exception as e:
                    fp = futures[future]
                    stats["errors"].append(f"{fp}: {str(e)}")

        # 批量处理所有文本块
        all_chunks = []
        all_metadatas = []
        for file_path, content in file_contents:
            chunks = self.chunk_text(content)
            for chunk in chunks:
                all_chunks.append(chunk)
                all_metadatas.append({"source": file_path})
            stats["files_processed"] += 1
            stats["total_chunks"] += len(chunks)

        # 批量编码和存储
        if all_chunks:
            embeddings = self._encode_batch(all_chunks)
            ids = [f"chunk_{i}" for i in range(len(all_chunks))]

            # 分批存入,避免单次数据量过大
            batch_size = 100
            for i in range(0, len(all_chunks), batch_size):
                end = min(i + batch_size, len(all_chunks))
                self.collection.add(
                    ids=ids[i:end],
                    embeddings=embeddings[i:end],
                    documents=all_chunks[i:end],
                    metadatas=all_metadatas[i:end]
                )

        return stats

5.3 让Junie帮你添加日志和错误处理

给整个项目添加完善的日志记录,使用 Python 的 logging 模块。
要求:
1. 不同级别用不同格式
2. 同时输出到控制台和文件
3. 关键操作要有日志记录
4. 添加全局异常处理中间件

Junie 会帮你生成一个 app/logger.py

# app/logger.py
"""
统一日志配置
"""
import logging
import sys
from logging.handlers import RotatingFileHandler

def setup_logger(name: str = "ai_search") -> logging.Logger:
    """配置并返回 logger 实例"""
    logger = logging.getLogger(name)
    logger.setLevel(logging.DEBUG)

    # 避免重复添加 handler
    if logger.handlers:
        return logger

    # 控制台 handler
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setLevel(logging.INFO)
    console_format = logging.Formatter(
        "[%(asctime)s] %(levelname)-8s %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S"
    )
    console_handler.setFormatter(console_format)

    # 文件 handler(自动轮转,最大5MB,保留3个备份)
    file_handler = RotatingFileHandler(
        "ai_search.log",
        maxBytes=5 * 1024 * 1024,
        backupCount=3,
        encoding="utf-8"
    )
    file_handler.setLevel(logging.DEBUG)
    file_format = logging.Formatter(
        "[%(asctime)s] %(levelname)-8s [%(filename)s:%(lineno)d] %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S"
    )
    file_handler.setFormatter(file_format)

    logger.addHandler(console_handler)
    logger.addHandler(file_handler)

    return logger

# 全局 logger 实例
logger = setup_logger()

然后在 app/main.py 中添加全局异常处理:

# app/main.py 中添加

from fastapi import Request
from fastapi.responses import JSONResponse
from app.logger import logger

@app.middleware("http")
async def log_requests(request: Request, call_next):
    """请求日志中间件"""
    logger.info(f"→ {request.method} {request.url.path}")
    response = await call_next(request)
    logger.info(f"← {response.status_code} {request.url.path}")
    return response

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    """全局异常处理"""
    logger.error(f"未处理的异常: {str(exc)}", exc_info=True)
    return JSONResponse(
        status_code=500,
        content={"detail": "服务器内部错误,请稍后重试"}
    )

六、常见问题与避坑指南

我当初学的时候踩过的坑,希望你不要再踩。

6.1 嵌入模型下载失败

现象:运行时报错 ConnectionError 或模型下载超时。

原因sentence-transformers 默认从 HuggingFace 下载模型,国内网络可能不通。

解决方案

# 方法1:设置 HuggingFace 镜像
import os
os.environ["HF_ENDPOINT"] = "https://hf-mirror.com"

# 方法2:手动下载模型
# 1. 在 https://hf-mirror.com/sentence-transformers/all-MiniLM-L6-v2 下载模型文件
# 2. 放到本地目录,比如 ./models/all-MiniLM-L6-v2
# 3. 修改配置
EMBEDDING_MODEL = "./models/all-MiniLM-L6-v2"  # 使用本地路径

6.2 搜索结果不准确

现象:搜索返回的内容和查询没什么关系。

排查清单

检查项 说明 解决方法
文档是否已索引 可能忘了执行索引 调用 /index 接口
文本块切分是否合理 块太大或太小都影响效果 调整 chunk_size(建议150-300)
嵌入模型是否匹配 中英文场景用不同模型 中文场景换用 paraphrase-multilingual-MiniLM-L12-v2
数据是否干净 文档里有大量无关内容 预处理文档,去除噪声

6.3 Junie 生成的代码跑不起来

现象:Junie 生成的代码有语法错误或逻辑问题。

解决方案

  1. 给 Junie 更多上下文:不要只说"帮我写个函数",而是说"帮我写一个函数,输入是xxx,输出是xxx,需要处理xxx异常情况"
  2. 分步骤来:复杂任务拆成小步骤,每步验证一下
  3. 明确技术栈:告诉 Junie 你用的 Python 版本、框架版本
  4. 让它自检:生成代码后,追加一句"请检查这段代码有没有明显的bug"

6.4 向量数据库占用空间过大

现象chroma_db 目录越来越大。

优化建议

# 1. 定期清理过期数据
# 在 searcher.py 中添加
def cleanup_old_documents(self, max_age_days: int = 30):
    """清理超过指定天数的文档"""
    import time
    cutoff_time = time.time() - (max_age_days * 86400)
    # 根据 metadata 中的时间戳过滤并删除
    # ...

# 2. 使用更小的嵌入模型
# all-MiniLM-L6-v2 输出 384 维向量
# 如果精度要求不高,可以换用更小的模型

# 3. 设置 ChromaDB 的数据压缩
# 在创建 collection 时指定压缩配置

七、学习建议与下一步

恭喜你完成了第一个 AI 搜索项目!但这只是起点,下面是我建议的学习路径:

7.1 短期(1-2周)

  • 给项目添加更多文档类型支持(PDF、Word、HTML)
  • 尝试接入大模型(如 OpenAI API),实现 RAG 问答
  • 学习 Docker,把项目容器化部署

7.2 中期(1-2月)

  • 研究混合搜索(向量搜索 + 关键词搜索)
  • 学习评估搜索质量(Recall、MRR、NDCG 等指标)
  • 尝试用 Junie 重构整个项目,体验 AI 辅助编程的效率提升

7.3 长期(3-6月)

  • 深入学习向量数据库原理(HNSW、IVF 等索引算法)
  • 研究 Fine-tuning 嵌入模型,提升特定领域的搜索效果
  • 构建完整的 AI 搜索产品,包括前端界面、权限管理、监控告警

7.4 推荐资源

资源 类型 说明
LangChain 官方文档 文档 RAG 应用开发框架
ChromaDB 文档 文档 向量数据库使用指南
JetBrains Junie 官方博客 博客 了解 Junie 最新功能
《动手学深度学习》 书籍 理解嵌入模型原理
HuggingFace 课程 课程 NLP 和嵌入模型实战

写在最后

回顾一下我们今天做了什么:

  1. 理解了 AI 搜索和 JetBrains Junie 的核心概念
  2. 搭建了完整的开发环境
  3. 从零实现了一个基于向量检索的 AI 搜索服务
  4. 用 Junie 进行了代码优化、测试编写和日志完善
  5. 总结了常见问题和学习路径

我当初学这些技术的时候,最大的感受就是:不要怕动手,不要怕犯错。AI 工具的意义就在于降低门槛、加速迭代。Junie 这样的工具不是要取代你,而是让你能把精力集中在更有创造力的事情上。

希望这篇教程对你有帮助。如果遇到问题,欢迎在评论区交流,我看到都会回复。

我们下篇见!

评论 0

最热最新
暂无评论
UI还原大师Lv.1
0
影响力
0
文章
0
粉丝