LV010-调用本地LLM

Ollama 提供了 Python SDK，可以让我们能够在 Python 环境中与本地运行的模型进行交互。通过 Ollama 的 Python SDK 能够轻松地将自然语言处理任务集成到 Python 项目中，执行各种操作，如文本生成、对话生成、模型管理等，且不需要手动调用命令行。

一、准备工作

1. Python安装

这里就不赘述了，我是windows下安装的Python：

D:\sumu_blog\python-ai> python -V
Python 3.8.6

2. Python SDK

我们需要安装 Ollama 的 Python SDK。可以使用 pip 安装：

pip install ollama

Tips：确保你的环境中已安装了 Python 3.x，并且网络环境能够访问 Ollama 本地服务。

安装完毕可以通过下面的命令查看版本：

pip show ollama

【例】

D:\sumu_blog\python-ai> pip show ollama
Name: ollama
Version: 0.6.0
Summary: The official Python client for Ollama.
Home-page: None
Author: None
Author-email: hello@ollama.com
License: None
Location: c:\program files\python38\lib\site-packages
Requires: pydantic, httpx
Required-by:

3. 启动ollama服务

ollama serve

二、Ollama SDK的应用

这里有一个文档可以作为参考：接口 Endpoints - Ollama 官方中文文档

1. Ollama 的 Python SDK 推理

安装了 SDK 并启动了本地服务后，我们就可以通过 Python 代码与 Ollama 进行交互。首先，从 ollama 库中导入 chat 和 ChatResponse：

from ollama import chat
from ollama import ChatResponse

通过 Python SDK，我们可以向指定的模型发送请求，生成文本或对话：

# -*- coding: utf-8 -*-
from ollama import chat
from ollama import ChatResponse

response: ChatResponse = chat(model='qwen3:0.6b', messages=[
  {
    'role': 'user',
    'content': '你是谁？',
  },
])
# 打印响应内容
print(response['message']['content'])

# 或者直接访问响应对象的字段
#print(response.message.content)

【例】下面就是运行的效果，运行的时候需要等待一小会，可能是电脑性能太弱了。

2. 流式响应

上面 示例是等模型推理完才输出数据，要等一段时间才会有输出，但是我们平时用的网页上的大模型都是打字机一样的输出，这个是流式输出，可以在发送请求的时候设置 stream=True 来启用响应流式传输。

# -*- coding: utf-8 -*-
from ollama import chat

stream = chat(
    model='qwen3:0.6b',
    messages=[{'role': 'user', 'content': '你是谁？'}],
    stream=True,
)

# 逐块打印响应内容
for chunk in stream:
    print(chunk['message']['content'], end='', flush=True)

三、其他方式

1. 通过httpx.post来实现

# -*- coding: utf-8 -*-
import httpx
import json

def chat_with_httpx():
    url = "http://localhost:11434/api/chat"  # Ollama API地址
    payload = {
        "model": "qwen3:0.6b",
        "messages": [{"role": "user", "content": "你是谁？"}],
        "stream": True
    }
    
    with httpx.stream("POST", url, json=payload) as response:
        for line in response.iter_lines():
            if line:
                chunk = json.loads(line)
                print(chunk['message']['content'], end='', flush=True)

if __name__ == "__main__":
    chat_with_httpx()

2. openai

2.1 openai简介

调用库（Libraries） | OpenAI 官方帮助文档中文版

OpenAI API 可以应用于几乎所有涉及生成自然语言、代码或图像的任务。这个库提供了一系列不同能力级别的 模型，适用于不同任务的，并且能够 微调（Fine-tune） 我们自己的自定义模型。这些模型可以用于从内容生成到语义搜索和分类的所有领域。

2.2 Python库

pip install openai

2.3 使用示例

# -*- coding: utf-8 -*-
from openai import OpenAI  

# 必填：从服务管控页面获取对应服务的APIKey和API Base
api_key = "ollama"
api_base = "http://localhost:11434/v1"

client = OpenAI(api_key=api_key, base_url=api_base)

def unified_chat_test(model_id, messages, use_stream=False, extra_body={}):
    """
    一个统一的函数，用于演示多种调用场景。

    :param model_id: 要调用的模型ID。
    :param messages: 对话消息列表。
    :param use_stream: 是否使用流式输出。
    :param extra_body: 包含额外请求参数的字典，如 response_format。
    """
    try:
        response = client.chat.completions.create(
            model=model_id,
            messages=messages,
            stream=use_stream,
            temperature=0.7,
            max_tokens=4096,
            extra_headers={"lora_id": "0"},  # 调用微调大模型时,对应替换为模型服务卡片上的resourceId
            stream_options={"include_usage": True},
            extra_body=extra_body
        )

        if use_stream:
            # 处理流式响应
            full_response = ""
            print("--- 流式输出 ---")
            for chunk in response:
                if hasattr(chunk.choices[0].delta, 'content') and chunk.choices[0].delta.content:
                    content = chunk.choices[0].delta.content
                    print(content, end="", flush=True)
                    full_response += content
            print("\n\n--- 完整响应 ---")
            print(full_response)
        else:
            # 处理非流式响应
            print("--- 非流式输出 ---")
            message = response.choices[0].message
            print(message.content)

    except Exception as e:
        print(f"请求出错: {e}")

if __name__ == "__main__":
    model_id = "qwen3:0.6b" # 必填：调用大模型时，对应为推理服务的模型卡片上对应的modelId

    # 1. 普通非流式调用
    print("********* 1. 普通非流式调用 *********")
    plain_messages = [{"role": "user", "content": "你好，请介绍一下自己。"}]
    unified_chat_test(model_id, plain_messages, use_stream=False)

    # 2. 普通流式调用
    print("\n********* 2. 普通流式调用 *********")
    stream_messages = [{"role": "user", "content": "写一首关于夏天的诗。"}]
    unified_chat_test(model_id, stream_messages, use_stream=True)

    # 3. JSON Mode 调用
    print("\n********* 3. JSON Mode 调用 *********")
    json_messages = [{"role": "user", "content": "请给我一个关于上海的JSON对象，包含城市名称(city)和人口数量(population)。"}]
    json_extra_body = {
        "response_format": {"type": "json_object"},
        "search_disable": True # JSON Mode下建议关闭搜索
    }
    unified_chat_test(model_id, json_messages, use_stream=False, extra_body=json_extra_body)

    # 4. 测试stop和前缀续写功能
    print("\n********* 4. 测试stop和前缀续写功能 *********")
    print("设置stop词: ['。', '！'] - 模型遇到句号或感叹号时会停止生成")
    stream_messages = [{"role": "user", "content": "给我解释下1加1等于多少。"}]
    unified_chat_test(model_id, stream_messages, use_stream=True, extra_body={"stop": ["。","！"],"continue_final_message":True})

四、总结

其实不管通过什么方法，我们最终都是向ollama提供的大模型api发送特定格式的请求，然后接收返回的数据并处理，只要符合特定格式就可以了。