Uso de la API de Ollama en Python

在本文中，我们将简单介绍如何在 Python 中使用 Ollama API。无论你是想进行简单的聊天对话、使用流式响应处理大数据、还是希望在本地进行模型的创建、复制、删除等操作，本文都可以为你提供指导。此外，我们还展示了如何使用自定义客户端和异步编程来优化你的应用程序性能，无论你是 Ollama 的新手还是经验丰富的开发者，本文都能帮助你在 Python 中更高效地使用 Ollama API。

本教程还为大家提供了一个 Jupyter Notebook 示例，来让大家更好的学习。

Preparación medioambiental

在开始使用 Python 与 Ollama API 交互之前，请确保您的开发环境满足以下条件：

• Python: 安装 Python 3.8 或更高版本。
• pip: 确保已安装 pip，Python 的包管理工具。
• ollama 库: 用于更方便地与 Ollama API 交互。安装命令如下：

pip install ollama

Utilización

from ollama import chat
from ollama import ChatResponse
response: ChatResponse = chat(model='llama3.1', messages=[
{
'role': 'user',
'content': '为什么天空是蓝色的？',
},
])
print(response['message']['content'])
print(response.message.content)

respuesta en tiempo real

可以通过设置 stream=True 启用响应流，使函数调用返回一个 Python 生成器，其中每个部分都是流中的一个对象。

from ollama import chat
stream = chat(
model='llama3.1',
messages=[{'role': 'user', 'content': '为什么天空是蓝色的？'}],
stream=True,
)
for chunk in stream:
print(chunk['message']['content'], end='', flush=True)

Salida estructurada

• 普通输出（Unstructured Output）
  • 直接生成自然语言文本。
  • 适合人类阅读，但不便于程序解析或自动化处理。
  • Por ejemplo:

    这是一只黑色的小猫，它正在草地上玩耍。
    

• 结构化输出（Structured Output）
  • 以 JSON 、 YAML 、 XML 或其他格式返回数据，使其更容易被机器解析和使用。
  • 适合 API 、自动化工作流和数据存储。
  • Por ejemplo:

    {
    "description": "这是一只黑色的小猫",
    "activity": "正在草地上玩耍"
    }
    

结构化输出的优势

（1）便于处理

• 机器可以轻松提取特定字段，如 description tal vez activity ，而无需 NLP 解析普通文本。

（2）提高可控性

• 结构化格式让开发者可以精确控制模型输出，避免冗长或不可预测的回答。
• 例如，在 AI 生成代码时：

  {
  "language": "Python",
  "code": "print('Hello, World!')"
  }
  

（3）便于存储与分析

• 结构化数据更适合存储到数据库中，方便查询和分析。
• Por ejemplo:

  {
  "date": "2025-01-20",
  "summary": "今天的销售额增长了10%。"
  }
  

from pydantic import BaseModel, Field
from ollama import chat
import json
class CountryInfo(BaseModel):
capital: str = Field(..., alias="首都")
number: str = Field(..., alias="人口")
area: str = Field(..., alias="占地面积")  
response = chat(
model='llama3.1',
messages=[{
'role': 'user',
'content': "请介绍美国的首都、人口、占地面积信息，并以 JSON 格式返回。"
}],
format="json",  
options={'temperature': 0}, 
)
response_content = response["message"]["content"]
if not response_content:
raise ValueError("Ollama 返回的 JSON 为空")
json_response = json.loads(response_content)  
print(json_response)
friends_response = CountryInfo.model_validate(json_response)  
print(friends_response)

API

Ollama Python 库提供了丰富的接口，简化了与 Ollama 的交互。这些接口设计直观，易于集成，旨在帮助开发者更便捷地调用和管理模型。如果你想了解更详细的底层实现和完整的 API 端点信息，建议参考 Guía del usuario de la API de Ollama.

chats

ollama.chat(model='llama3.1', messages=[{'role': 'user', 'content': '为什么天空是蓝色的？'}])

generando

ollama.generate(model='llama3.1', prompt='为什么天空是蓝色的？')

本地模型列表

ollama.list()

Visualización de la información sobre el modelo

ollama.show('llama3.1')

创建模型

modelfile='''
FROM llama3.1
SYSTEM 你是超级马里奥兄弟中的马里奥。
'''
ollama.create(model='example', modelfile=modelfile)

Modelos de réplica

ollama.copy('llama3.1', 'user/llama3.1')

Borrar modelo

ollama.delete('llama3.1')

modelo pull

ollama.pull('llama3.1')

推送模型

ollama.push('user/llama3.1')

Generar incrustación

ollama.embeddings(model='llama3.1', prompt='天空是蓝色的因为瑞利散射')

# 批量生成embedding
ollama.embed(model='llama3.1', input=['天空是蓝色的', '草是绿色的'])

进程

ollama.ps()

自定义客户端

可以通过通过 ollama 实例化 Client tal vez AsyncClient 来创建自定义客户端。

可以使用以下字段创建自定义客户端：

• host: 要连接的 Ollama 主机
• timeout: 请求超时时间

所有关键字参数参见httpx.Client.

同步客户端

使用的是同步客户端 (Client) 意味着当你调用 client.chat() 方法时，程序会等待该请求完成并返回结果后才会继续执行后续代码。这种方式更直观、简单，适合于编写流程较为线性、不需要处理大量并发任务的应用。

from ollama import Client
client = Client(
host='http://localhost:11434',
headers={'x-some-header': 'some-value'}
)
response = client.chat(model='llama3.1', messages=[
{
'role': 'user',
'content': '为什么天空是蓝色的？',
},
])
print(response)

异步客户端

这段代码使用了异步客户端 (AsyncClient) ，并且定义了一个异步函数 chat() 。通过await关键字，可以暂停该函数的执行直到 AsyncClient().chat() 请求完成，但在此期间不会阻塞其他操作。这对于需要高效率处理 I/O 操作（如网络请求）或希望同时执行多个任务的应用来说非常有用。此外，使用 asyncio.run(chat()) 来运行这个异步函数。

import asyncio
from ollama import AsyncClient
import nest_asyncio
nest_asyncio.apply()
async def chat():
message = {'role': 'user', 'content': '为什么天空是蓝色的？'}
response = await AsyncClient().chat(model='llama3.1', messages=[message])
print(response)
asyncio.run(chat())

establecer stream=True 修改函数以返回 Python 异步生成器：

import asyncio
from ollama import AsyncClient
import nest_asyncio
nest_asyncio.apply()
async def chat():
message = {'role': 'user', 'content': '为什么天空是蓝色的？'}
async for part in await AsyncClient().chat(model='llama3.1', messages=[message], stream=True):
print(part['message']['content'], end='', flush=True)
asyncio.run(chat())

同步 & 异步客户端不同调用次数耗时对比测试

下面的这段代码分别调用同步和异步客户端重复 test_num 次问答过程，对比所需要的总时间和单次时间，用户可以更改以下的参数进行测试：

• test_messages: 测试数据
• test_num: 测试次数
• model_name: 测试模型

import time
import asyncio
from ollama import Client, AsyncClient
import nest_asyncio
# 应用nest_asyncio以支持Jupyter中的异步操作
nest_asyncio.apply()
# 初始化客户端
client = Client(host='http://localhost:11434')
async_client = AsyncClient(host='http://localhost:11434')
# 同步请求处理函数
def request_example(client, model_name, messages):
start_time = time.time()
try:
# 同步请求返回
response = client.chat(model=model_name, messages=messages)
except Exception as e:
print(f"同步请求失败: {e}")
response = None
end_time = time.time()
duration = end_time - start_time
print(f"同步请求时间: {duration}")
return response, duration
# 异步请求处理函数
async def async_request_example(client, model_name, messages):
start_time = time.time()
try:
# 异步请求返回
response = await client.chat(model=model_name, messages=messages)
except Exception as e:
print(f"异步请求失败: {e}")
response = None
end_time = time.time()
duration = end_time - start_time
print(f"异步请求时间: {duration}")
return response, duration
# 异步请求测试函数
async def async_client_test(test_num, model_name, messages):
tasks = [asyncio.create_task(async_request_example(async_client, model_name, messages)) 
for _ in range(test_num)]
results= await asyncio.gather(*tasks)
return results
# 运行同步测试
def sync_test(model_name, messages, test_num):
total_time = 0
for i in range(test_num):
_, duration = request_example(client, model_name, messages)
total_time += duration
return total_time / test_num
# 运行异步测试
async def async_test(model_name, messages, test_num):
start_time = time.time()
await async_client_test(test_num, model_name, messages)
end_time = time.time()
return (end_time - start_time) / test_num
# 准备测试数据
test_messages = [{'role': 'user', 'content': '为什么天空是蓝色的？'}]
test_num = 10
model_name = 'llama3.1'
# 运行同步测试并输出结果
print("运行同步测试")
sync_avg_time = sync_test(model_name, test_messages, test_num)
print(f"同步测试平均时间: {sync_avg_time:.2f} 秒")
# 运行异步测试并输出结果
print("运行异步测试")
async_avg_time = asyncio.run(async_test(model_name, test_messages, test_num))
print(f"异步测试平均时间: {async_avg_time:.2f} 秒")

错误

如果请求返回错误状态或在流式传输时检测到错误，则会引发错误。

import ollama
model = 'does-not-yet-exist'
try:
ollama.chat(model)
except ollama.ResponseError as e:
print('错误:', e.error)
if e.status_code == 404:
ollama.pull(model)

Consulte la documentación:Ollama python