本地模型部署与知识库搭建

一、准备环境与模型拉取

1. 安装 Ollama

$ brew install ollama
# 启动ollama
$ brew services start ollama

这一步可能会顺便更新 Homebrew，所以时间可能比较久。

执行完成后，可以通过 ollama list 查看已下载的模型。

2. 配置ollama镜像源

在终端执行以下命令，创建 Ollama 的配置文件：

# 创建.ollama目录
$ mkdir -p ~/.ollama
# 写入配置
$ cat > ~/.ollama/config.json << 'EOF'
{
    "registry": {
        "mirrors": {
            "registry.ollama.ai": "https://ollama.modelscope.cn"
        }
    }
}
EOF

镜像源

地址

阿里云

https://registry.ollama.ai（上面的配置已使用此地址）

魔搭社区

https://ollama.modelscope.cn

DeepSeek官方

https://ollama.deepseek.com

建议先用阿里云（已配置），如果速度不理想，后续可以替换为其他地址。

# 然后重新启动 Ollama
$ brew services restart ollama

3. 拉取模型

借助ollama拉取deepseek免费模型，后期可在配置文件 config.yaml 替换为其他模型

$ ollama pull deepseek-coder-v2:16b

模型大小：8.9GB（Q4_0 量化） 预计耗时：10–20 分钟（取决于镜像源速度）

Ollama 支持断点续传，可以重新执行命令继续下载，其他模型可以在 https://ollama.com/library 中寻找。

二、Python版本与依赖安装

1. 安装[email protected]版本

先判断是否有conda，通过：

$ conda --version

查看是否有返回类似 conda 25.11.1 的版本号。如果没返回版本号，自行上网寻找。

然后执行：

# 使用conda创建Python 3.11环境
$ conda create -n rag python=3.11 -y
# 激活该环境，会进入rag环境
$ conda activate rag
# 检查python版本
(rag) $ python --version
Python 3.11.15

1、 activate后想退出rag环境，可使用：conda deactivate 。
2、 rag环境中可通过 which python 得到python在conda中的安装位置。

2. 依赖安装

现在确保你处于 (rag) 环境下，然后执行：

(rag) $ pip install langchain chromadb sentence-transformers fastapi uvicorn pypdf python-multipart langchain-community langchain-huggingface

依赖说明：

包名

作用

langchain

构建 RAG 流程的框架，负责加载文档、切分文本、管理检索链

chromadb

向量数据库，存储知识库的嵌入向量，支持本地持久化和相似度检索

sentence-transformers

生成文本的向量嵌入，将文档和查询转换为语义向量

fastapi

提供 REST API 服务，供 VSCode 插件调用，接收代码并返回审查结果

uvicorn

ASGI 服务器，用于运行 FastAPI 应用

pypdf

解析 PDF 文件，提取其中的文本内容，供知识库学习

python-multipart

处理 API 中接收到的 multipart/form-data 数据（如上传代码文件）

langchain-community

提供 LangChain 的社区维护集成，包括各种文档加载器（如 TextLoader、PyPDFLoader、DirectoryLoader）、向量存储、工具等。我们用它来加载知识库文件。

langchain-huggingface

提供 HuggingFace 模型的 LangChain 集成，包括嵌入模型（HuggingFaceEmbeddings）和对话模型。我们用它来生成文本向量。

这些包共同构成了 RAG 服务的核心功能。

如果出现安装失败，尝试： pip install --upgrade pip setuptools wheel # 升级 pip 和安装工具
pip cache purge # 清缓存
然后再安装一次，如果还不行，就每个包都单独安装，具体哪个包错误再查问题。

三、配置文件

现在用户目录下创建 ~/ai-code-review 文件夹，用于存放所有相关代码和配置文件。

$ mkdir -p ~/ai-code-review
$ cd ~/ai-code-review
$ touch config.yaml

然后编辑 config.yaml：

# 模型配置
model:
  name: "deepseek-coder-v2:16b"          # Ollama 模型名称
  temperature: 0.1                       # 生成温度（越低越保守）
  max_tokens: 2048                       # 最大输出 token 数

# 嵌入模型配置
embedding:
  model_name: "BAAI/bge-small-en-v1.5"   # 嵌入模型（用于向量化）
  device: "mps"                          # 使用 Mac GPU 加速（cpu 或 mps）

# 知识库配置
knowledge_base:
  root_dir: "/Users/你的用户名/company_knowledge"  # 学习资料存放根目录（请修改为实际路径）
  chunk_size: 500                         # 文本切分大小（字符数）
  chunk_overlap: 50                       # 切分重叠大小
  chroma_persist_dir: "./chroma_db"       # 向量数据库持久化目录

# API 服务配置
api:
  host: "0.0.0.0"                         # 监听所有内网 IP
  port: 8000
  ollama_url: "http://localhost:11434"    # Ollama 服务地址

保存退出后，通过以下指令验证 YAML 格式是否正确：

$ python -c "import yaml; yaml.safe_load(open('config.yaml'))"

如果没有报错，说明格式正确。

四、配置RAG并初次构建知识库

在 ~/ai-code-review/ 目录下执行：

bash

$ touch rag.py

然后将以下代码完整复制到 rag.py 中：

import os
import yaml
import json
import fnmatch
from pathlib import Path
from typing import List, Optional, Dict, Any

from langchain_community.document_loaders import (
    TextLoader,
    PyPDFLoader,
    DirectoryLoader,
)
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_core.documents import Document

# ==================== 配置加载 ====================
with open("config.yaml", "r", encoding="utf-8") as f:
    config = yaml.safe_load(f)

# ==================== 初始化嵌入模型 ====================
embedding_model_name = config["embedding"]["model_name"]
embedding_device = config["embedding"]["device"]
embeddings = HuggingFaceEmbeddings(
    model_name=embedding_model_name,
    model_kwargs={"device": embedding_device},
    encode_kwargs={"normalize_embeddings": True},
)

# ==================== 初始化向量库 ====================
persist_dir = config["knowledge_base"]["chroma_persist_dir"]
vectorstore = Chroma(
    persist_directory=persist_dir,
    embedding_function=embeddings,
)

# ==================== 文件状态管理 ====================
FILE_STATE_PATH = "file_state.json"  # 存储每个文件的路径和最后修改时间

def _load_file_state() -> Dict[str, float]:
    """加载文件状态索引，返回 {文件路径: 修改时间}"""
    if os.path.exists(FILE_STATE_PATH):
        with open(FILE_STATE_PATH, "r", encoding="utf-8") as f:
            return json.load(f)
    return {}

def _save_file_state(state: Dict[str, float]) -> None:
    """保存文件状态索引"""
    with open(FILE_STATE_PATH, "w", encoding="utf-8") as f:
        json.dump(state, f, indent=2, ensure_ascii=False)

def _get_file_mtime(file_path: str) -> float:
    """获取文件的最后修改时间戳"""
    return os.path.getmtime(file_path)

# ==================== 多源配置加载 ====================
SOURCES_FILE = "knowledge_sources.json"  # 知识库源配置文件

def _load_sources() -> List[Dict[str, Any]]:
    """加载知识库源配置，格式见文档"""
    if not os.path.exists(SOURCES_FILE):
        print(f"警告: 未找到 {SOURCES_FILE}，将使用旧版单一目录模式")
        return []
    with open(SOURCES_FILE, "r", encoding="utf-8") as f:
        data = json.load(f)
        return data.get("sources", [])

def _should_include_file(file_path: str, source_config: Dict[str, Any]) -> bool:
    """
    判断文件是否应被纳入知识库
    规则：
    - 如果 source_config 包含 include_extensions，则文件扩展名必须在其中
    - 如果 source_config 包含 exclude_dirs，则文件路径中的任何部分不能匹配这些目录名
    - 如果 source_config 包含 exclude_files，则文件名不能匹配任意通配符模式
    """
    # 扩展名检查
    ext = os.path.splitext(file_path)[1].lower()
    allowed_exts = source_config.get("include_extensions", [])
    if allowed_exts and ext not in allowed_exts:
        return False

    # 排除目录检查（精确匹配路径中的目录名）
    exclude_dirs = source_config.get("exclude_dirs", [])
    path_parts = Path(file_path).parts
    for ex_dir in exclude_dirs:
        if ex_dir in path_parts:
            return False

    # 排除文件名模式检查（支持通配符）
    exclude_files = source_config.get("exclude_files", [])
    base_name = os.path.basename(file_path)
    for pattern in exclude_files:
        if fnmatch.fnmatch(base_name, pattern):
            return False

    return True

def _load_documents_from_file(file_path: str) -> List[Document]:
    """
    根据文件扩展名加载单个文件，返回 Document 列表（每个文件可能产生多个 Document 块，但此处只加载原始文档，不切分）
    加载后自动为每个 Document 添加 metadata['source'] = file_path，方便后续删除。
    """
    ext = os.path.splitext(file_path)[1].lower()
    try:
        if ext == ".pdf":
            loader = PyPDFLoader(file_path)
        else:
            # 尝试 utf-8，若失败可回退到其他编码（这里简单处理）
            loader = TextLoader(file_path, encoding="utf-8")
        docs = loader.load()
        # 为每个文档添加 source 元数据
        for doc in docs:
            doc.metadata["source"] = file_path
        return docs
    except Exception as e:
        print(f"加载文件 {file_path} 失败: {e}")
        return []

# ==================== 知识库更新（增量） ====================
def update_knowledge_base():
    """
    根据 knowledge_sources.json 中的配置增量更新知识库。
    检测新增、修改、删除的文件，并同步到向量库。
    """
    sources = _load_sources()
    if not sources:
        print("未找到任何知识库源，请创建 knowledge_sources.json")
        return

    file_state = _load_file_state()
    new_file_state = {}      # 本次更新后的状态（新路径 -> 新修改时间）
    all_files_processed = set()

    # 用于记录哪些文件被更新了（以便后续删除不再存在的文件）
    updated_files = set()

    for src in sources:
        print(f"处理源: {src['name']} ({src['path']})")
        root_dir = src["path"]
        if not os.path.exists(root_dir):
            print(f"  路径不存在，跳过")
            continue

        # 递归遍历目录
        for root, dirs, files in os.walk(root_dir):
            # 跳过排除目录（精确匹配）
            exclude_dirs = src.get("exclude_dirs", [])
            dirs[:] = [d for d in dirs if d not in exclude_dirs]

            for file in files:
                file_path = os.path.join(root, file)
                if not _should_include_file(file_path, src):
                    continue

                all_files_processed.add(file_path)
                current_mtime = _get_file_mtime(file_path)
                old_mtime = file_state.get(file_path, 0)

                # 文件未变化，直接保留原状态
                if current_mtime == old_mtime:
                    new_file_state[file_path] = current_mtime
                    continue

                # 文件有变化（新增或修改）
                print(f"  更新文件: {file_path}")
                updated_files.add(file_path)

                # 1. 删除该文件在向量库中的所有旧块（如果存在）
                #    通过元数据 source 字段定位
                try:
                    # 获取该 source 的所有文档的 ids
                    result = vectorstore.get(where={"source": file_path})
                    ids_to_delete = result["ids"]
                    if ids_to_delete:
                        vectorstore.delete(ids=ids_to_delete)
                        vectorstore.persist()
                except Exception as e:
                    print(f"    删除旧块失败: {e}")

                # 2. 加载新文档并切分
                docs = _load_documents_from_file(file_path)
                if not docs:
                    # 加载失败，但文件存在，不更新状态（保留原状？这里跳过并记录）
                    print(f"    加载失败，跳过该文件")
                    # 注意：如果加载失败，我们不应更新状态，但原状态可能已删除，需谨慎。
                    # 简单处理：如果文件存在但加载失败，保持旧状态（如果旧状态存在）？
                    # 更合理：保留旧状态或清空？此处我们选择保留旧状态（不更新 new_file_state）
                    # 但需要避免将该文件标记为已处理，因此下面不写入 new_file_state
                    continue

                # 切分
                text_splitter = RecursiveCharacterTextSplitter(
                    chunk_size=config["knowledge_base"]["chunk_size"],
                    chunk_overlap=config["knowledge_base"]["chunk_overlap"],
                    separators=["\n\n", "\n", " ", ""],
                )
                chunks = text_splitter.split_documents(docs)

                # 3. 添加新块到向量库
                if chunks:
                    vectorstore.add_documents(chunks)
                    vectorstore.persist()
                    print(f"    添加了 {len(chunks)} 个文档块")

                # 更新状态（只有成功加载并添加后才更新）
                new_file_state[file_path] = current_mtime

    # 处理文件删除：那些在旧状态中存在但本次遍历未出现的文件，说明已被删除
    for old_path in file_state:
        if old_path not in all_files_processed:
            print(f"检测到文件已删除: {old_path}")
            try:
                result = vectorstore.get(where={"source": old_path})
                ids_to_delete = result["ids"]
                if ids_to_delete:
                    vectorstore.delete(ids=ids_to_delete)
                    vectorstore.persist()
            except Exception as e:
                print(f"  删除向量块失败: {e}")
            # 不写入 new_file_state，相当于从状态中移除

    # 保存新的文件状态
    _save_file_state(new_file_state)
    print("知识库增量更新完成")

# ==================== 原有函数（保持兼容） ====================
def load_knowledge_base(reload: bool = False):
    """
    [旧版] 从配置的单一根目录加载所有文档，并更新向量库。
    如果 reload=True，会清空现有数据后重建；否则增量更新（按文件路径去重）。
    注意：此函数保留仅为兼容，推荐使用 update_knowledge_base() 进行多源增量更新。
    """
    root_dir = config["knowledge_base"]["root_dir"]
    if not os.path.exists(root_dir):
        print(f"知识库目录不存在: {root_dir}")
        return

    # 支持的文档扩展名
    extensions = [".txt", ".md", ".pdf", ".py", ".js", ".c", ".cpp", ".h", ".hpp"]
    loaders = []
    for ext in extensions:
        loaders.append(
            DirectoryLoader(
                root_dir,
                glob=f"**/*{ext}",
                loader_cls=TextLoader if ext != ".pdf" else PyPDFLoader,
                loader_kwargs={"encoding": "utf-8"} if ext != ".pdf" else {},
                recursive=True,
                show_progress=True,
            )
        )

    all_docs = []
    for loader in loaders:
        try:
            docs = loader.load()
            all_docs.extend(docs)
        except Exception as e:
            print(f"加载 {loader} 时出错: {e}")

    if not all_docs:
        print("未找到任何文档")
        return

    # 文本切分
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=config["knowledge_base"]["chunk_size"],
        chunk_overlap=config["knowledge_base"]["chunk_overlap"],
        separators=["\n\n", "\n", " ", ""],
    )
    chunks = text_splitter.split_documents(all_docs)

    if reload:
        # 清空现有数据
        vectorstore.delete_collection()
        vectorstore.persist()

    # 批量添加到向量库
    vectorstore.add_documents(chunks)
    vectorstore.persist()
    print(f"成功加载 {len(chunks)} 个文档块")

def retrieve(query: str, top_k: int = 5) -> List[Document]:
    """
    根据查询文本检索最相关的文档块。
    """
    results = vectorstore.similarity_search(query, k=top_k)
    return results

# ==================== 测试入口 ====================
if __name__ == "__main__":
    print("测试嵌入模型...")
    test_text = "Hello, world!"
    emb = embeddings.embed_query(test_text)
    print(f"嵌入维度: {len(emb)}")

    # 推荐使用新的增量更新函数（需要 knowledge_sources.json）
    print("\n开始增量更新知识库...")
    update_knowledge_base()

    # 可选：也可测试旧版函数
    # print("\n使用旧版加载...")
    # load_knowledge_base(reload=True)

    print("\n测试检索...")
    results = retrieve("代码审查", top_k=2)
    for i, doc in enumerate(results):
        print(f"\n--- 结果 {i+1} ---")
        print(doc.page_content[:200])

说明

load_knowledge_base(reload=False)：从 company_knowledge 目录加载所有文档（支持 .txt, .md, .pdf, 常见代码文件），切分后存入向量库。reload=True 会清空旧数据重新构建。
retrieve(query, top_k=5)：根据查询文本返回最相关的文档块。
代码中使用了 langchain_community 的文档加载器和 langchain_huggingface 的嵌入模型，这些依赖已在安装中包含。

接下来直接运行：

# 配置国内镜像源(可选)
$ export HF_ENDPOINT=https://hf-mirror.com
$ python rag.py

运行后可以测试配置是否正确，并初次构建知识库。

如果能出现大致这结果：

测试嵌入模型...
嵌入维度: 384

加载知识库...
0it [00:00, ?it/s]
....
未找到任何文档

测试检索...

就相当于通过了。

五、FastAPI 服务

在 ~/ai-code-review/ 目录下创建 main.py，内容如下：

import os
import yaml
import requests
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
import rag  # 导入我们之前写的 rag 模块

# 加载配置
with open("config.yaml", "r", encoding="utf-8") as f:
    config = yaml.safe_load(f)

app = FastAPI(title="代码审查助手", version="1.0")

# 请求体模型
class CodeReviewRequest(BaseModel):
    code: str
    language: Optional[str] = None
    file_path: Optional[str] = None

# 响应体模型
class CodeReviewResponse(BaseModel):
    result: str

@app.post("/review", response_model=CodeReviewResponse)
async def review_code(request: CodeReviewRequest):
    # 1. 根据代码片段检索知识库
    docs = rag.retrieve(request.code, top_k=5)
    knowledge_context = "\n\n".join([doc.page_content for doc in docs])
    
    # 2. 构造提示词
    prompt = f"""你是一个资深的代码审查专家，请基于以下参考资料，分析提供的代码逻辑是否正确，指出潜在问题、改进建议。

【参考资料】
{knowledge_context}

【待审查代码】
{request.code}

请给出详细的审查结果，按问题严重性排序。如果代码正确，也请确认。"""
    
    # 3. 调用 Ollama 模型
    ollama_url = config["api"]["ollama_url"] + "/api/generate"
    payload = {
        "model": config["model"]["name"],
        "prompt": prompt,
        "stream": False,
        "options": {
            "temperature": config["model"]["temperature"],
            "num_predict": config["model"]["max_tokens"]
        }
    }
    
    try:
        response = requests.post(ollama_url, json=payload, timeout=60)
        response.raise_for_status()
        result = response.json()["response"]
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"模型调用失败: {str(e)}")
    
    return CodeReviewResponse(result=result)

@app.get("/health")
async def health_check():
    return {"status": "ok"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host=config["api"]["host"], port=config["api"]["port"])

然后直接运行 python main.py，如果能出现：

$ python main.py
Loading weights: 100%|██████████████████████| 199/199 [00:00<00:00, 8887.85it/s]
BertModel LOAD REPORT from: BAAI/bge-small-en-v1.5
Key                     | Status     |  | 
------------------------+------------+--+-
embeddings.position_ids | UNEXPECTED |  | 

Notes:
- UNEXPECTED: can be ignored when loading from different task/architecture; not ok if you expect identical arch.
/Users/zohar/ai-code-review/rag.py:32: LangChainDeprecationWarning: The class `Chroma` was deprecated in LangChain 0.2.9 and will be removed in 1.0. An updated version of the class exists in the `langchain-chroma package and should be used instead. To use it run `pip install -U `langchain-chroma` and import as `from `langchain_chroma import Chroma``.
  vectorstore = Chroma(
INFO:     Started server process [69681]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

需要安装 requests（如果未安装，执行 pip install requests）。
首次启动前，确保 Ollama 服务已在后台运行（之前已通过 brew services start ollama 启动）

服务启动后，简单测试一句代码审查：

$ curl -X POST http://localhost:8000/review \
>   -H "Content-Type: application/json" \
>   -d '{"code": "def add(a,b):\n    return a+b"}'

返回可能会有点久，要耐心等一会。到目前这一步，由于知识库还没有东西，所以理论上会返回类似：

{"result":" 好的，我会对你的代码进行详细的审查，并指出潜在的问题和改进建议。以下是基于提供的参考资料对代码的分析：\n\n### 待审查代码\n```python\ndef add(a, b):\n    return a + b\n```\n\n### 审查结果\n\n1. **代码正确性**\n   - **问题描述**：代码实现了一个简单的加法函数，没有明显的语法错误或逻辑错误。\n   - **严重性**：低\n   - **改进建议**：无需改进，代码是正确的。\n\n2. **代码风格**\n   - **问题描述**：代码简洁明了，符合Python的编码规范。\n   - **严重性**：低\n   - **改进建议**：可以考虑添加一些注释来解释函数的功能，特别是在团队协作中，这有助于其他开发者理解代码。\n\n### 具体改进建议\n\n1. **添加注释**\n   - **问题描述**：虽然代码功能简单，但为了提高可读性和可维护性，建议在函数定义前添加一行注释，说明该函数的用途。\n   - **改进建议**：\n     ```python\n     def add(a, b):\n         \"\"\"\n         返回两个数 a 和 b 的和。\n         :param a: int or float 第一个加数\n         :param b: int or float 第二个加数\n         :return: int or float 和\n         \"\"\"\n         return a + b\n     ```\n   - **严重性**：低\n\n### 总结\n代码实现了一个简单的加法函数，没有明显的错误。为了提高可读性和可维护性，建议添加注释说明函数的用途。总体来说，代码是正确的，但可以通过改进注释来提升代码质量。\n\n希望这些分析和建议对你有帮助！如果有更多问题或需要进一步的审查，请随时告知

但无论如何，已经有返回了。

关于速度慢的原因
模型较大：deepseek-coder-v2:16b 是 16B 参数，量化后 8.9GB，在 M2 上推理约 10–20 token/s，首次调用会加载模型（耗时数十秒），后续调用会快一些。
建议：如果追求速度，可换成更小的模型，如 qwen2.5-coder:7b-instruct-q4_K_M（约 4.7GB），推理速度会明显提升。

六、知识库索引json

在 ~/ai-code-review 下：

$ touch knowledge_sources.json
$ vim knowledge_sources.json

然后写入你的参考资料路径，可配置排除检索的目录及文件，比如：

{
  "sources": [
    {
      "name": "Sky750串口自测软件(Python版)",
      "path": "/Users/zohar/Documents/iw/sky750-serial/",
      "include_extensions": [".py", ".bat", ".txt", ".md"],
      "exclude_dirs": ["venv"]
    },
    { 
      "name": "道闸雷达Uniapp",
      "path": "/Users/zohar/Documents/iw/iw-radar-uniapp/",
      "include_extensions": [".js", ".ts", ".vue", ".md"],
      "exclude_dirs": ["node_modules", "dist", "keystore"],
      "exclude_files": []
    }
  ]
}

字段说明：

name：来源名称（用于日志）
path：目录绝对路径（可指向单个文件，但我们假设是目录）
include_extensions：要加载的文件扩展名（留空则加载所有支持的扩展名，但需与排除规则配合）
exclude_dirs：要跳过的目录名（支持通配符？这里简化为精确匹配目录名，若需复杂匹配可扩展）
exclude_files：要跳过的文件名模式（支持通配符如 *.log）

接下来，在 ~/ai-code-review/ 下创建 update_kb.py，内容：

import rag

if __name__ == "__main__":
    rag.update_knowledge_base()

运行方式：

$ python update_kb.py

即可增量更新知识库。会出现类似：

BertModel LOAD REPORT from: BAAI/bge-small-en-v1.5
Key                     | Status     |  | 
------------------------+------------+--+-
embeddings.position_ids | UNEXPECTED |  | 

Notes:
- UNEXPECTED: can be ignored when loading from different task/architecture; not ok if you expect identical arch.
/Users/zohar/ai-code-review/rag.py:33: LangChainDeprecationWarning: The class `Chroma` was deprecated in LangChain 0.2.9 and will be removed in 1.0. An updated version of the class exists in the `langchain-chroma package and should be used instead. To use it run `pip install -U `langchain-chroma` and import as `from `langchain_chroma import Chroma``.
  vectorstore = Chroma(
处理源: Sky750串口自测软件(Python版) (/Users/zohar/Documents/iw/sky750-serial/)
  更新文件: /Users/zohar/Documents/iw/sky750-serial/requirements.txt
/Users/zohar/ai-code-review/rag.py:203: LangChainDeprecationWarning: Since Chroma 0.4.x the manual persistence method is no longer supported as docs are automatically persisted.
  vectorstore.persist()
    添加了 1 个文档块
  更新文件: /Users/zohar/Documents/iw/sky750-serial/README.md
    添加了 4 个文档块

七、测试知识库

做完上一步的增量更新，直接在 ~/ai-code-review 下执行：

$ python main.py

运行起服务后，新开一个终端测试：

$ curl -X POST http://localhost:8000/review \
  -H "Content-Type: application/json" \
  -d '{"code": "def add(a,b): return a+b"}'

看看返回了什么，与第五步有没有区别。我这里返回的是：

$ curl -X POST http://localhost:8000/review \
>   -H "Content-Type: application/json" \
>   -d '{"code": "def add(a,b): return a+b"}'
{"result":" 您提供的代码是Python代码，用于两个向量的加法操作。让我们详细分析一下这段代码：\n\n```python\ndef add(a, b):\n    return a + b\n```\n\n### 审查结果\n\n1. **代码正确性**：\n   - 从语法上看，这段代码是正确的。它定义了一个函数 `add`，该函数接受两个参数 `a` 和 `b`，并返回它们的和。这在Python中是完全合法的。\n\n2. **潜在问题**：\n   - 虽然代码在语法上是正确的，但我们需要考虑以下几点：\n     - **输入类型检查**：这段代码没有对输入进行类型检查，假设传入的两个参数都是可以相加的（例如，如果它们是数字或具有 `+` 操作符的重载）。这可能会导致运行时错误，特别是如果传入的对象不支持加法操作。\n     - **性能考虑**：对于简单的加法操作，这种实现方式是合理的。但如果涉及到更复杂的向量运算（例如，涉及不同类型的向量或需要更多的上下文处理），可能需要更复杂的逻辑来确保正确性和效率。\n\n### 改进建议\n\n1. **输入类型检查**：\n   - 为了提高代码的健壮性，可以添加对输入参数的类型检查。例如：\n     ```python\n     def add(a, b):\n         if not isinstance(a, (int, float)) or not isinstance(b, (int, float)):\n             raise ValueError(\"Inputs must be numbers\")\n         return a + b\n     ```\n   - 如果参数是自定义对象或向量，可以添加相应的类型检查：\n     ```python\n     def add(a, b):\n         if not isinstance(a, (int, float)) or not isinstance(b, (int, float)):\n             raise ValueError(\"Inputs must be numbers\")\n         return a + b\n     \n     def add_vector(v1, v2):\n         if not isinstance(v1, Vector) or not isinstance(v2, Vector):\n             raise ValueError(\"Inputs must be vectors\")\n         return Vector(v1.x + v2.x, v1.y + v2.y, v1.z + v2.z)\n     ```\n\n2. **文档和注释**：\n   - 添加函数文档字符串，说明函数的用途、参数和返回值。例如：\n     ```python\n     def add(a, b):\n         \"\"\"\n         对两个数进行加法运算。\n         \n         参数:\n             a (int, float): 第一个加数。\n             b (int, float): 第二个加数。\n         \n         返回:\n             int, float: 两数的和。\n         \"\"\"\n         if not isinstance(a, (int, float)) or not isinstance(b, (int, float)):\n             raise ValueError(\"Inputs must be numbers\")\n         return a + b\n     ```\n\n3. **性能优化**：\n   - 对于简单的加法操作，这种实现方式是高效的。但如果涉及到大量运算或性能敏感的应用程序，可以考虑使用更高效的数据结构和算法。\n\n### 结论\n\n这段代码在语法上是正确的，但可以通过添加输入类型检查和文档注释来提高其健壮性和可读性。如果不需要复杂的逻辑，这种实现方式是合理的。如果有更多的上下文需求或性能要求，可以考虑进一步优化和扩展。"}

至此，就完成了知识库的更新、重启，并获取到返回数据。

八、通过对话形式调用大模型

需要一个交互式脚本，每次输入问题，系统自动检索知识库，然后调用模型生成回答。

在 ~/ai-code-review/ 目录下创建 chat.py，内容如下：

import rag
import requests
import yaml

# 加载配置
with open("config.yaml", "r", encoding="utf-8") as f:
    config = yaml.safe_load(f)

def ask_question(question: str):
    # 1. 检索知识库
    docs = rag.retrieve(question, top_k=5)
    context = "\n\n".join([doc.page_content for doc in docs])
    
    # 2. 构造提示词
    prompt = f"""你是一个知识渊博的助手，请基于以下参考资料回答用户的问题。

【参考资料】
{context}

【用户问题】
{question}

请根据参考资料回答，如果参考资料中没有相关信息，请说明你不知道。"""
    
    # 3. 调用 Ollama
    ollama_url = config["api"]["ollama_url"] + "/api/generate"
    payload = {
        "model": config["model"]["name"],
        "prompt": prompt,
        "stream": False,
        "options": {
            "temperature": config["model"]["temperature"],
            "num_predict": config["model"]["max_tokens"]
        }
    }
    try:
        response = requests.post(ollama_url, json=payload, timeout=60)
        response.raise_for_status()
        result = response.json()["response"]
        return result
    except Exception as e:
        return f"模型调用失败: {e}"

if __name__ == "__main__":
    print("对话模式已启动（输入 exit 退出）")
    while True:
        user_input = input("\n你: ")
        if user_input.lower() in ["exit", "quit"]:
            break
        print("思考中...")
        answer = ask_question(user_input)
        print(f"助手: {answer}")

运行对话：

$ python chat.py

效果：

上一页ex4nicegui 0.9.3 使用文档

最后更新于 2分钟前

hashtag一、准备环境与模型拉取

hashtag1. 安装 Ollama

hashtag2. 配置ollama镜像源

hashtag3. 拉取模型

hashtag二、Python版本与依赖安装

hashtag三、配置文件

hashtag四、配置RAG并初次构建知识库

hashtag五、FastAPI 服务

hashtag六、知识库索引json

hashtag七、测试知识库

hashtag八、通过对话形式调用大模型