1. 环境准备与基础入门

如果你对本地运行大模型感兴趣，那Ollama绝对是你的首选工具。它就像一个开箱即用的“模型管理器”，让你能在自己的电脑上轻松运行Llama、Gemma、Qwen这些热门模型。而Python作为我们最熟悉的胶水语言，通过Ollama官方提供的Python库，就能以极低的门槛调用这些强大的AI能力。我自己在项目里用Ollama API已经大半年了，从最初的简单对话到后来的复杂工具调用，踩过不少坑，也积累了不少实战经验，今天就来和你详细聊聊。

首先，你得确保Ollama本体已经安装并运行在你的机器上。去Ollama官网下载对应操作系统的安装包，安装完成后在终端输入 ollama --version 确认安装成功。接着，你需要拉取一个模型，比如目前很受欢迎的DeepSeek-R1，命令很简单：ollama pull deepseek-r1。这个步骤会把模型文件下载到本地，通常存储在 ~/.ollama/models 目录下（Windows用户在 C:\Users\<用户名>\.ollama\models）。模型大小从几GB到几十GB不等，取决于你选择的型号，所以请确保你的硬盘有足够空间。

Python环境的准备就更简单了。我强烈建议使用虚拟环境来管理依赖，避免污染全局环境。你可以用 venv 或者 conda 创建一个新环境。然后，安装Ollama的Python库只需要一行命令：pip install ollama。这个库封装了所有与Ollama服务交互的细节，让我们能用非常Pythonic的方式调用API。安装完成后，写个最简单的“Hello World”测试一下连接是否正常。先确保Ollama服务在后台运行（安装后通常会自动启动一个后台服务，监听11434端口），然后运行一段测试代码。如果一切顺利，你就已经迈出了第一步，接下来我们可以深入探索各种调用方式了。

2. 基础文本生成与聊天对话

2.1 使用Generate API进行单次文本生成

Ollama提供了两种核心的文本生成接口：generate 和 chat。我们先从更基础的 generate 开始，它适合一次性的问答或内容补全任务。它的工作方式很直接：你给它一个提示（prompt），它返回模型生成的完整响应。用官方库调用起来非常简单，导入 generate 函数，指定模型名和你的问题即可。我实测下来，对于像“解释技术概念”或“写一段代码”这样的任务，generate 非常高效。

这里有个细节需要注意，generate 函数返回的是一个字典，里面包含 response 键，存放着模型生成的文本。除了文本，返回对象里还有一些有用的元信息，比如 total_duration（总耗时）、eval_count（生成的token数）等，这些对于性能监控和调试很有帮助。比如，你可以根据 eval_count 来估算API调用的成本（如果你未来使用按token计费的云服务）。下面是一个完整的例子，我加上了错误处理，因为在实际项目中网络波动或模型未加载都可能导致调用失败。用 try-except 包裹起来是个好习惯，可以捕获 ResponseError 并做相应处理，比如提示用户模型不存在并自动拉取。

from ollama import generate

def simple_generate():
    try:
        response = generate(
            model='deepseek-r1',
            prompt='用简单的语言解释一下什么是机器学习？'
        )
        # 直接访问响应文本
        answer = response['response']
        print(f"模型回答：{answer}")
        
        # 查看元信息
        print(f"生成耗时：{response.get('total_duration', 0) / 1e9:.2f} 秒")
        print(f"生成Token数：{response.get('eval_count', 0)}")
        
    except Exception as e:
        print(f"调用出错：{e}")
        # 这里可以添加更细致的错误处理，比如检查是否是404错误（模型不存在）
        # 然后自动调用 ollama.pull('deepseek-r1')

if __name__ == '__main__':
    simple_generate()

2.2 使用Chat API进行多轮对话

虽然 generate 能用，但大多数交互式场景，比如聊天机器人、客服助手，我们更需要 chat API。它最大的优势是支持消息历史。你可以传入一个消息列表，其中每条消息都有 role（角色，如 user、assistant、system）和 content（内容）。模型会根据整个对话上下文来生成回复，这让多轮对话变得连贯自然。

system 角色消息特别有用，你可以用它来设定AI的“人设”或行为指令。比如，你可以让模型“扮演一位资深的Python导师，用幽默风趣的方式回答问题”。这个系统提示词会持续影响整个会话。在代码里，你需要构建一个 messages 列表，然后传给 chat 函数。每次获得AI回复后，记得要把这次交互的 user 消息和 assistant 消息都追加到 messages 列表末尾，这样下次提问时，模型就记得之前聊过什么了。我刚开始用的时候忘了维护历史，结果每次对话AI都像失忆了一样，闹了不少笑话。

from ollama import chat

def multi_turn_chat():
    # 初始化对话历史，包含系统指令
    messages = [
        {
            'role': 'system',
            'content': '你是一位乐于助人且知识渊博的AI助手，回答要简洁明了。'
        },
        {
            'role': 'user',
            'content': 'Python中的列表和元组有什么区别？'
        }
    ]
    
    # 第一轮对话
    response = chat(model='deepseek-r1', messages=messages)
    ai_reply = response['message']['content']
    print(f"AI: {ai_reply}")
    
    # 将AI回复加入历史
    messages.append({'role': 'assistant', 'content': ai_reply})
    
    # 用户继续提问（基于上下文）
    new_question = '那我应该在什么情况下用元组呢？'
    messages.append({'role': 'user', 'content': new_question})
    print(f"用户: {new_question}")
    
    # 第二轮对话，传入包含历史的新messages
    response = chat(model='deepseek-r1', messages=messages)
    print(f"AI: {response['message']['content']}")

# 运行对话
multi_turn_chat()

3. 提升性能与体验：异步与流式调用

3.1 异步调用应对高并发场景

当你的应用需要同时处理多个用户请求，或者需要在不阻塞主线程的情况下调用模型时，同步的 chat 或 generate 就会成为性能瓶颈。这时候就该 AsyncClient 出场了。Ollama Python库提供了完整的异步支持，底层基于 httpx 和 asyncio。使用异步客户端，你可以在一个事件循环中并发发起多个API请求，极大提升吞吐量。我在开发一个需要同时查询多个模型做对比的工具时，异步调用将总耗时从线性累加降低到了几乎只取决于最慢的那个请求。

使用 AsyncClient 的步骤很简单：导入 AsyncClient 类，在异步函数中用 await 调用其方法。记得要把所有相关操作都放在异步上下文中。一个常见的模式是定义一个 main 异步函数，然后用 asyncio.run(main()) 来执行。下面这个例子展示了如何异步并发地向同一个模型问两个不同的问题，你可以轻松扩展到同时调用不同模型。

import asyncio
from ollama import AsyncClient

async def concurrent_requests():
    client = AsyncClient()
    questions = [
        '简述区块链的工作原理。',
        '如何用Python快速读取一个大文件？'
    ]
    
    # 创建多个异步任务
    tasks = []
    for q in questions:
        task = client.chat(
            model='deepseek-r1',
            messages=[{'role': 'user', 'content': q}]
        )
        tasks.append(task)
    
    # 并发执行所有任务
    responses = await asyncio.gather(*tasks)
    
    for idx, resp in enumerate(responses):
        print(f"问题 {idx+1}: {questions[idx]}")
        print(f"回答: {resp['message']['content'][:100]}...")  # 只打印前100字符
        print("-" * 50)

# 运行异步函数
asyncio.run(concurrent_requests())

3.2 流式输出实现打字机效果

如果你用过ChatGPT的网页版，一定很喜欢它那种逐字输出的“打字机”效果。这种体验不仅能降低用户等待的焦虑感，还能在生成长篇内容时提前看到部分结果。Ollama API原生支持流式输出，在Python里用起来非常优雅。你只需要在调用 chat 或 generate 时加上 stream=True 参数，函数就会返回一个生成器（Generator）。然后你可以遍历这个生成器，每次迭代得到一个“块”（chunk），里面包含了当前刚生成的那部分文本。

在遍历时，我习惯用 print(chunk['message']['content'], end='', flush=True) 来打印，end='' 避免换行，flush=True 确保内容立即显示而不是缓存在缓冲区。对于异步流式调用，则使用 async for 循环。流式输出在处理长文本时优势明显，比如让模型写一篇千字文章，你不需要等它全部生成完就能开始阅读开头部分。不过要注意，因为响应被分成了很多小块，你无法直接拿到像总耗时这样的完整元数据，这些信息通常只在流式响应的最后一个块里提供。

from ollama import chat

def stream_response():
    messages = [{'role': 'user', 'content': '写一个关于人工智能未来的短篇故事，大约200字。'}]
    
    print("AI正在创作：", end='', flush=True)
    full_response = []
    
    # 传入 stream=True，返回生成器
    for chunk in chat(model='deepseek-r1', messages=messages, stream=True):
        # chunk 是一个字典，结构类似非流式响应，但只包含当前片段
        text_piece = chunk['message']['content']
        print(text_piece, end='', flush=True)
        full_response.append(text_piece)
    
    # 如果需要完整的响应文本，可以拼接起来
    final_story = ''.join(full_response)
    print(f"\n\n故事总长度：{len(final_story)} 字符")

# 体验流式输出
stream_response()

4. 高级功能实战：工具调用与函数执行

4.1 为模型赋予调用函数的能力

这是Ollama API最强大的功能之一，也是让AI从“聊天”走向“执行”的关键。工具调用（Tool Calling）允许大模型根据你的请求，决定是否需要调用一个外部函数（工具）来获取信息或执行操作，然后将函数结果融入它的回答中。比如，你问“北京和上海现在的气温各是多少？”，模型自己无法获取实时天气，但它可以识别出需要调用一个 get_weather(city) 函数，然后利用函数返回的真实数据来组织答案。

在Ollama中实现工具调用需要三步：第一，定义好你的函数（工具），包括函数名、描述和参数模式；第二，在调用 chat 时通过 tools 参数把这些工具描述传给模型；第三，处理模型的响应，如果它返回了工具调用请求，你就去执行对应的真实函数，再把执行结果以特定格式发回给模型，让模型生成最终回答。听起来有点绕，但代码结构其实很清晰。Ollama库支持你直接传入Python函数对象，它会自动提取函数签名和文档字符串来构建工具描述，这大大简化了流程。

from ollama import chat, ChatResponse
import json

# 1. 定义工具函数
def get_current_temperature(city: str) -> float:
    """获取指定城市的当前气温（摄氏度）。
    
    参数:
        city: 城市名，例如 '北京'、'上海'。
    """
    # 这里应该是调用真实天气API，为了示例我们模拟数据
    mock_data = {'北京': 22.5, '上海': 25.0, '广州': 28.3}
    return mock_data.get(city, 0.0)

def calculator(operation: str, a: float, b: float) -> float:
    """执行简单的数学运算。
    
    参数:
        operation: 运算类型，支持 'add', 'subtract', 'multiply', 'divide'。
        a: 第一个数字。
        b: 第二个数字。
    """
    if operation == 'add':
        return a + b
    elif operation == 'subtract':
        return a - b
    elif operation == 'multiply':
        return a * b
    elif operation == 'divide':
        return a / b if b != 0 else float('inf')
    else:
        raise ValueError(f"不支持的运算: {operation}")

def main():
    # 2. 准备对话和工具列表
    messages = [{'role': 'user', 'content': '北京和上海的气温相差多少度？顺便算一下15乘以3等于几。'}]
    
    # 可以直接传入函数对象，库会自动处理
    response: ChatResponse = chat(
        model='qwen3:0.6b',  # 使用一个支持工具调用的模型
        messages=messages,
        tools=[get_current_temperature, calculator],
    )
    
    # 3. 检查模型是否要求调用工具
    if response.message.tool_calls:
        print("模型请求调用工具...")
        # 准备一个函数名到实际函数的映射
        available_functions = {
            'get_current_temperature': get_current_temperature,
            'calculator': calculator,
        }
        
        tool_outputs = []
        # 遍历所有工具调用请求（可能多个）
        for tool_call in response.message.tool_calls:
            func_name = tool_call.function.name
            if func_name in available_functions:
                # 解析模型传来的参数（JSON字符串）
                kwargs = json.loads(tool_call.function.arguments)
                print(f"  调用函数 {func_name}，参数: {kwargs}")
                
                # 执行真实函数
                result = available_functions[func_name](**kwargs)
                print(f"  结果: {result}")
                
                # 记录结果，准备发回给模型
                tool_outputs.append({
                    'role': 'tool',
                    'content': str(result),
                    'name': func_name,
                    'tool_call_id': tool_call.id  # 关联对应的工具调用
                })
            else:
                print(f"  警告：未找到函数 {func_name}")
        
        # 4. 将工具执行结果追加到消息历史
        messages.append(response.message)  # 加入模型刚才的消息（包含工具调用）
        messages.extend(tool_outputs)       # 加入所有工具执行结果
        
        # 5. 再次调用模型，让它基于工具结果生成最终回答
        final_response = chat(model='qwen3:0.6b', messages=messages)
        print(f"\n最终回答：{final_response.message.content}")
    else:
        print("模型未调用工具，直接回答：", response.message.content)

if __name__ == '__main__':
    main()

4.2 异步环境下的工具调用

在异步应用（比如FastAPI后端）中，你同样可以使用工具调用功能，只需结合 AsyncClient 即可。逻辑和同步版本完全一致，只是函数调用都变成了 await 形式。这里有个小技巧：工具函数本身如果是同步的（比如上面模拟的天气函数），在异步上下文中直接调用也没问题，但如果工具函数内部涉及网络I/O（比如真的去请求天气API），最好也将其改写成异步函数，或者用 asyncio.to_thread 在单独线程中运行，避免阻塞事件循环。下面的示例展示了异步工具调用的完整流程，我加上了更健壮的错误处理，因为在实际生产环境中，工具函数可能会失败（API超时、参数错误等），我们需要给模型反馈一个错误信息，让它能调整回答。

import asyncio
import json
from ollama import AsyncClient, ChatResponse

# 假设的异步工具函数
async def fetch_weather_async(city: str) -> dict:
    """异步获取天气信息。"""
    await asyncio.sleep(0.5)  # 模拟网络延迟
    mock_data = {'北京': {'temp': 22.5, 'condition': '晴'}, '上海': {'temp': 25.0, 'condition': '多云'}}
    return mock_data.get(city, {'temp': 0.0, 'condition': '未知'})

async def async_tool_calling():
    client = AsyncClient()
    messages = [{'role': 'user', 'content': '北京和上海的天气怎么样？'}]
    
    try:
        response: ChatResponse = await client.chat(
            model='qwen3:0.6b',
            messages=messages,
            tools=[fetch_weather_async],  # 注意：这里传的是函数引用，库会处理描述
        )
        
        if response.message.tool_calls:
            print("检测到工具调用请求。")
            # 这里简化处理，假设只有一个工具调用
            for tool_call in response.message.tool_calls:
                if tool_call.function.name == 'fetch_weather_async':
                    cities = json.loads(tool_call.function.arguments).get('city', '')
                    # 实际中可能需要解析出多个城市，这里简单处理
                    print(f"模型想查询城市: {cities}")
                    
                    # 执行异步工具函数
                    weather_info = await fetch_weather_async(cities)
                    
                    # 将结果格式化成模型能理解的文本
                    tool_message = {
                        'role': 'tool',
                        'content': json.dumps(weather_info, ensure_ascii=False),
                        'name': 'fetch_weather_async',
                        'tool_call_id': tool_call.id
                    }
                    messages.append(response.message)
                    messages.append(tool_message)
                    
                    # 获取最终回答
                    final_response = await client.chat(model='qwen3:0.6b', messages=messages)
                    print(f"整合天气信息后的回答：{final_response.message.content}")
                    return
        
        # 如果没有工具调用，直接输出
        print(f"直接回答：{response.message.content}")
        
    except Exception as e:
        print(f"工具调用过程中出错：{e}")
        # 可以考虑向用户返回一个友好的错误信息

# 运行异步示例
asyncio.run(async_tool_calling())

5. 深入探索：模型管理与云服务集成

5.1 本地模型的生命周期管理

除了核心的生成和聊天功能，Ollama Python库还提供了一套完整的模型管理API，让你能在代码里直接操作本地模型，就像在命令行使用 ollama 命令一样方便。这对于构建需要动态加载、切换模型的应用非常有用。list() 函数可以列出所有已下载的模型及其详细信息；show(model_name) 能查看某个模型的详细配置，包括它的Modelfile内容、参数设置、许可证等；pull(model_name) 用于拉取新模型，你可以显示下载进度；create() 则允许你基于现有模型创建自定义版本，比如修改系统提示词。

我常用 list() 来做一个模型选择器，让用户在我的图形界面里下拉选择可用模型。而 create() 功能非常强大，比如你可以基于 llama3.2 创建一个专用于代码审查的模型，给它一个特定的系统指令。下面这段代码演示了如何列出模型，并基于其中一个创建定制化版本。注意，create 操作可能需要一些时间，因为它涉及到模型的复制和重新配置。

from ollama import list_models, create_model, pull_model

def manage_local_models():
    # 1. 列出所有本地模型
    print("本地已安装的模型：")
    models_info = list_models()
    for model in models_info.get('models', []):
        print(f"  - {model['name']} (大小: {model.get('size', '未知')})")
    
    # 2. 拉取一个新模型（例如，尝试拉取一个较小的模型）
    new_model_name = 'gemma3:2b'  # 假设的模型名，请替换为实际可用模型
    print(f"\n正在尝试拉取模型 {new_model_name}...")
    try:
        # pull函数会返回一个生成器，用于流式显示下载进度
        for progress in pull_model(new_model_name, stream=True):
            # progress 包含状态、已完成/总量等信息
            if 'completed' in progress and 'total' in progress:
                percent = (progress['completed'] / progress['total']) * 100
                print(f"下载进度: {percent:.1f}%", end='\r')
        print(f"\n模型 {new_model_name} 拉取完成！")
    except Exception as e:
        print(f"拉取失败: {e}。可能模型名不存在或网络问题。")
    
    # 3. 创建自定义模型
    base_model = 'llama3.2'  # 确保这个模型已存在
    custom_model_name = 'my-coder-assistant'
    modelfile_content = f"""
FROM {base_model}
SYSTEM 你是一个专业的Python代码助手，擅长代码审查、重构和优化。回答时优先给出可运行的代码示例。
PARAMETER temperature 0.7
"""
    
    print(f"\n正在创建自定义模型 '{custom_model_name}'...")
    try:
        # create 也会返回生成器显示创建进度
        for status in create_model(custom_model_name, modelfile=modelfile_content, stream=True):
            print(f"状态: {status.get('status', '处理中')}", end='\r')
        print(f"\n自定义模型 '{custom_model_name}' 创建成功！")
        
        # 现在你可以像使用其他模型一样使用它
        # response = chat(model=custom_model_name, messages=[{'role':'user', 'content':'帮我优化这段代码...'}])
        
    except Exception as e:
        print(f"创建模型失败: {e}")

if __name__ == '__main__':
    manage_local_models()

5.2 连接Ollama云服务与OpenAI兼容接口

如果你本地显卡不够强，想跑更大的模型，或者希望获得更稳定的服务，Ollama提供了云模型服务。这些模型运行在Ollama的云端服务器上，你通过API调用，按使用量付费。Python库连接云服务有两种方式：第一种是配置客户端指向云API端点（https://ollama.com）并设置API密钥；第二种更简单，如果你本地已经安装了Ollama，可以通过 ollama signin 登录账户，然后直接 pull 一个云模型（模型名带 -cloud 后缀），之后调用方式就和本地模型一模一样，库会自动将请求路由到云端。

另一个超级实用的功能是OpenAI API兼容性。Ollama提供了一个兼容OpenAI API格式的端点（http://localhost:11434/v1），这意味着你可以直接使用熟悉的 openai Python包来调用本地Ollama模型！这对于那些已经基于OpenAI API开发了应用，想快速迁移到本地模型的开发者来说，简直是零成本切换。你只需要将 OpenAI 客户端的 base_url 指向Ollama的v1端点，并随便设置一个 api_key（Ollama会忽略它，但库要求有值）。之后，client.chat.completions.create 等所有方法都可以照常使用。

# 示例：使用OpenAI兼容接口调用Ollama
from openai import OpenAI

# 指向本地Ollama服务的v1端点
client = OpenAI(
    base_url='http://localhost:11434/v1/',
    api_key='ollama',  # 任意字符串，Ollama不验证但参数必填
)

# 现在你可以像调用GPT一样调用本地模型了！
completion = client.chat.completions.create(
    model='llama3.2',  # 使用你本地有的模型名
    messages=[
        {"role": "system", "content": "你是一个有帮助的助手。"},
        {"role": "user", "content": "用Python写一个快速排序函数。"}
    ],
    stream=True,  # 甚至也支持流式！
    temperature=0.8,
    max_tokens=500
)

# 处理流式响应
for chunk in completion:
    if chunk.choices[0].delta.content is not None:
        print(chunk.choices[0].delta.content, end='', flush=True)

这种兼容性大大降低了生态工具的使用门槛。许多现有的LLM应用框架（如LangChain、LlamaIndex）、开源项目（如聊天界面、RAG系统）都可以无缝接入Ollama。你只需要修改配置中的API地址和模型名，就能让它们跑在你的本地环境或私有云上，既保护了数据隐私，又节省了API费用。我在好几个内部知识库项目中都采用了这种方案，效果非常稳定。