跳至主要內容

Python 调用本地大模型实战——Ollama + Python 构建离线 AI 应用

老张2026/4/30大约 8 分钟

Python 调用本地大模型实战——Ollama + Python 构建离线 AI 应用

适读人群：想在本地或内网运行大模型的工程师 | 阅读时长：约17分钟 | 核心价值：从安装到生产，完整掌握 Ollama 的使用，构建完全离线的 AI 应用

小徐在一家医疗信息公司做开发，他们想给内部做一个 AI 助手，帮医生快速查询诊疗规范、药品信息。但医疗数据属于高度敏感信息，绝对不能传到任何云端 API。他们需要一个能在内网服务器上运行的解决方案。

他调研了一圈，看了本地部署 Hugging Face 模型的方案，发现配置复杂，CUDA 版本、模型格式、推理框架，各种依赖搞得团队头疼。有同事推荐他试试 Ollama，他问我靠不靠谱。

我说：Ollama 是目前本地部署大模型最简单的方案，没有之一。它把模型下载、推理服务、API 接口全部封装好了，三条命令就能跑起一个本地 AI 服务，而且兼容 OpenAI 的 API 格式，切换成本极低。

今天我们从零开始，把 Ollama + Python 的完整方案讲透。

一、Ollama 快速安装与模型管理

1.1 安装 Ollama

# macOS / Linux（一键安装）
curl -fsSL https://ollama.ai/install.sh | sh

# Windows：下载安装包
# https://ollama.ai/download

# 验证安装
ollama --version

1.2 常用模型管理命令

# 下载并运行 Qwen2.5（推荐，中文最强）
ollama run qwen2.5:7b

# 下载不运行
ollama pull qwen2.5:7b
ollama pull qwen2.5:14b        # 更大更强，需要更多内存
ollama pull llama3.2:3b        # Meta 的模型，英文强
ollama pull deepseek-r1:7b     # 推理能力强

# 查看已下载的模型
ollama list

# 删除模型
ollama rm qwen2.5:7b

# 查看运行中的模型
ollama ps

# 启动服务（后台运行，默认端口 11434）
ollama serve &

1.3 模型文件大小参考

模型大小内存要求速度（M1 Mac）
qwen2.5:0.5b394MB2GB~80 tokens/s
qwen2.5:1.5b986MB3GB~60 tokens/s
qwen2.5:7b4.7GB8GB~30 tokens/s
qwen2.5:14b9.0GB16GB~15 tokens/s
qwen2.5:32b20GB32GB~8 tokens/s

二、Python 接入 Ollama

2.1 使用 ollama Python 包

# pip install ollama

import ollama

# 最简单的调用
response = ollama.chat(
    model="qwen2.5:7b",
    messages=[
        {"role": "user", "content": "解释一下 Python 的装饰器"}
    ]
)
print(response["message"]["content"])

# 流式输出
def stream_chat(prompt: str, model: str = "qwen2.5:7b"):
    """流式输出，边生成边打印"""
    stream = ollama.chat(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        stream=True
    )
    
    full_response = ""
    for chunk in stream:
        text = chunk["message"]["content"]
        print(text, end="", flush=True)
        full_response += text
    
    print()  # 换行
    return full_response

stream_chat("用 Python 写一个二分查找函数，要有详细注释")

2.2 使用 OpenAI 兼容接口（无缝切换）

Ollama 提供了与 OpenAI API 完全兼容的接口，只需修改 base_url 即可：

from openai import OpenAI

# 关键：base_url 指向本地 Ollama 服务
client = OpenAI(
    base_url="http://localhost:11434/v1",
    api_key="ollama"  # 随便填，Ollama 不验证 key
)

def chat_with_local_model(
    messages: list,
    model: str = "qwen2.5:7b",
    temperature: float = 0.7,
    max_tokens: int = 2048,
    stream: bool = False
):
    """
    通过 OpenAI 兼容接口调用本地 Ollama 模型
    可以一键切换到云端 OpenAI（只需修改 client 的 base_url 和 api_key）
    """
    if stream:
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            temperature=temperature,
            max_tokens=max_tokens,
            stream=True
        )
        
        full_text = ""
        for chunk in response:
            text = chunk.choices[0].delta.content or ""
            print(text, end="", flush=True)
            full_text += text
        print()
        return full_text
    else:
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            temperature=temperature,
            max_tokens=max_tokens
        )
        return response.choices[0].message.content

# 多轮对话
messages = [
    {"role": "system", "content": "你是一个专业的医疗咨询助手，回答要专业准确。"},
]

def medical_chat(user_input: str) -> str:
    messages.append({"role": "user", "content": user_input})
    
    response = chat_with_local_model(messages, model="qwen2.5:7b", stream=True)
    
    messages.append({"role": "assistant", "content": response})
    return response

# 测试
print(medical_chat("常见的高血压症状有哪些？"))
print(medical_chat("那饮食上有什么需要注意的？"))  # 上下文保持

三、RAG + Ollama 构建离线知识库问答

from openai import OpenAI
from typing import List, Dict
import numpy as np
import json
import os

# 全部使用本地服务
ollama_client = OpenAI(
    base_url="http://localhost:11434/v1",
    api_key="ollama"
)

def get_local_embedding(text: str) -> List[float]:
    """
    使用 Ollama 的 nomic-embed-text 模型生成向量
    完全离线，无需任何 API key
    """
    response = ollama_client.embeddings.create(
        model="nomic-embed-text",  # ollama pull nomic-embed-text
        input=text
    )
    return response.data[0].embedding

class LocalKnowledgeBase:
    """基于 Ollama 的完全离线知识库"""
    
    def __init__(self, persist_path: str = "./local_kb.json"):
        self.persist_path = persist_path
        self.documents = []
        self.embeddings = []
        self._load_from_disk()
    
    def _load_from_disk(self):
        """从磁盘加载已有知识库"""
        if os.path.exists(self.persist_path):
            with open(self.persist_path, "r", encoding="utf-8") as f:
                data = json.load(f)
                self.documents = data["documents"]
                self.embeddings = data["embeddings"]
            print(f"加载知识库：{len(self.documents)} 个文档")
    
    def _save_to_disk(self):
        """持久化知识库"""
        with open(self.persist_path, "w", encoding="utf-8") as f:
            json.dump({
                "documents": self.documents,
                "embeddings": self.embeddings
            }, f, ensure_ascii=False)
    
    def add_documents(self, texts: List[str], metadatas: List[Dict] = None):
        """添加文档到知识库"""
        if metadatas is None:
            metadatas = [{} for _ in texts]
        
        for i, text in enumerate(texts):
            embedding = get_local_embedding(text)
            self.documents.append({"text": text, "metadata": metadatas[i]})
            self.embeddings.append(embedding)
            
            if (i + 1) % 10 == 0:
                print(f"已向量化 {i+1}/{len(texts)} 个文档")
        
        self._save_to_disk()
        print(f"知识库总计 {len(self.documents)} 个文档")
    
    def search(self, query: str, top_k: int = 5) -> List[Dict]:
        """语义搜索"""
        query_embedding = get_local_embedding(query)
        query_vec = np.array(query_embedding)
        
        similarities = []
        for emb in self.embeddings:
            emb_vec = np.array(emb)
            sim = np.dot(query_vec, emb_vec) / (np.linalg.norm(query_vec) * np.linalg.norm(emb_vec))
            similarities.append(sim)
        
        top_indices = np.argsort(similarities)[::-1][:top_k]
        
        return [
            {
                **self.documents[idx],
                "score": float(similarities[idx])
            }
            for idx in top_indices
            if similarities[idx] > 0.5  # 相似度阈值
        ]
    
    def rag_query(self, question: str, top_k: int = 3) -> str:
        """RAG 问答"""
        # 检索相关文档
        relevant_docs = self.search(question, top_k=top_k)
        
        if not relevant_docs:
            context = "暂无相关资料"
        else:
            context = "\n\n".join([
                f"[资料{i+1}] {doc['text']}"
                for i, doc in enumerate(relevant_docs)
            ])
        
        # 生成回答（完全离线）
        messages = [
            {
                "role": "system",
                "content": "你是一个专业助手，只根据提供的参考资料回答问题。"
                           "如果资料中没有相关信息，请明确说明。"
            },
            {
                "role": "user",
                "content": f"参考资料：\n{context}\n\n问题：{question}"
            }
        ]
        
        return chat_with_local_model(messages, model="qwen2.5:7b")

# 使用示例
kb = LocalKnowledgeBase("./medical_kb.json")

# 添加医疗知识
medical_docs = [
    "高血压是指在未使用降压药物的情况下，非同日三次测量血压，收缩压≥140mmHg和(或)舒张压≥90mmHg。",
    "高血压患者饮食原则：减少钠盐摄入（每日不超过6g），增加钾的摄入，限制饮酒，减少脂肪摄入。",
    "二甲双胍是2型糖尿病的一线用药，主要机制是抑制肝糖原分解，改善胰岛素抵抗。",
]

kb.add_documents(medical_docs)

# 离线问答
answer = kb.rag_query("高血压患者在饮食方面有哪些注意事项？")
print(answer)

四、踩坑实录一：Ollama 服务启动慢或连接超时

现象：ollama serve 后，第一次请求很慢，有时超时；或者换一个模型后很慢。

原因：模型首次加载需要把权重文件读入内存/显存，7B 模型大约需要 10-30 秒，14B 需要 30-60 秒。之后会保持模型在内存中（默认保持5分钟）。

解法：

import httpx
import time

def wait_for_ollama_ready(model: str = "qwen2.5:7b", timeout: int = 120):
    """等待 Ollama 服务和模型就绪"""
    print(f"等待模型 {model} 加载...")
    start = time.time()
    
    while time.time() - start < timeout:
        try:
            response = httpx.post(
                "http://localhost:11434/api/generate",
                json={"model": model, "prompt": "1+1=", "stream": False},
                timeout=10
            )
            if response.status_code == 200:
                elapsed = time.time() - start
                print(f"模型已就绪（耗时 {elapsed:.1f}s）")
                return True
        except (httpx.ConnectError, httpx.TimeoutException):
            pass
        
        time.sleep(2)
    
    raise TimeoutError(f"模型 {model} 加载超时（{timeout}s）")

# 应用启动时预加载模型
wait_for_ollama_ready("qwen2.5:7b")

五、踩坑实录二：模型被卸载出内存导致下次请求慢

现象：第一次请求很慢（模型加载），之后几分钟内请求很快，但间隔超过5分钟后又变慢了。

原因：Ollama 默认在模型空闲5分钟后自动卸载，下次请求需要重新加载。

解法：

import subprocess

# 方案一：通过环境变量修改保持时间
# 在启动 Ollama 时设置：OLLAMA_KEEP_ALIVE=30m ollama serve

# 方案二：通过 API 保持模型活跃
def keep_model_alive(model: str = "qwen2.5:7b"):
    """定期发送一个轻量请求，保持模型在内存中"""
    import threading
    
    def ping():
        while True:
            try:
                httpx.post(
                    "http://localhost:11434/api/generate",
                    json={"model": model, "prompt": "ping", "stream": False, "keep_alive": "30m"},
                    timeout=5
                )
            except Exception:
                pass
            time.sleep(4 * 60)  # 每4分钟 ping 一次
    
    thread = threading.Thread(target=ping, daemon=True)
    thread.start()

# 方案三：在 API 请求中直接指定 keep_alive
response = httpx.post(
    "http://localhost:11434/api/chat",
    json={
        "model": "qwen2.5:7b",
        "messages": [{"role": "user", "content": "hello"}],
        "keep_alive": "30m"  # 保持30分钟
    }
)

六、踩坑实录三：Modelfile 自定义失效

现象：创建了自定义 Modelfile（修改了 system prompt 和参数），但通过 API 调用时自定义的 system prompt 没有生效。

原因：通过 API 调用时传入的 system 字段会覆盖 Modelfile 中的 system prompt。

解法：

# Modelfile 中的配置会被 API 的 system 字段覆盖
# 正确做法：代码里统一管理 system prompt，不依赖 Modelfile

# 如果你想给不同应用场景设置不同的"默认人设"，可以封装成工厂函数
def create_model_client(persona: str = "general"):
    PERSONAS = {
        "general": "你是一个专业的 AI 助手。",
        "medical": "你是一个专业的医疗咨询助手，回答要严谨准确，遇到需要就医的情况要建议用户去医院。",
        "coder": "你是一个资深 Python 工程师，给出的代码必须能直接运行，有详细注释。",
    }
    
    system_prompt = PERSONAS.get(persona, PERSONAS["general"])
    
    def chat(messages: list, **kwargs):
        full_messages = [{"role": "system", "content": system_prompt}] + messages
        return chat_with_local_model(full_messages, **kwargs)
    
    return chat

medical_ai = create_model_client("medical")
response = medical_ai([{"role": "user", "content": "头痛怎么办？"}])

七、性能调优与硬件要求

# 查看 Ollama 的性能统计
def get_model_performance(model: str, test_prompt: str) -> dict:
    """测试模型生成速度"""
    import time
    
    start = time.time()
    response = httpx.post(
        "http://localhost:11434/api/generate",
        json={
            "model": model,
            "prompt": test_prompt,
            "stream": False
        },
        timeout=120
    ).json()
    elapsed = time.time() - start
    
    return {
        "model": model,
        "prompt_eval_duration_ms": response.get("prompt_eval_duration", 0) / 1e6,
        "eval_duration_ms": response.get("eval_duration", 0) / 1e6,
        "tokens_generated": response.get("eval_count", 0),
        "tokens_per_second": response.get("eval_count", 0) / (response.get("eval_duration", 1) / 1e9),
        "total_elapsed_s": elapsed
    }

# 实测（M1 Pro 16GB）
# qwen2.5:7b: ~28 tokens/s
# qwen2.5:14b: ~14 tokens/s
# llama3.2:3b: ~45 tokens/s

Ollama 是本地部署大模型的最低门槛方案，适合需要数据隐私保护、内网部署、降低长期 API 成本的场景。配合 Python 的 OpenAI 兼容接口，切换云端/本地成本极低。

编辑此页
最近更新:
贡献者: xiaozhangge