用JetBrains Junie打造你的AI搜索工具:从入门到实战优化
作者按:最近掘金上好多小伙伴私信问我,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 搜索的"执行层"。
想象一下这个场景:
- 你通过 AI 搜索找到了一个开源的搜索框架文档
- 你想在项目里实践这个框架
- 你告诉 Junie:"根据这份文档,帮我在项目里搭建一个基础的搜索服务"
- 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 生成的代码有语法错误或逻辑问题。
解决方案:
- 给 Junie 更多上下文:不要只说"帮我写个函数",而是说"帮我写一个函数,输入是xxx,输出是xxx,需要处理xxx异常情况"
- 分步骤来:复杂任务拆成小步骤,每步验证一下
- 明确技术栈:告诉 Junie 你用的 Python 版本、框架版本
- 让它自检:生成代码后,追加一句"请检查这段代码有没有明显的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 和嵌入模型实战 |
写在最后
回顾一下我们今天做了什么:
- 理解了 AI 搜索和 JetBrains Junie 的核心概念
- 搭建了完整的开发环境
- 从零实现了一个基于向量检索的 AI 搜索服务
- 用 Junie 进行了代码优化、测试编写和日志完善
- 总结了常见问题和学习路径
我当初学这些技术的时候,最大的感受就是:不要怕动手,不要怕犯错。AI 工具的意义就在于降低门槛、加速迭代。Junie 这样的工具不是要取代你,而是让你能把精力集中在更有创造力的事情上。
希望这篇教程对你有帮助。如果遇到问题,欢迎在评论区交流,我看到都会回复。
我们下篇见!


评论 0