【KnowPivot 开源实战 05】技术架构拆解：FastAPI + Chroma + Ollama 全本地闭环

webajax

478人浏览 · 2026-08-04 14:31:42

webajax · 2026-08-04 14:31:42 发布

本文从工程视角完整拆解 KnowPivot 的技术架构：如何用 FastAPI 构建异步 API 层、用 ChromaDB 实现文件型向量存储、用 Ollama 跑通本地 LLM 推理，以及配置热更新、结构化日志、Prometheus 监控等工程化能力的实现细节。适合正在设计私有化 RAG 系统架构的后端开发者和架构师阅读，读完你将获得一套"零外网依赖、零额外中间件"的全本地闭环架构设计参考。

前置阅读：建议先阅读第 3 篇：7 天从 0 到 1了解整体设计，以及第 4 篇：四层检索链路了解检索增强实现。

适用版本：KnowPivot v2.0+，Python 3.10+，FastAPI 0.111+，ChromaDB 0.5+，Ollama 0.3+

一、架构设计目标

1.1 全本地闭环的含义

"全本地闭环"不是口号，而是四条硬约束：

约束	含义	工程影响
数据不出本地	文档、向量、对话记录全部存储在服务器本地	不依赖云存储、云数据库
模型不出本地	LLM 和 Embedding 模型在本地推理	使用 Ollama + 本地 transformers 模型
中间件零依赖	不需要 Docker、Redis、MQ 等额外服务	SQLite 替代 PostgreSQL，内存缓存替代 Redis
一个进程跑起来	`python main.py` 即可启动全部服务	无需编排多个容器

注意：全本地不等于不能联网。KnowPivot 也支持通过 OpenAI 兼容 API 调用云端模型（如 DeepSeek、智谱）。全本地是默认模式和保底能力，而非唯一模式。

1.2 为什么选这条路线

在内部企业环境中，全本地路线有明确优势：

合规性：数据不经过任何外部网络，满足数据安全审计要求
成本：零 API 费用，零云服务费用，硬件成本仅为一台服务器
部署：不需要运维团队介入，pip install + python main.py 即可运行
可控性：模型版本、数据格式、检索逻辑全部自己掌控

代价是性能和规模上限：本地 LLM 推理速度受限于 GPU/CPU，知识库规模受限于单机存储。但中小规模场景（数千到数万文档，日均数百到数千次问答）完全够用。

二、整体架构分层

2.1 三层架构总览

KnowPivot 的架构分为三层，每层职责清晰、单向依赖：

在这里插入图片描述

图 1：KnowPivot 整体架构——API 层负责路由与协议，Core 层承载全部业务逻辑，Storage 层提供数据持久化

┌─────────────────────────────────────────────────────────┐
│                    API 层（FastAPI）                      │
│  health / kb / file / fragment / qa / chat / agent / llm │
├─────────────────────────────────────────────────────────┤
│                   Core 层（业务逻辑）                     │
│  rag_engine / chroma_vector / ollama_llm / api_embedding │
│  hybrid_search / reranker / query_rewriter / metrics     │
│  config_watcher / cache / json_logger / entity_graph     │
├─────────────────────────────────────────────────────────┤
│                  Storage 层（数据持久化）                  │
│  SQLite（业务数据） / ChromaDB（向量数据） / 本地文件系统   │
└─────────────────────────────────────────────────────────┘

2.2 分层依赖原则

核心规则：上层依赖下层，下层不感知上层。

api/ 层只做路由分发和参数校验，不写业务逻辑
core/ 层是全部业务能力的实现，不直接操作 HTTP 请求/响应
db/ 层只提供数据模型和连接，不包含业务逻辑

这样设计的好处是：后续如果要加 CLI 入口或 gRPC 接口，core/ 层完全不用改。

2.3 API 路由清单

系统共注册 9 个路由模块，覆盖完整的知识库管理链路：

路由模块	前缀	职责
`health_router`	`/api/health`	健康检测（FastAPI + Ollama + Chroma 三组件状态）
`kb_router`	`/api/kb`	知识库 CRUD
`file_router`	`/api/file`	文档上传、解析、分片
`fragment_router`	`/api/fragment`	片段管理（自动分片 + 手动片段）
`qa_router`	`/api/qa`	问答对管理
`chat_router`	`/api/chat`	RAG 问答 + 纯模型问答
`agent_router`	`/api/agent`	Agent 配置与管理、Agent 对话
`llm_router`	`/api/llm`	模型管理（Chat / Embedding / Rerank）
`analysis_router`	`/api/analysis`	检索质量分析与统计

三、API 层：FastAPI 异步框架

3.1 为什么选 FastAPI

候选框架	异步支持	文档生成	性能	生态	选择理由
FastAPI	原生 async	自动 Swagger	高	Python 生态丰富	异步 + 文档 + 类型安全
Flask	需异步扩展	需第三方	中	成熟	异步支持弱
Django	部分异步	需第三方	中	全栈但过重	对 RAG 后端过重
AIOHTTP	原生 async	无	高	偏底层	缺框架级便利

FastAPI 的核心优势是基于 Pydantic 的类型安全：请求体和响应体都有类型定义，参数校验自动完成，开发效率和代码质量同时得到保障。

3.2 统一响应格式

所有 API 返回统一的 {code, msg, data} 结构，通过全局异常处理器实现：

# 成功响应
{"code": 0, "msg": "success", "data": {...}}

# 参数校验失败
{"code": 422, "msg": "参数校验失败: kb_id: field required", "data": {"errors": [...]}}

# 服务端错误
{"code": 500, "msg": "服务器内部错误: ...", "data": null}

实现方式是注册三个全局异常处理器：

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    """HTTP 异常 → 统一格式"""
    return JSONResponse(status_code=exc.status_code,
        content={"code": exc.status_code, "msg": exc.detail, "data": None})

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    """参数校验失败 → 422 + 错误详情"""
    errors = exc.errors()
    detail = "; ".join([f"{e['loc'][-1]}: {e['msg']}" for e in errors])
    return JSONResponse(status_code=422,
        content={"code": 422, "msg": f"参数校验失败: {detail}", "data": {"errors": errors}})

@app.exception_handler(Exception)
async def unhandled_exception_handler(request, exc):
    """未捕获异常 → 500 + 日志记录"""
    logger.error(f"未捕获异常: {type(exc).__name__}: {exc}", exc_info=True)
    return JSONResponse(status_code=500,
        content={"code": 500, "msg": f"服务器内部错误: {str(exc)[:200]}", "data": None})

关键设计：未捕获异常处理器是兜底防线。任何未被业务代码捕获的异常，都会在这里被拦截并转为统一格式，避免直接向客户端暴露堆栈信息。

3.3 请求监控中间件

每个 API 请求都会经过监控中间件，记录耗时和状态码：

@app.middleware("http")
async def metrics_middleware(request: Request, call_next):
    method = request.method
    path = request.url.path
    start_time = time.time()

    response = await call_next(request)

    duration = time.time() - start_time
    # 记录 Prometheus 指标
    record_http_request(method, path, response.status_code, duration)
    # 结构化日志（仅 API 路径）
    if path.startswith("/api/"):
        logger.info(f"{method} {path} -> {response.status_code} ({duration*1000:.1f}ms)",
            extra={"method": method, "path": path, "status_code": response.status_code,
                   "duration_ms": round(duration * 1000, 1)})
    return response

注意：中间件会对所有请求（包括静态文件）执行。日志输出做了 path.startswith("/api/") 过滤，避免静态资源请求刷屏日志。

四、Core 层：多模型兼容的 LLM 调用

4.1 为什么不直接用 LangChain 的 LLM 接口

KnowPivot 在 LLM 调用上选择了零 LangChain LLM 依赖的路线，直接用 httpx 调用 HTTP 接口：

方案	优点	缺点
`langchain-openai`	封装完善	版本冲突频繁、抽象层开销
`langchain-ollama`	开箱即用	不支持 `think` 参数等原生特性
直接 httpx 调用	零依赖冲突、完全掌控	需自己处理错误和重试

核心考量是可控性：Ollama 的 /api/chat 接口支持 think 参数控制 Qwen3 系列的思考模式，LangChain 封装层不一定及时跟进。直接调用 HTTP 接口，所有参数完全可控。

4.2 多 Provider 统一接口

KnowPivot 支持四种 LLM Provider，通过统一的 chat_llm() 函数分发：

async def chat_llm(prompt, model=None, temperature=0.7, 
                   base_url=None, api_key=None, provider="ollama", **kwargs):
    """统一 LLM 调用入口（带并发限制）"""
    async with _llm_semaphore:  # 信号量限制并发，避免 GPU 显存不足
        if provider == "ollama":
            return await _chat_ollama_http(prompt, model, temperature, base_url, ...)
        elif provider in ("openai", "azure", "custom"):
            return await _chat_openai_http(prompt, model, temperature, base_url, api_key)
        else:
            raise ValueError(f"不支持的 provider: {provider}")

Provider 与 API 路径映射：

Provider	API 路径	典型场景
`ollama`	`{base_url}/api/chat`	本地 Ollama 部署
`openai`	`{base_url}/v1/chat/completions`	OpenAI 或兼容 API
`azure`	`{base_url}/v1/chat/completions`	Azure OpenAI
`custom`	`{base_url}/v1/chat/completions`	自定义 API 端点

关键设计：LLM 调用使用全局信号量 _llm_semaphore = asyncio.Semaphore(3) 限制并发。本地 GPU 显存有限，同时跑 3 个推理任务已是上限。这个参数可通过修改源码调整。

4.3 Ollama 调用的关键细节

Ollama 原生接口有几个需要注意的点：

payload = {
    "model": model,
    "messages": [{"role": "user", "content": prompt}],
    "stream": False,
    "options": {"temperature": temperature},
}
# 关键：关闭 Qwen3/Qwen3.5 思考模式，避免输出 <think> 标签
if not enable_think:
    payload["think"] = False

think 参数说明：Qwen3 系列模型默认开启思考模式，输出会包含 <think> 标签包裹的推理过程。在 RAG 场景中，思考内容不应出现在最终回答里。通过 think=False 关闭，确保输出干净。

错误处理针对 Ollama 的典型错误做了细化：

if resp.status_code == 404:
    raise RuntimeError(
        f"Ollama API 请求失败 [404]: 模型 '{model}' 不存在。"
        f"请确认该模型已下载（ollama pull {model}）。"
    )

风险提示：Ollama 超时设为 300 秒。本地 LLM 推理速度受 GPU 性能影响大，7B 模型在纯 CPU 上生成 500 字回答可能需要 30-60 秒。如果使用更大模型（如 32B），建议适当调大超时或使用 GPU。

4.4 Embedding 多源适配

Embedding 同样支持多来源，通过工厂函数按 Provider 分发：

def create_embedding_function(model_key, provider, base_url=None, api_key=None):
    """根据配置创建 Embedding 实例，自动包装缓存层"""
    if provider == "local":
        # 本地 transformers 模型（从 models/ 目录加载）
        inner = LocalTransformersEmbeddings(model_path=base_url or get_local_model_path())
        return CachedEmbeddings(inner, cache_prefix="emb_local")
    if provider == "ollama":
        inner = OllamaEmbeddings(model=model_key, base_url=base_url)
        return CachedEmbeddings(inner, cache_prefix="emb_ollama")
    # openai / azure / custom → OpenAI 兼容接口
    inner = OpenAICompatibleEmbeddings(model=model_key, base_url=base_url, api_key=api_key)
    return CachedEmbeddings(inner, cache_prefix="emb_api")

三种 Embedding 来源对比：

来源	模型示例	延迟	依赖	适用场景
本地 transformers	bge-base-zh-v1.5	10-30ms	torch + transformers	全本地部署
Ollama	nomic-embed-text	5-15ms	Ollama 服务	已有 Ollama 环境
API	text-embedding-3-small	50-100ms	外网 API	不要求全本地

所有来源都会自动包装 CachedEmbeddings 缓存层，相同文本不重复向量化。

五、Core 层：缓存与配置热更新

5.1 双后端缓存

KnowPivot 的缓存层设计了一个统一的抽象接口，支持内存缓存和 Redis 两种后端，自动切换：

def _get_cache():
    """自动选择缓存后端：有 Redis 用 Redis，否则用内存"""
    if REDIS_URL:
        rc = RedisCache(REDIS_URL)
        if rc._client is not None:
            return rc  # Redis 连接成功
    return MemoryCache()  # 回退到内存缓存

设计意图：单机部署时不需要 Redis，内存缓存够用。多实例部署时配置 Redis 即可自动切换，业务代码无感知。

后端	最大容量	过期策略	持久化	适用场景
内存缓存	5000 条	超 LRU 淘汰一半	否	单机部署
Redis	无上限	TTL 过期	是	多实例部署

Embedding 缓存是最核心的应用场景。文档入库时，相同文本片段不会重复向量化；查询时，相同问题不会重复 Embedding。实测缓存命中率可达 60-80%，大幅降低 Embedding 调用延迟。

5.2 配置热更新

KnowPivot 支持修改 .env 文件后自动热加载，无需重启服务：

def _watch_loop(interval: int):
    """后台线程：每隔 interval 秒检查 .env 文件 mtime"""
    env_path = os.path.join(BASE_DIR, ".env")
    last_mtime = _get_mtime(env_path)

    while not _stop_event.is_set():
        _stop_event.wait(interval)  # 非阻塞等待
        current_mtime = _get_mtime(env_path)
        if current_mtime and current_mtime != last_mtime:
            last_mtime = current_mtime
            settings.reload()
            logger.info("[CONFIG WATCHER] 检测到 .env 变更，配置已热更新")

热更新的边界：

配置项	热更新支持	说明
`OLLAMA_BASE_URL`	支持	下次 LLM 调用即生效
`DEFAULT_TOP_K`	支持	下次检索即生效
`SIMILARITY_THRESHOLD`	支持	下次检索即生效
`LOG_LEVEL`	支持	立即生效
`DATABASE_URL`	不支持	数据库连接已初始化，需重启
`CHROMA_PERSIST_DIR`	不支持	Chroma 实例已创建，需重启

注意：热更新检查间隔默认 30 秒，通过 config_watch_interval 配置。设为 0 可禁用热更新。Agent 级别的参数（Top-K、阈值、温度等）存储在数据库中，通过 API 修改后立即生效，不依赖 .env 热更新。

六、Core 层：可观测性三件套

6.1 结构化 JSON 日志

所有日志输出为 JSON 格式，方便 ELK / Loki 等日志系统采集和检索：

{
  "timestamp": "2026-07-25T10:30:00.123456+00:00",
  "level": "INFO",
  "logger": "main",
  "module": "main",
  "line": 156,
  "message": "POST /api/chat/rag-chat -> 200 (350.2ms)",
  "service": "KnowPivot",
  "method": "POST",
  "path": "/api/chat/rag-chat",
  "status_code": 200,
  "duration_ms": 350.2
}

实现方式是自定义 JSONFormatter，在标准日志字段基础上附加结构化业务字段：

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "service": self.service_name,
        }
        # 附加通过 extra 传入的结构化字段
        safe_attrs = {"agent_id", "kb_id", "request_id", "method", "path",
                      "status_code", "duration_ms", "provider", "model",
                      "error_type", "query_len", "ref_count"}
        for attr in safe_attrs:
            val = getattr(record, attr, None)
            if val is not None:
                log_entry[attr] = val
        return json.dumps(log_entry, ensure_ascii=False, default=str)

设计要点：日志使用 stderr 输出，避免与 uvicorn 的 stdout 冲突。第三方库（httpx、chromadb、langchain 等）的日志级别被抑制到 WARNING，避免冗余输出。

6.2 Prometheus 指标收集

KnowPivot 内置 Prometheus 指标收集，通过 /metrics 端点暴露：

指标	类型	标签	用途
`http_requests_total`	Counter	method, path, status	API 请求总数
`http_request_duration_seconds`	Histogram	method, path	API 请求延迟分布
`llm_requests_total`	Counter	provider, model	LLM 调用次数
`llm_request_duration_seconds`	Histogram	provider, model	LLM 调用延迟
`llm_request_errors_total`	Counter	provider, model	LLM 调用失败数
`rag_vector_search_duration_seconds`	Histogram	kb_id	向量检索延迟
`rag_rerank_duration_seconds`	Histogram	-	重排序延迟
`rag_references_count`	Histogram	agent_id	检索返回的参考文档数
`db_query_duration_seconds`	Histogram	operation	数据库查询延迟

指标收集采用延迟导入策略，未安装 prometheus_client 时自动降级为空操作：

def _ensure_prometheus():
    """延迟导入，未安装时回退到空操作"""
    global _REGISTRY
    if _REGISTRY is not None:
        return True
    from config import settings
    if not settings.enable_metrics:
        return False
    try:
        from prometheus_client import CollectorRegistry
        _REGISTRY = CollectorRegistry()
        return True
    except ImportError:
        logger.warning("prometheus_client 未安装，指标收集已禁用")
        return False

注意：/metrics 端点挂载为独立 Starlette 应用，不经过主应用的中间件链，避免指标采集本身被监控中间件记录形成循环。

6.3 Sentry 错误追踪

通过环境变量配置启用 Sentry，未配置则跳过：

from config import settings
if settings.sentry_dsn:
    try:
        import sentry_sdk
        sentry_sdk.init(
            dsn=settings.sentry_dsn,
            traces_sample_rate=settings.sentry_traces_sample_rate,
            environment=settings.sentry_environment,
        )
    except ImportError:
        logger.warning("sentry-sdk 未安装，错误追踪已禁用")

三件套的协作关系：
在这里插入图片描述

图 2：可观测性三件套协作关系——监控中间件统一采集，正常请求记日志，异常请求同时触发日志和 Sentry 上报

请求进入 → 监控中间件记录 Prometheus 指标
              ├→ 正常响应 → 结构化日志（INFO 级）
              └→ 异常发生 → 结构化日志（ERROR 级）+ Sentry 上报

七、Storage 层：数据持久化设计

7.1 SQLite 业务数据

KnowPivot 使用 SQLite 存储全部业务数据，共 9 张表：

表名	职责	关键字段
`knowledge_base`	知识库元信息	`embedding_model_id`（绑定向量化模型）
`file_record`	文件上传记录	`save_path`（本地存储路径）
`chunk_fragment`	文档分片 + 手动片段	`fragment_type`（auto/manual）、`vector_id`
`qa_pair`	问答对	`vector_id`（关联 ChromaDB）
`llm_model`	模型配置	`provider`、`model_type`（chat/embedding）
`agent`	Agent 配置	20+ 字段，包含全部检索增强参数
`chat_history`	对话历史	`references_json`、`retrieval_stats_json`
`entity` / `entity_occurrence`	实体图谱	轻量 GraphRAG 支持
`system_config`	系统配置	键值对存储（如自动清理天数）

为什么选 SQLite 而不是 PostgreSQL？

维度	SQLite	PostgreSQL
部署	零安装，单文件	需安装数据库服务
并发	写锁串行（单写多读）	多写并发
运维	零运维	需备份、监控、调优
适用规模	中小型（<100GB）	中大型

KnowPivot 的场景是内网单用户或低并发，SQLite 的写锁串行不会成为瓶颈。且 SQLite 的单文件特性让备份变得极其简单——复制 .db 文件即可。

7.2 ChromaDB 向量数据

ChromaDB 按 Collection 隔离不同知识库的向量数据：

# 每个知识库独立的 Collection
collection_name = f"kb_{kb_id}"        # 文档片段向量
qa_collection_name = f"qa_{kb_id}"     # 问答对向量

vectorstore = Chroma(
    collection_name=collection_name,
    embedding_function=emb_fn,
    persist_directory=os.path.join(CHROMA_PERSIST_DIR, collection_name),
)

目录结构：

chroma_data/
├── kb_{uuid}/           # 知识库的文档片段向量
│   ├── chroma.sqlite3   # Collection 元数据
│   └── {hash}/          # HNSW 索引文件
│       ├── data_level0.bin
│       ├── header.bin
│       └── link_lists.bin
├── qa_{uuid}/           # 问答对向量（独立 Collection）
│   └── ...
└── _hybrid_index/       # BM25 索引文件
    └── bm25_{uuid}_fragment.pkl

为什么 QA 和文档分片用不同的 Collection？ 因为两者检索时使用不同的阈值（QA 阈值 1.0，片段阈值 1.5），分开存储可以独立检索、独立调参，实现分层检索优先级。

7.3 向量库实例缓存

为避免并发时重复创建 Chroma 实例导致文件锁冲突，KnowPivot 对向量库实例做了线程安全缓存：

_chroma_cache = {}       # Collection 实例缓存
_chroma_lock = threading.Lock()

def get_vector_store(kb_id: str, is_qa: bool = False):
    cache_key = f"{kb_id}_{'qa' if is_qa else 'kb'}"
    with _chroma_lock:
        if cache_key in _chroma_cache:
            return _chroma_cache[cache_key]
    # 创建新实例...
    with _chroma_lock:
        _chroma_cache[cache_key] = vs
    return vs

注意：修改知识库的 Embedding 模型配置后，必须调用 clear_embedding_cache(kb_id) 清除缓存，否则会继续使用旧的 Embedding 实例。

八、数据库自动迁移

8.1 为什么不用 Alembic

Alembic 是 SQLAlchemy 的标准迁移工具，但 KnowPivot 没有使用它，而是在 main.py 启动时做手动迁移：

# 启动时自动检测并添加缺失字段
with engine.connect() as conn:
    inspector = inspect(engine)
    agent_cols = {c['name'] for c in inspector.get_columns('agent')}
    if 'use_rerank' not in agent_cols:
        conn.execute(text("ALTER TABLE agent ADD COLUMN use_rerank INTEGER DEFAULT 0"))
        conn.commit()
        logger.info("[MIGRATE] Agent 表新增 use_rerank 字段")

取舍逻辑：

方案	优点	缺点
Alembic	版本化管理、支持回滚	学习成本、额外依赖、对 SQLite 支持有限
启动时手动迁移	零依赖、零学习成本、SQLite 友好	不支持回滚、迁移逻辑散落在代码中

对于单文件 SQLite + 小团队迭代场景，手动迁移足够简单实用。SQLite 的 ALTER TABLE ADD COLUMN 语句不会锁表，迁移过程对运行中的服务无影响。

8.2 迁移覆盖范围

启动时迁移会检查并创建以下内容：

Agent 表字段：use_rerank、rerank_model_id、rerank_top_k、rerank_threshold、use_hybrid_search、multi_query_count、multi_query_model_id 等 10 个字段
ChatHistory 表字段：kb_id、is_answered、references_json、retrieval_stats_json
新表创建：chat_history、entity、entity_occurrence、system_config

设计原则：每次新增功能字段，都在启动迁移中加一段 if field not in cols 检查。旧版本升级到新版本时，启动即自动迁移，无需手动执行迁移命令。

九、后台任务与生命周期管理

9.1 配置热更新线程

# daemon 线程，主进程退出时自动终止
_watcher_thread = threading.Thread(target=_watch_loop, args=(interval,), daemon=True)
_watcher_thread.start()

9.2 对话历史自动清理

后台线程定时清理过期对话历史，清理周期由 system_config 表中的 auto_cleanup_days 配置控制：

def cleanup_loop():
    time.sleep(10)  # 首次启动等待服务初始化
    while True:
        try:
            setting = db.query(SystemConfig).filter(
                SystemConfig.config_key == "auto_cleanup_days").first()
            days = int(setting.config_value) if setting and setting.config_value else 0
            if days > 0:
                cutoff = datetime.utcnow() - timedelta(days=days)
                old_records = db.query(ChatHistory).filter(
                    ChatHistory.create_time < cutoff).all()
                # 按会话维度删除（保留会话完整性）
                session_ids = set(r.session_id for r in old_records)
                if session_ids:
                    db.query(ChatHistory).filter(
                        ChatHistory.session_id.in_(session_ids)).delete(synchronize_session=False)
                    db.commit()
        except Exception as e:
            logger.error(f"[AUTO_CLEANUP] 自动清理失败: {e}")
        time.sleep(3600)  # 每小时检查一次

关键设计：清理按会话维度删除而非按单条记录，确保对话历史的完整性——要么整个会话保留，要么整个会话删除。

9.3 线程安全与并发控制

KnowPivot 在多处使用信号量和锁来控制并发：

机制	作用	位置
`_llm_semaphore` (Semaphore=3)	限制同时 LLM 调用数	`ollama_llm.py`
`_rag_retrieval_semaphore` (Semaphore=3)	限制同时 RAG 检索数	`rag_engine.py`
`_chroma_lock` (threading.Lock)	保护 Chroma 实例缓存	`chroma_vector.py`
`_embedding_lock` (threading.Lock)	保护 Embedding 实例缓存	`chroma_vector.py`
`_bm25_lock` (threading.Lock)	保护 BM25 索引缓存	`hybrid_search.py`
`_reranker_lock` (threading.Lock)	保护重排序模型缓存	`reranker.py`

注意：这些并发限制是针对本地部署场景的保守策略。如果使用云端 API（无 GPU 显存限制），可以适当调大信号量。但 ChromaDB 和 Embedding 的文件锁冲突与部署方式无关，锁保护始终需要。

十、依赖管理与部署

10.1 依赖清单

KnowPivot 的依赖保持精简，全部在 requirements.txt 中：

依赖	版本	用途	是否必须
`fastapi`	0.111+	Web 框架	必须
`uvicorn`	0.30+	ASGI 服务器	必须
`sqlalchemy`	2.0+	ORM	必须
`chromadb`	0.5+	向量数据库	必须
`langchain-chroma`	0.1+	Chroma 持久化封装	必须
`pydantic-settings`	2.0+	配置管理	必须
`pdfplumber`	0.11+	PDF 解析	必须
`python-docx`	1.2+	Word 解析	可选
`rank-bm25`	0.2.2+	BM25 检索	Hybrid Search 需要
`sentence-transformers`	3.0+	本地 Embedding	本地模型需要
`torch`	-	模型推理	本地模型 + 重排序需要
`prometheus-client`	0.20+	指标收集	监控需要
`sentry-sdk`	2.0+	错误追踪	可选
`redis`	-	Redis 缓存	多实例部署需要

注意：torch 和 sentence-transformers 是最重的依赖（约 2GB 安装体积）。如果只使用 API Embedding（不使用本地模型），可以不安装这两个包，系统会自动降级为 API 调用模式。

10.2 部署方式

# 1. 克隆项目
git clone https://github.com/your-username/KnowPivot.git
cd KnowPivot/mini_rag_backend

# 2. 安装依赖
pip install -r requirements.txt

# 3. 配置环境
cp .env.example .env
# 编辑 .env，配置 Ollama 地址、端口等

# 4. 启动服务
python main.py
# 服务默认监听 127.0.0.1:8001

# 5. 验证
curl http://127.0.0.1:8001/api/health
# 预期返回：{"code":0,"msg":"success","data":{"fastapi":"ok","ollama":"ok","chroma":"ok"}}

风险提示：生产环境部署时，host 应设为 127.0.0.1，通过 Nginx 反向代理对外暴露。直接将服务暴露到公网存在安全风险——KnowPivot 默认无鉴权，任何人都能访问所有 API。

十一、总结

11.1 架构设计要点

KnowPivot 的架构设计围绕三条主线：

分层解耦：API / Core / Storage 三层单向依赖，Core 层不感知 HTTP，可复用于 CLI 或其他入口
多模型兼容：LLM 和 Embedding 统一接口，支持 Ollama / OpenAI / 本地 transformers 三种来源，按需切换
渐进式增强：缓存、监控、热更新等工程化能力是叠加层，不影响核心链路，可按需开关

11.2 适用边界

本架构适用于：

中小规模知识库（数千到数万片段）
内网私有化部署场景
对数据安全有合规要求的团队
需要快速验证 RAG 效果的项目

不适用于：

高并发场景（SQLite 写锁串行 + 本地 LLM 推理瓶颈）——需换 PostgreSQL + GPU 集群
百万级文档——ChromaDB 文件型存储的检索性能会下降，需换 Milvus
多租户 SaaS——缺少用户体系和权限隔离，需增加鉴权层

11.3 架构演进方向

当前架构的已知瓶颈和演进方向：

瓶颈	当前方案	演进方向
SQLite 写锁	单机单写	切换 PostgreSQL，支持多写并发
单进程 LLM	Semaphore=3	多 GPU 实例 + 负载均衡
ChromaDB 扩展性	文件型存储	切换 Milvus / Qdrant
无鉴权	内网信任	增加 API Token / JWT 鉴权层
单机部署	`python main.py`	Docker Compose 编排 + K8s

关键认知：架构演进应该是"到瓶颈再演进"，而不是"一开始就上全套"。KnowPivot 的轻量化设计让 MVP 能在 7 天内跑起来，演进时每个环节都有清晰的替换路径。

11.4 下一篇预告

第 6 篇：《从单 Agent 到多 Agent：KnowPivot 的智能体配置体系设计》，将拆解 Agent 配置模型的设计思路——如何让不同业务场景使用不同的知识库、模型和检索策略，以及兜底模型切换、答案一致性校验等高级能力。

项目地址：https://gitcode.com/mzsoft/KnowPivot.git

你在私有化 RAG 部署中遇到过哪些"中间件太多、部署太重"的痛点？欢迎在评论区交流！

系列导航：

上一篇：混合检索 + 重排序 + Multi-Query：四层 RAG 检索链路实战
下一篇：《Agent 智能对话配置实战：从知识库到客服系统的完整链路》（即将发布）

亚马逊云科技技术品牌专区

更多推荐

基于STM32的全屋环境智能监测与预警终端设计

亚马逊云科技技术品牌专区

Keras：面向人类的深度学习库

Keras3支持多后端（TensorFlow、PyTorch、JAX），通过设置环境变量切换，兼容指定版本。pip安装后可用Sequential API快速构建CNN模型，如MNIST分类示例，实

亚马逊云科技技术品牌专区

AI行业日报｜2026-09-25：5个热点事件

本期行业动态聚焦于端侧算力重构、自主智能体工作流、开发工具经济模型、网络隔离架构设计与公共部门合规审查。以下结合一线技术场景进行结构化梳理，供 AI 研究员与软件工程团队参考。 PrismML 正将其轻量级语言模型适配至搭载高通处理器的智能眼镜终端 1 。该平台的核心战略指向开源权重模型的本地化运行，旨在最大化挖掘穿戴设备现有算力的利用效率 1 。对于嵌入式与边缘计算开发者而言，此类微型模型向低功

亚马逊云科技技术品牌专区

所有评论(0)

查看更多评论

webajax

@webajax

已为社区贡献1条内容

【KnowPivot 开源实战 05】技术架构拆解：FastAPI + Chroma + Ollama 全本地闭环

webajax

一、架构设计目标

1.1 全本地闭环的含义

1.2 为什么选这条路线

二、整体架构分层

2.1 三层架构总览

2.2 分层依赖原则

2.3 API 路由清单

三、API 层：FastAPI 异步框架

3.1 为什么选 FastAPI

3.2 统一响应格式

3.3 请求监控中间件

四、Core 层：多模型兼容的 LLM 调用

4.1 为什么不直接用 LangChain 的 LLM 接口

4.2 多 Provider 统一接口

4.3 Ollama 调用的关键细节

4.4 Embedding 多源适配

五、Core 层：缓存与配置热更新

5.1 双后端缓存

5.2 配置热更新

六、Core 层：可观测性三件套

6.1 结构化 JSON 日志

6.2 Prometheus 指标收集

6.3 Sentry 错误追踪

七、Storage 层：数据持久化设计

7.1 SQLite 业务数据

7.2 ChromaDB 向量数据

7.3 向量库实例缓存

八、数据库自动迁移

8.1 为什么不用 Alembic

8.2 迁移覆盖范围

九、后台任务与生命周期管理

9.1 配置热更新线程

9.2 对话历史自动清理

9.3 线程安全与并发控制

十、依赖管理与部署

10.1 依赖清单

10.2 部署方式

十一、总结

11.1 架构设计要点

11.2 适用边界

11.3 架构演进方向

11.4 下一篇预告

所有评论(0)

温馨提示：您尚未绑定手机号

webajax