企业级AI知识库构建指南：基于ollama-python与Django的本地化解决方案

2026-03-16 04:47:32作者：宣海椒Queenly

一、企业知识管理的困境与破局之道

当企业积累了海量文档却无法快速检索有效信息时，当客服团队反复回答相同问题导致效率低下时，当敏感数据因使用云端AI服务而面临泄露风险时——你是否意识到传统知识管理系统已无法满足智能化需求？LLM（大语言模型，可理解为能理解人类语言的AI系统）技术的成熟为企业知识管理带来了新可能，但如何在保障数据安全的前提下实现本地化部署？

ollama-python作为Ollama服务的Python客户端，为这一问题提供了理想答案。它允许开发者直接在企业内网部署大语言模型，构建专属的AI知识库。与传统检索系统相比，其优势体现在：

特性	传统关键词检索	LLM驱动的知识库
理解能力	基于字符串匹配，无法理解语义	理解上下文和意图，支持自然语言提问
知识整合	分散存储，需手动关联	自动关联相关知识，生成综合回答
更新维护	需手动更新索引	支持增量学习，动态更新知识
部署成本	低但功能有限	适中但功能全面

二、技术原理：ollama-python如何驱动本地知识库

核心架构解析

ollama-python的核心价值在于将复杂的LLM交互封装为简洁的API。其架构包含三个关键模块：

客户端层：ollama/_client.py实现了同步/异步两种调用模式，支持流式响应和批量处理
数据类型层：ollama/_types.py定义了ChatRequest、Embedding等数据结构，确保类型安全
工具函数层：ollama/_utils.py提供embedding计算、工具调用等辅助功能

知识检索的工作流

基于ollama-python构建的知识库系统通常包含以下流程：

知识预处理：将文档转换为向量存储（使用examples/embed.py中的Embedding功能）
用户提问：接收自然语言查询并转换为向量
相似检索：在向量数据库中查找相关知识片段
内容生成：将检索结果作为上下文传入LLM生成回答

三、实践案例：构建企业产品知识库

环境准备

首先部署Ollama服务并安装客户端：

# 安装Ollama服务
curl -fsSL https://ollama.com/install.sh | sh

# 拉取适合知识问答的模型（约3.8GB）
ollama run llama3:8b

# 创建Python虚拟环境
python -m venv venv
source venv/bin/activate  # Linux/Mac
venv\Scripts\activate     # Windows

# 安装依赖
pip install django ollama python-dotenv

项目搭建

创建Django项目与应用：

django-admin startproject enterprise_ai
cd enterprise_ai
python manage.py startapp knowledge_base

核心实现

1. 配置文件设置

创建.env文件存储敏感配置：

# .env文件
OLLAMA_HOST=http://localhost:11434
DEFAULT_MODEL=llama3:8b
VECTOR_DB_PATH=./vector_db

修改Django设置：

# enterprise_ai/settings.py
import os
from pathlib import Path
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

INSTALLED_APPS = [
    # ...默认应用
    'knowledge_base',
]

# Ollama配置
OLLAMA_HOST = os.getenv('OLLAMA_HOST', 'http://localhost:11434')
DEFAULT_MODEL = os.getenv('DEFAULT_MODEL', 'llama3:8b')
VECTOR_DB_PATH = Path(os.getenv('VECTOR_DB_PATH', './vector_db'))

2. 向量存储服务

创建向量存储服务类：

# knowledge_base/services/vector_service.py
from ollama import Client
from django.conf import settings
from pathlib import Path
import numpy as np
import json
import os

class VectorStoreService:
    def __init__(self):
        self.client = Client(host=settings.OLLAMA_HOST)
        self.db_path = settings.VECTOR_DB_PATH
        # 创建存储目录
        self.db_path.mkdir(exist_ok=True)
    
    def embed_text(self, text: str) -> list[float]:
        """将文本转换为向量"""
        response = self.client.embeddings(model=settings.DEFAULT_MODEL, prompt=text)
        return response["embedding"]
    
    def store_embedding(self, text: str, metadata: dict = None) -> str:
        """存储文本向量及元数据"""
        embedding = self.embed_text(text)
        # 生成唯一ID
        doc_id = f"doc_{hash(text)}_{os.urandom(4).hex()}"
        doc_path = self.db_path / f"{doc_id}.json"
        
        # 存储数据
        with open(doc_path, 'w', encoding='utf-8') as f:
            json.dump({
                "text": text,
                "embedding": embedding,
                "metadata": metadata or {},
                "created_at": str(datetime.now())
            }, f)
        
        return doc_id
    
    def search_similar(self, query: str, top_k: int = 3) -> list[dict]:
        """搜索相似文本"""
        query_embedding = self.embed_text(query)
        results = []
        
        # 遍历所有文档计算相似度
        for doc_file in self.db_path.glob("*.json"):
            with open(doc_file, 'r', encoding='utf-8') as f:
                doc = json.load(f)
            
            # 计算余弦相似度
            similarity = np.dot(query_embedding, doc["embedding"]) / (
                np.linalg.norm(query_embedding) * np.linalg.norm(doc["embedding"])
            )
            
            results.append({
                "id": doc_file.stem,
                "text": doc["text"],
                "metadata": doc["metadata"],
                "similarity": float(similarity)
            })
        
        # 按相似度排序并返回top_k结果
        return sorted(results, key=lambda x: x["similarity"], reverse=True)[:top_k]

3. 知识问答视图

实现异步API视图：

# knowledge_base/views.py
from django.http import JsonResponse
from django.views import View
from django.utils.decorators import method_decorator
from django.views.decorators.csrf import csrf_exempt
import json
import asyncio
from .services.vector_service import VectorStoreService
from .services.ollama_service import OllamaService

@method_decorator(csrf_exempt, name='dispatch')
class KnowledgeBaseAPI(View):
    async def post(self, request):
        try:
            data = json.loads(request.body)
            query = data.get('query')
            
            if not query:
                return JsonResponse({'error': '查询内容不能为空'}, status=400)
            
            # 1. 检索相似知识
            vector_service = VectorStoreService()
            similar_docs = vector_service.search_similar(query)
            
            # 2. 构建提示词
            context = "\n\n".join([doc["text"] for doc in similar_docs])
            prompt = f"""基于以下上下文回答问题：
{context}

问题：{query}

要求：
1. 仅使用提供的上下文信息回答
2. 保持回答简洁准确
3. 如果上下文没有相关信息，回答"没有找到相关知识"
"""
            
            # 3. 调用LLM生成回答
            ollama_service = OllamaService()
            response = await ollama_service.async_chat_completion(
                model=settings.DEFAULT_MODEL,
                messages=[{"role": "user", "content": prompt}]
            )
            
            return JsonResponse({
                "answer": response,
                "sources": [{"id": doc["id"], "similarity": doc["similarity"]} for doc in similar_docs]
            })
            
        except Exception as e:
            return JsonResponse({'error': str(e)}, status=500)

4. URL配置

# enterprise_ai/urls.py
from django.contrib import admin
from django.urls import path
from knowledge_base.views import KnowledgeBaseAPI, DocumentUploadAPI

urlpatterns = [
    path('admin/', admin.site.urls),
    path('api/knowledge/query/', KnowledgeBaseAPI.as_view(), name='knowledge_query'),
    path('api/knowledge/upload/', DocumentUploadAPI.as_view(), name='document_upload'),
]

关键功能实现

文档上传与处理

# knowledge_base/views.py (添加文档上传视图)
class DocumentUploadAPI(View):
    @method_decorator(csrf_exempt)
    def post(self, request):
        if 'file' not in request.FILES:
            return JsonResponse({'error': '未找到文件'}, status=400)
            
        file = request.FILES['file']
        metadata = json.loads(request.POST.get('metadata', '{}'))
        
        # 读取文件内容（此处简化处理，实际应根据文件类型使用不同解析方式）
        content = file.read().decode('utf-8')
        
        # 存储到向量数据库
        vector_service = VectorStoreService()
        doc_id = vector_service.store_embedding(content, metadata)
        
        return JsonResponse({'doc_id': doc_id, 'status': 'success'})

异步Ollama服务

# knowledge_base/services/ollama_service.py
from ollama import AsyncClient
from django.conf import settings

class OllamaService:
    def __init__(self):
        self.host = settings.OLLAMA_HOST
        self.default_model = settings.DEFAULT_MODEL
    
    async def async_chat_completion(self, model: str = None, messages: list = None):
        """异步调用Ollama聊天接口"""
        model = model or self.default_model
        messages = messages or []
        
        try:
            async with AsyncClient(host=self.host) as client:
                response = await client.chat(model=model, messages=messages)
                return response['message']['content']
        except Exception as e:
            return f"服务调用失败: {str(e)}"