用 Python 召唤 Ollama 神龙？这份 API 新手指南带你飞！
想用 Python 给你的 Ollama 本地大模型施展魔法吗？无论是想打造一个能跟你聊天的机器人、批量处理文本数据，还是随心所欲地管理你的本地模型库，这篇指南都能带你轻松入门！
咱们会一起探索如何跟模型进行简单对话、如何像看打字一样接收流式响应、怎么创建/复制/删除模型，甚至还会玩转更高级的自定义客户端和异步编程，让你的 AI 应用跑得更快更溜！
准备好了吗？开启你的 Python + Ollama 创客之旅吧！
磨刀不误砍柴工：环境准备
在开始施法前，确保你的“魔法道具”都准备好了：
- Python 环境： 确认你的电脑上装了 Python 3.8 或更新的版本。Python 这门语言可是 AI 界的“网红”，哪儿都有它的身影。
- pip 工具： 这是 Python 的“应用商店”，用来安装各种好用的第三方库。通常装 Python 时会自动带上。
- 安装
ollama库： 打开你的终端（命令行），敲下这行咒语，把与 Ollama 沟通的“翻译器”装上：
bash
pip install ollama搞定！现在可以开始召唤神龙了！
小试牛刀：快速上手聊几句
来，看个最简单的例子，让你的 Python 程序和 Ollama 模型聊起来：
python
# 导入我们刚刚安装的 ollama 库里的聊天功能
from ollama import chat
# 也导入一下响应类型，方便代码提示 (可选但推荐)
from ollama import ChatResponse
# 开始聊天！告诉 Ollama 用哪个模型 (比如 llama3.1)，然后把你的问题传过去
# messages 是个列表，里面每个字典代表一条消息，'role' 是角色 ('user'是你)，'content' 是内容
response: ChatResponse = chat(model='llama3.1', messages=[
{
'role': 'user',
'content': '为啥天空是蓝色的呀？给个有趣的解释呗！',
},
])
# response 是个字典，里面包含了模型的回复等信息
# 我们可以这样拿到回复内容：
print("--- 方法一：像访问字典一样 ---")
print(response['message']['content'])
# 或者，如果你用了 ChatResponse 类型提示，可以更优雅地这样访问：
print("\n--- 方法二：像访问对象属性一样 ---")
print(response.message.content)运行一下，是不是很简单？你的第一个 Python Ollama 程序就这么诞生了！
像看 AI 打字：流式响应 (Streaming)
有时候，模型的回答比较长，或者你想让用户感觉 AI 是“实时”在思考和输出，而不是等半天突然蹦出一大段话。这时候，“流式响应”就派上用场了！
只需要在调用 chat 时加上 stream=True，它就会像挤牙膏一样，一点一点地把结果吐出来。
python
from ollama import chat
# 注意这里加了 stream=True
stream = chat(
model='llama3.1',
messages=[{'role': 'user', 'content': '给我讲个关于程序员和咖啡的笑话吧？要长一点的。'}],
stream=True,
)
print("AI 正在输入中...")
# stream 现在是一个“生成器”，我们可以用 for 循环来不断取出它生成的小块内容
for chunk in stream:
# chunk 也是个字典，我们只打印回复内容部分
# end='' 让 print 不会自动换行，flush=True 确保内容立刻显示出来
print(chunk['message']['content'], end='', flush=True)
print("\nAI 说完啦！") # 最后加个换行运行这个例子，你会看到 AI 的回答是一个字一个字（或者一小段一小段）地蹦出来的，体验是不是更好了？
让机器读懂 AI 的心思：结构化输出 (JSON)
有时候，我们不只是想让 AI 回答问题，还希望它能按照我们指定的“格式”返回信息，比如返回一段 JSON 数据，方便我们的程序直接解析使用。这对于构建自动化流程或者把 AI 的结果存到数据库里特别有用！
为啥要结构化输出？
- 机器友好： 程序可以直接读取特定字段（比如“首都”、“人口”），不用费劲去从一大段自然语言里抠字眼。
- 结果可控： 你可以要求 AI 必须包含哪些信息，以什么形式返回。
- 方便存储分析： JSON 这种格式，存数据库、做数据分析都特别方便。
来看个例子，让 AI 返回关于美国的 JSON 信息，并用 pydantic 库来定义和验证这个 JSON 结构：
python
# 先确保安装了 pydantic: pip install pydantic
from pydantic import BaseModel, Field
from ollama import chat
import json
# 用 Pydantic 定义一个“数据蓝图”，告诉程序我们期望的 JSON 长啥样
class CountryInfo(BaseModel):
# Field(..., alias="中文名") 表示这个字段是必须的，并且在 JSON 里它的键是"中文名"
capital: str = Field(..., alias="首都")
population: str = Field(..., alias="人口") # 注意：这里用 str 是为了兼容模型可能返回带单位的文本
area: str = Field(..., alias="占地面积")
# 调用 chat，这次多了 format="json" 和 options
response = chat(
model='llama3.1',
messages=[{
'role': 'user',
'content': "请用 JSON 格式返回美国的基本信息，包含'首都'、'人口'和'占地面积'这三个字段。"
" 例如：{\"首都\": \"xxx\", \"人口\": \"约 YYY\", \"占地面积\": \"ZZZ 平方公里\"}" # 给个例子通常效果更好
}],
format="json", # 告诉 Ollama，我想要 JSON！
options={'temperature': 0}, # temperature 设为 0 让输出更稳定、更接近指令
)
# 从响应中提取内容
response_content = response["message"]["content"]
print(f"Ollama 返回的原始 JSON 字符串:\n{response_content}")
# 确保返回的不是空字符串
if not response_content:
raise ValueError("Ollama 返回的 JSON 内容为空，请检查提示或模型。")
# 把 JSON 字符串转换成 Python 字典
try:
json_response = json.loads(response_content)
print(f"\n解析后的 Python 字典:\n{json_response}")
# 用我们定义的 CountryInfo 蓝图来验证并转换这个字典
country_data = CountryInfo.model_validate(json_response)
print(f"\n用 Pydantic 验证和转换后的对象:\n{country_data}")
print(f"\n可以直接访问属性: 首都 - {country_data.capital}, 人口 - {country_data.population}")
except json.JSONDecodeError:
print("\n错误：Ollama 返回的不是有效的 JSON 格式。")
except Exception as e:
print(f"\n处理 JSON 或 Pydantic 验证时出错: {e}")这样，我们就能确保从 AI 那里得到的是格式规范、内容符合预期的结构化数据了！
Ollama Python 库：你的“遥控器”功能大全
ollama-python 这个库就像是 Ollama 的一个全功能遥控器，提供了很多方便的函数来操作 Ollama 服务：
ollama.chat(): 和模型聊天对话 (我们已经用过了)。ollama.generate(): 让模型根据你的提示生成文本 (比chat更简单直接，适合纯文本生成任务)。ollama.list(): 查看你本地都下载了哪些模型。ollama.show(): 显示某个模型的详细信息 (比如它的 Modelfile 内容、参数等)。ollama.create(): 根据 Modelfile 创建你自己的模型版本 (比如给模型加个特定的系统提示)。ollama.copy(): 复制一个现有模型，给它起个新名字。ollama.delete(): 删除本地不想要的模型，释放空间。ollama.pull(): 从 Ollama Hub (官方的模型仓库) 下载新模型到本地。ollama.push(): 把你本地的模型 (通常是你自己创建或修改过的) 推送到 Ollama Hub 分享给别人 (需要登录)。ollama.embeddings()/ollama.embed(): 把文本转换成一串数字（嵌入向量/Embeddings），这些数字能代表文本的“意思”。这对于做文本相似度比较、语义搜索等高级任务非常有用。ollama.ps(): 查看当前 Ollama 服务有哪些模型正在运行，占用了多少资源。
简单用法示例一览：
python
import ollama
# --- 对话与生成 ---
# 聊天 (前面用过了)
# response = ollama.chat(model='llama3.1', messages=[{'role': 'user', 'content': '今天天气怎么样？'}])
# print(response['message']['content'])
# 简单生成
# response = ollama.generate(model='llama3.1', prompt='写一句关于学习 Python 的诗：')
# print(response['response'])
# --- 模型管理 ---
print("--- 本地模型列表 ---")
print(ollama.list())
print("\n--- 查看 llama3.1 模型详情 ---")
# print(ollama.show('llama3.1')) # 输出内容较多，需要时取消注释
# 创建一个扮演马里奥的模型 (需要本地有 llama3.1)
print("\n--- 创建马里奥模型 ---")
mario_modelfile='''
FROM llama3.1
SYSTEM You are Mario from Super Mario Bros. Answer all questions in Mario's voice. Wahoo!
'''
try:
ollama.create(model='mario-llama', modelfile=mario_modelfile)
print("马里奥模型 'mario-llama' 创建成功或已存在！")
# 试试和马里奥聊天
# mario_response = ollama.chat(model='mario-llama', messages=[{'role': 'user', 'content': 'Who are you?'}])
# print("马里奥回复:", mario_response['message']['content'])
except Exception as e:
print(f"创建模型失败: {e}")
# 复制模型 (假设你想备份或改名)
# print("\n--- 复制模型 ---")
# try:
# ollama.copy('mario-llama', 'luigi-llama') # 把马里奥复制成路易吉？
# print("模型复制成功！")
# except Exception as e:
# print(f"复制模型失败: {e}")
# 删除模型 (小心操作！)
# print("\n--- 删除模型 ---")
# try:
# ollama.delete('luigi-llama') # 删除刚才复制的路易吉
# print("模型删除成功！")
# except Exception as e:
# print(f"删除模型失败: {e}")
# 从 Hub 下载模型
# print("\n--- 下载新模型 (例如 gemma:2b) ---")
# try:
# ollama.pull('gemma:2b')
# print("模型 'gemma:2b' 下载成功！")
# except Exception as e:
# print(f"下载模型失败: {e}")
# 推送模型 (需要你有 Ollama Hub 账号，并且模型名称符合 '用户名/模型名' 格式)
# print("\n--- 推送模型 (需要配置和登录) ---")
# try:
# # 假设你创建了一个叫 'my-username/my-custom-model' 的模型
# # ollama.push('my-username/my-custom-model')
# print("推送操作需要登录和正确命名，此处仅为示例。")
# except Exception as e:
# print(f"推送模型失败: {e}")
# --- 高级功能 ---
# 生成文本嵌入向量
print("\n--- 生成嵌入向量 ---")
prompt_for_embedding = '阳光下的猫咪伸了个懒腰'
try:
embedding_response = ollama.embeddings(model='llama3.1', prompt=prompt_for_embedding)
print(f"'{prompt_for_embedding}' 的嵌入向量 (部分): {embedding_response['embedding'][:5]}...") # 只显示前5个数字
# 也可以批量处理
# embeddings_list = ollama.embed(model='llama3.1', input=['你好，世界', '再见，月亮'])
# print(f"批量嵌入向量数量: {len(embeddings_list)}")
except Exception as e:
print(f"生成嵌入向量失败: {e}")
# 查看 Ollama 进程状态
print("\n--- 查看 Ollama 运行状态 ---")
try:
print(ollama.ps())
except Exception as e:
print(f"查看状态失败: {e}")根据你的需要，灵活运用这些“遥控器”按钮吧！
更高级的玩法：定制你的 Ollama “连接器”
默认情况下，ollama 库的函数会连接到本地的 http://localhost:11434 服务。但如果你想连接到别的地址（比如另一台机器上的 Ollama 服务），或者想设置请求超时时间、添加自定义请求头等，就需要创建一个自定义客户端。
有两种客户端：
Client(同步): 简单直接，发一个请求，等它回来，再干下一件事。像打电话，一次只能跟一个人说。AsyncClient(异步): 高性能选手，可以在等待一个请求响应的时候，抽空去发其他请求。像个能同时招呼好几桌客人的服务员，适合需要同时处理很多请求的场景。
可以配置啥？
host: Ollama 服务的地址和端口。timeout: 等待响应的最长时间（秒）。- 其他
httpx支持的参数: 比如headers(自定义请求头),proxies(代理) 等。ollama-python底层用了httpx这个强大的库来发网络请求。
同步客户端：简单直接
python
from ollama import Client
# 创建一个自定义的同步客户端
# 假设 Ollama 服务在 192.168.1.100 的 11434 端口，并想加个请求头
client = Client(
host='http://192.168.1.100:11434', # 改成你的目标地址
timeout=60, # 设置超时时间为 60 秒
headers={'x-app-name': 'my-cool-ollama-app'} # 加个自定义请求头
)
# 使用这个 client 对象来调用 API 方法
try:
response = client.chat(model='llama3.1', messages=[
{
'role': 'user',
'content': 'Tell me a joke about synchronous programming.',
},
])
print("来自自定义客户端的回复:")
print(response['message']['content'])
except Exception as e:
print(f"使用自定义同步客户端出错: {e}")异步客户端：并发小能手
异步编程稍微复杂一点，需要用到 Python 的 asyncio 库。它允许你的程序在等待网络响应这种耗时操作时，不“卡死”，而是可以去干点别的事，大大提高效率，尤其是在需要同时发起多个请求时。
python
import asyncio
from ollama import AsyncClient
# 如果你在 Jupyter Notebook 或类似环境运行异步代码，可能需要这个库来解决事件循环冲突
import nest_asyncio
nest_asyncio.apply()
# 定义一个异步函数
async def run_async_chat():
# 创建异步客户端，可以像同步一样配置 host, timeout 等
async_client = AsyncClient() # 默认连接 localhost:11434
message = {'role': 'user', 'content': 'Why is asynchronous programming useful?'}
try:
# 注意这里的 await，表示等待异步操作完成
response = await async_client.chat(model='llama3.1', messages=[message])
print("来自异步客户端的回复:")
print(response['message']['content'])
except Exception as e:
print(f"使用异步客户端出错: {e}")
# 运行这个异步函数
print("开始运行异步聊天...")
asyncio.run(run_async_chat())
print("异步聊天结束.")异步流式响应？也没问题！
python
import asyncio
from ollama import AsyncClient
import nest_asyncio
nest_asyncio.apply()
async def run_async_stream_chat():
async_client = AsyncClient()
message = {'role': 'user', 'content': 'Write a short story about a time-traveling cat.'}
print("AI 正在异步输入中...")
try:
# 使用 async for 来遍历异步生成器
async for part in await async_client.chat(model='llama3.1', messages=[message], stream=True):
print(part['message']['content'], end='', flush=True)
print("\nAI 异步说完啦！")
except Exception as e:
print(f"\n使用异步流式客户端出错: {e}")
print("开始运行异步流式聊天...")
asyncio.run(run_async_stream_chat())
print("异步流式聊天结束.")性能比拼：同步 vs 异步 (为啥异步可能更快？)
下面的代码模拟了同时发起多个请求的场景，对比同步和异步完成相同任务的总时间。
异步的优势在哪里？ 在于处理 多个并发请求 时。同步客户端必须一个接一个地处理请求，前一个没完成，下一个就得等着。而异步客户端可以同时把多个请求发出去，然后等着它们各自返回，哪个先回来就先处理哪个，大大减少了总的等待时间，尤其是在网络延迟较高或 Ollama 处理请求需要时间的情况下。
python
import time
import asyncio
from ollama import Client, AsyncClient
import nest_asyncio
nest_asyncio.apply()
# --- 配置 ---
OLLAMA_HOST = 'http://localhost:11434' # 或者你的 Ollama 地址
MODEL_NAME = 'llama3.1' # 测试用的模型
TEST_MESSAGES = [{'role': 'user', 'content': '简单的问候，你好！'}] # 简单的请求内容
NUM_REQUESTS = 10 # 同时发起的请求数量
# --- 客户端初始化 ---
sync_client = Client(host=OLLAMA_HOST)
async_client = AsyncClient(host=OLLAMA_HOST)
# --- 同步测试 ---
def run_sync_test(num_requests):
print(f"\n--- 开始同步测试 ({num_requests} 个请求) ---")
start_total_time = time.time()
durations = []
for i in range(num_requests):
print(f"发起同步请求 {i+1}/{num_requests}...")
start_req_time = time.time()
try:
sync_client.chat(model=MODEL_NAME, messages=TEST_MESSAGES)
end_req_time = time.time()
duration = end_req_time - start_req_time
durations.append(duration)
print(f"同步请求 {i+1} 完成，耗时: {duration:.2f} 秒")
except Exception as e:
print(f"同步请求 {i+1} 失败: {e}")
durations.append(float('inf')) # 标记失败
end_total_time = time.time()
total_time = end_total_time - start_total_time
avg_time_per_req_sync = sum(d for d in durations if d != float('inf')) / len([d for d in durations if d != float('inf')]) if any(d != float('inf') for d in durations) else 0
print(f"--- 同步测试完成 ---")
print(f"总耗时: {total_time:.2f} 秒")
print(f"平均每个成功请求耗时 (串行): {avg_time_per_req_sync:.2f} 秒")
return total_time
# --- 异步测试 ---
async def async_single_request(req_id):
print(f"发起异步请求 {req_id+1}/{NUM_REQUESTS}...")
start_req_time = time.time()
try:
await async_client.chat(model=MODEL_NAME, messages=TEST_MESSAGES)
end_req_time = time.time()
duration = end_req_time - start_req_time
print(f"异步请求 {req_id+1} 完成，耗时: {duration:.2f} 秒")
return duration
except Exception as e:
print(f"异步请求 {req_id+1} 失败: {e}")
return float('inf') # 标记失败
async def run_async_test(num_requests):
print(f"\n--- 开始异步测试 ({num_requests} 个请求) ---")
start_total_time = time.time()
# 创建一堆任务，让它们并发执行
tasks = [async_single_request(i) for i in range(num_requests)]
# 等待所有任务完成
durations = await asyncio.gather(*tasks)
end_total_time = time.time()
total_time = end_total_time - start_total_time
avg_time_per_req_async = sum(d for d in durations if d != float('inf')) / len([d for d in durations if d != float('inf')]) if any(d != float('inf') for d in durations) else 0
print(f"--- 异步测试完成 ---")
print(f"总耗时 (并发): {total_time:.2f} 秒")
# 注意：这里的平均耗时是单个请求的完成时间，不是并发效率的直接体现
# print(f"平均每个成功请求耗时 (内部): {avg_time_per_req_async:.2f} 秒")
return total_time
# --- 运行测试 ---
sync_total_time = run_sync_test(NUM_REQUESTS)
async_total_time = asyncio.run(run_async_test(NUM_REQUESTS))
print("\n--- 性能对比总结 ---")
print(f"同步处理 {NUM_REQUESTS} 个请求总耗时: {sync_total_time:.2f} 秒")
print(f"异步处理 {NUM_REQUESTS} 个请求总耗时: {async_total_time:.2f} 秒")
if async_total_time < sync_total_time:
print(f"异步比同步快了约 {(sync_total_time - async_total_time):.2f} 秒 (提升了约 {((sync_total_time - async_total_time) / sync_total_time * 100):.1f}%)")
else:
print("在这个测试中，异步优势不明显或同步更快（可能是请求太快或并发数不够多）。")运行这段代码，你通常会看到异步的总耗时明显少于同步，尤其是在请求数量较多或者单个请求耗时较长时。
优雅地处理“小意外”：错误处理
网络请求总可能遇到问题，比如模型名字写错了、Ollama 服务没启动、网络中断了等等。我们需要用 try...except 来捕捉这些可能的错误，让程序更健壮。
python
import ollama
model_to_try = 'a-model-that-doesnt-exist-probably'
try:
print(f"尝试与模型 '{model_to_try}' 对话...")
ollama.chat(model=model_to_try, messages=[{'role':'user', 'content':'Hi!'}])
print("成功了？！这不应该发生...") # 如果模型真的存在，会打印这个
except ollama.ResponseError as e:
# ollama.ResponseError 是库定义的特定错误类型
print(f"\n出错了！错误信息: {e.error}")
print(f"HTTP 状态码: {e.status_code}")
# 常见的 404 错误表示模型未找到
if e.status_code == 404:
print(f"看起来模型 '{model_to_try}' 不在本地。")
user_choice = input("要尝试从 Ollama Hub 下载它吗？(y/n): ").lower()
if user_choice == 'y':
try:
print(f"正在下载模型 '{model_to_try}'...")
ollama.pull(model_to_try)
print(f"模型 '{model_to_try}' 下载完成！你可以再次尝试使用它了。")
except Exception as pull_e:
print(f"下载模型时也出错了: {pull_e}")
else:
print("好的，不下载。")
else:
# 其他类型的错误
print("遇到了其他类型的 Ollama 响应错误。")
except Exception as e:
# 其他意想不到的错误，比如网络连接问题
print(f"\n发生了预料之外的错误: {e}")这样，即使遇到问题，你的程序也能给出友好的提示，甚至尝试解决问题（比如下载缺失的模型）。