安装与基础用法
Ollama 官方提供了 Python 和 JavaScript 两个 SDK，Python 库的设计风格与 OpenAI SDK 相似，学习成本极低：
# 安装官方 Python 库
pip install ollama
# 或使用 uv（更快的包管理器）
uv add ollama
# 验证安装
python3 -c "import ollama; print(ollama.__version__)"import ollama
# ── 最简单的用法：generate ──────────────────────────────
response = ollama.generate(
model="llama3.2",
prompt="用一句话解释深度学习"
)
print(response["response"])
# ── chat 多轮对话 ────────────────────────────────────────
response = ollama.chat(
model="llama3.2",
messages=[
{"role": "system", "content": "你是一个 Python 专家"},
{"role": "user", "content": "什么是装饰器？"},
]
)
print(response["message"]["content"])
# ── 传入生成参数 ──────────────────────────────────────
response = ollama.generate(
model="qwen2.5:7b",
prompt="写一段快速排序的 Python 代码",
options={
"temperature": 0.1,
"num_ctx": 4096,
"seed": 42 # 固定种子，结果可复现
}
)
print(response["response"])
# ── 列出已安装模型 ────────────────────────────────────
models = ollama.list()
for model in models["models"]:
print(f"{model['name']:30s} {model['size'] / 1e9:.1f} GB")流式输出（stream=True）
流式输出让 AI 回复像打字机一样逐字出现，大幅提升用户体验：
import ollama
import sys
# ── 流式 generate ────────────────────────────────────
print("AI: ", end="", flush=True)
for chunk in ollama.generate(
model="llama3.2",
prompt="用100字介绍 Python 的历史",
stream=True
):
print(chunk["response"], end="", flush=True)
# chunk["done"] == True 时为最后一个 chunk
if chunk["done"]:
# 最后一个 chunk 包含性能统计
total_ms = chunk["total_duration"] / 1_000_000
tokens = chunk["eval_count"]
speed = tokens / (chunk["eval_duration"] / 1e9)
print(f"\n\n[统计] {tokens} tokens | {speed:.1f} tok/s | {total_ms:.0f}ms")
# ── 流式 chat ────────────────────────────────────────
messages = [{"role": "user", "content": "写一首关于编程的七言绝句"}]
print("AI: ", end="", flush=True)
for chunk in ollama.chat(
model="qwen2.5:7b",
messages=messages,
stream=True
):
content = chunk["message"]["content"]
print(content, end="", flush=True)
print()结构化输出（JSON 模式）
通过 format="json" 强制模型以 JSON 格式输出，便于程序解析：
import ollama
import json
from typing import TypedDict
# ── 基础 JSON 模式 ──────────────────────────────────────
response = ollama.generate(
model="qwen2.5:7b",
prompt="""
分析以下代码的质量，以 JSON 格式返回：
{
"score": 1-10的评分,
"issues": ["问题1", "问题2"],
"suggestions": ["建议1", "建议2"],
"summary": "一句话总结"
}
代码：
def f(x):
if x > 0:
return x * 2
return 0
""",
format="json",
options={"temperature": 0.1}
)
result = json.loads(response["response"])
print(f"评分：{result['score']}/10")
print(f"问题：{result['issues']}")
print(f"建议：{result['suggestions']}")
# ── 批量信息提取示例 ────────────────────────────────────
def extract_entities(text: str) -> dict:
"""从文本中提取实体信息（姓名、日期、地点等）。"""
response = ollama.generate(
model="qwen2.5:7b",
prompt=f"""从以下文本提取实体，以 JSON 返回：
{{
"persons": ["人名列表"],
"dates": ["日期列表"],
"locations": ["地点列表"],
"organizations": ["组织列表"]
}}
文本：{text}""",
format="json",
options={"temperature": 0}
)
return json.loads(response["response"])
sample_text = "2025年3月，张伟在北京参加了阿里巴巴举办的技术峰会"
entities = extract_entities(sample_text)
print(json.dumps(entities, ensure_ascii=False, indent=2))异步客户端（async/await）
在 Web 服务（FastAPI/aiohttp）等异步场景中，使用异步客户端避免阻塞事件循环：
import asyncio
import ollama
async def generate_async(prompt: str) -> str:
"""异步生成文本。"""
client = ollama.AsyncClient()
response = await client.generate(
model="llama3.2",
prompt=prompt
)
return response["response"]
async def stream_async(prompt: str):
"""异步流式生成。"""
client = ollama.AsyncClient()
async for chunk in await client.generate(
model="llama3.2",
prompt=prompt,
stream=True
):
print(chunk["response"], end="", flush=True)
print()
# 并发请求：同时处理多个问题
async def batch_generate(prompts: list[str]) -> list[str]:
"""并发生成多个响应。"""
client = ollama.AsyncClient()
tasks = [
client.generate(model="llama3.2", prompt=p)
for p in prompts
]
results = await asyncio.gather(*tasks)
return [r["response"] for r in results]
# 主程序
async def main():
questions = [
"什么是 Python GIL？",
"解释事件循环（Event Loop）",
"async/await 的原理是什么？"
]
answers = await batch_generate(questions)
for q, a in zip(questions, answers):
print(f"\nQ: {q}\nA: {a[:100]}...\n")
asyncio.run(main())多模态：图片理解
支持多模态的模型（LLaVA、Gemma 3、Llama 3.2-Vision）可以分析图片：
import ollama
import base64
from pathlib import Path
# 首先确保已安装多模态模型
# ollama pull llava:13b ← 通用视觉模型
# ollama pull llama3.2-vision ← Meta 视觉版
# ollama pull gemma3:12b ← Gemma 3 支持多模态
def analyze_image(image_path: str, question: str) -> str:
"""
分析图片并回答问题。
image_path: 本地图片文件路径
question: 关于图片的问题
"""
# 读取并转换为 base64
image_data = Path(image_path).read_bytes()
image_b64 = base64.b64encode(image_data).decode()
response = ollama.chat(
model="llava:13b",
messages=[{
"role": "user",
"content": question,
"images": [image_b64] # base64 字符串列表
}]
)
return response["message"]["content"]
# 使用示例
answer = analyze_image("./screenshot.png", "描述图中的内容，如果有代码请解释")
print(answer)
# 也可以直接传图片文件路径（更简洁）
response = ollama.chat(
model="llava:13b",
messages=[{
"role": "user",
"content": "这张图里有什么？",
"images": ["./photo.jpg"] # 文件路径（ollama 库自动读取）
}]
)完整实战：命令行聊天机器人
将本章所有知识整合，构建一个功能完整的命令行聊天机器人：
#!/usr/bin/env python3
"""
local_chat.py — 基于 Ollama 的命令行聊天机器人
功能：流式输出、历史记录、多模型切换、会话导出
"""
import ollama
import json
import sys
from datetime import datetime
from pathlib import Path
class LocalChat:
def __init__(self, model: str = "qwen2.5:7b", system: str = ""):
self.model = model
self.messages = []
if system:
self.messages.append({"role": "system", "content": system})
def chat(self, user_input: str) -> str:
"""发送消息，流式打印响应，返回完整文本。"""
self.messages.append({"role": "user", "content": user_input})
print(f"\n\033[92m{self.model}\033[0m: ", end="", flush=True)
full_response = ""
for chunk in ollama.chat(
model=self.model,
messages=self.messages,
stream=True
):
token = chunk["message"]["content"]
print(token, end="", flush=True)
full_response += token
print("\n")
self.messages.append({"role": "assistant", "content": full_response})
return full_response
def clear_history(self):
"""清除对话历史，保留 system 消息。"""
system_msgs = [m for m in self.messages if m["role"] == "system"]
self.messages = system_msgs
print("[历史已清除]")
def export_history(self, path: str = None):
"""导出对话历史到 JSON 文件。"""
if not path:
path = f"chat_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
Path(path).write_text(
json.dumps(self.messages, ensure_ascii=False, indent=2),
encoding="utf-8"
)
print(f"[已导出到 {path}]")
def run(self):
"""启动交互式会话。"""
print(f"=== 本地 AI 对话 [{self.model}] ===")
print("命令：/clear 清除历史 | /switch 切换模型 | /export 导出 | /quit 退出\n")
while True:
try:
user_input = input("\033[96m你\033[0m: ").strip()
except (EOFError, KeyboardInterrupt):
print("\n再见！")
break
if not user_input:
continue
elif user_input == "/quit":
print("再见！")
break
elif user_input == "/clear":
self.clear_history()
elif user_input == "/export":
self.export_history()
elif user_input.startswith("/switch "):
self.model = user_input.split()[1]
print(f"[已切换到 {self.model}]")
else:
try:
self.chat(user_input)
except Exception as e:
print(f"[错误] {e}")
if __name__ == "__main__":
model = sys.argv[1] if len(sys.argv) > 1 else "qwen2.5:7b"
bot = LocalChat(model=model, system="你是一个有帮助的中文助手")
bot.run()# 使用方式
python3 local_chat.py # 默认使用 qwen2.5:7b
python3 local_chat.py llama3.2 # 指定模型
python3 local_chat.py deepseek-r1:14b # 推理模型
生产集成建议
在 FastAPI 中集成 Ollama 时，使用
AsyncClient 配合 StreamingResponse 将流式 token 直接推送到前端。示例：return StreamingResponse(stream_generator(), media_type="text/event-stream")。避免在同步 FastAPI 路由中调用同步 ollama 函数，这会阻塞事件循环。
本章小结
Python 集成要点：安装
ollama 库后，ollama.generate() 单轮生成，ollama.chat() 多轮对话；stream=True 实现实时输出；format="json" 实现结构化输出；AsyncClient 用于异步场景；多模态传 images 参数。下一章将这些能力组合，构建完整的本地 RAG 系统。