English العربية Português 日本語

Ollama: Integração com SDK Python

O SDK Python é a chave que abre a porta para IA local — três linhas de código, transição perfeita de script para aplicação.

💡 Dica: O SDK Python (biblioteca ollama) é essencialmente um wrapper da API REST do Ollama — o método chat() encapsula /api/chat, generate() encapsula /api/generate, e list() encapsula /api/tags. Entender essa relação ajuda na solução de problemas: quando o SDK lança um erro, você pode usar curl para chamar a API diretamente e determinar se é um problema do SDK ou do serviço Ollama.

📋 Pré-requisitos: Você precisa dominar o seguinte primeiro

1. O Que Você Vai Aprender

2. Uma História Real de Uma Fundadora de SaaS

⚠️ Aviso: Quando o SDK Python chama a API do Ollama diretamente, o Ollama não possui autenticação embutida. Se seu Ollama está vinculado a 0.0.0.0, informações de conexão do SDK (por exemplo, http://your-server:11434) podem ser exploradas por outros. Em produção, sempre adicione uma camada de autenticação.

(1) O Problema: Scripts Shell Não São Suficientes

Alice construiu um protótipo SupportBot com curl, mas scripts Shell são difíceis de manter para estado de conversa, tratamento de erros e integração com serviços web. Ela precisa de uma linguagem de programação adequada para construir aplicações de nível de produção.

(2) A Solução: Integração em Três Linhas com SDK Python

PYTHON 
import ollama

response = ollama.chat(model='qwen2.5', messages=[
    {'role': 'user', 'content': 'Hello'}
])
print(response['message']['content'])

3. Instalação e Visão Geral da API

ℹ️ Info: O SDK Python ollama se conecta por padrão a http://localhost:11434, sem necessidade de configuração adicional. Se o Ollama estiver rodando em um endereço diferente, você pode sobrescrever definindo a variável de ambiente OLLAMA_HOST ou especificando Client(host='http://...') no código.

💡 Dica: O método chat() do SDK Python é recomendado sobre generate() — chat suporta diálogo de múltiplas rodadas (array messages), enquanto generate suporta apenas rodada única. Mesmo para perguntas de uso único, a distinção de papel do chat (system/user/assistant) produz melhor qualidade de saída.

(1) Instalação e Verificação de Conexão

BASH 
# Install the ollama Python package
pip install ollama

# Verify connection to Ollama server
python3 -c "import ollama; print(ollama.list())"

(2) Comparação API Síncrona vs Assíncrona

⚠️ Nota: Ao usar a API assíncrona (AsyncClient), todas as chamadas devem ser prefixadas com await, por exemplo, await client.chat(...). Esquecer await retorna um objeto coroutine em vez do resultado real — o programa não erro mas não produz saída correta. Em frameworks async como FastAPI, você deve usar a API assíncrona; caso contrário, bloqueia o loop de eventos e prejudica o desempenho concorrente.

Dimensão API Síncrona API Assíncrona
Módulo ollama ollama (AsyncClient)
Estilo de chamada ollama.chat() await client.chat()
Bloqueio Bloqueia thread atual Não bloqueante, concorrente
Caso de uso Scripts, ferramentas simples Serviços web, processamento concorrente
Suporte streaming for chunk in stream async for chunk in stream

▶ Exemplo 1: Chamadas Básicas Síncronas e Assíncronas

PYTHON 
import ollama
import asyncio

# Synchronous call
def sync_chat():
    response = ollama.chat(
        model='qwen2.5',
        messages=[{'role': 'user', 'content': 'Hello!'}]
    )
    print(response['message']['content'])

# Asynchronous call
async def async_chat():
    client = ollama.AsyncClient()
    response = await client.chat(
        model='qwen2.5',
        messages=[{'role': 'user', 'content': 'Hello!'}]
    )
    print(response['message']['content'])

sync_chat()
asyncio.run(async_chat())

Saída：

TEXT 
# Function defined successfully

4. Aprofundamento nos Métodos Centrais

(1) Método chat()

Parâmetro Tipo Descrição
model str Nome do Modelo
messages list[dict] Lista de mensagens, cada uma contendo role/content
stream bool Se a saída é streaming
format str Formato de saída: json
options dict Parâmetros de Inferência (temperature, etc.)
keep_alive str Tempo de retenção do Modelo na memória

(2) Método generate()

Parâmetro Tipo Descrição
model str Nome do Modelo
prompt str Texto do Prompt
system str System Prompt
stream bool Se é streaming
options dict Parâmetros de Inferência

▶ Exemplo 2: Comparação chat vs generate

PYTHON 
import ollama

# chat(): multi-turn with message history
response = ollama.chat(
    model='qwen2.5',
    messages=[
        {'role': 'system', 'content': 'You are a SQL expert.'},
        {'role': 'user', 'content': 'Write a query for top 5 customers'}
    ],
    stream=False,
    options={'temperature': 0.3}
)
print('chat:', response['message']['content'])

# generate(): single-shot text generation
response = ollama.generate(
    model='qwen2.5',
    prompt='Write a haiku about debugging',
    system='You are a poet.',
    stream=False
)
print('generate:', response['response'])

Saída：

TEXT 
chat:
generate:

5. Implementação de Resposta Streaming

(1) Princípio da Saída Streaming

100% 
sequenceDiagram
    participant P as App Python
    participant O as Servidor Ollama
    P->>O: chat(stream=True)
    loop Cada chunk de token
        O-->>P: chunk {"content": "word"}
        P->>P: print(word, end="")
    end
    O-->>P: chunk {"done": true}

▶ Exemplo 3: Saída Streaming Síncrona

PYTHON 
import ollama

# Stream chat response in real-time
stream = ollama.chat(
    model='qwen2.5',
    messages=[{'role': 'user', 'content': 'Explain RAG in 3 sentences'}],
    stream=True
)

for chunk in stream:
    content = chunk['message']['content']
    print(content, end='', flush=True)

print()  # newline at end

Saída：

TEXT 
# Execution successful

▶ Exemplo 4: Saída Streaming Assíncrona

PYTHON 
import ollama
import asyncio

async def stream_chat():
    client = ollama.AsyncClient()
    stream = await client.chat(
        model='qwen2.5',
        messages=[{'role': 'user', 'content': 'Tell me about Ollama'}],
        stream=True
    )
    async for chunk in stream:
        content = chunk['message']['content']
        print(content, end='', flush=True)
    print()

asyncio.run(stream_chat())

Saída：

TEXT 
# Function defined successfully

6. Tratamento de Erros e Anotações de Tipo

(1) Tipos Comuns de Erro

Erro Condição de Disparo Tratamento
ConnectionError Serviço Ollama não está rodando Iniciar serviço ou retry
ResponseError Modelo não encontrado / parâmetros inválidos Verificar nome do Modelo e parâmetros
TimeoutError Timeout de Inferência Reduzir num_ctx ou aumentar timeout
JSONDecodeError Anomalia na saída format=json Adicionar validação JSON e retry

▶ Exemplo 5: Tratamento de Erros Robusto

PYTHON 
import ollama
import json
from typing import Optional

def safe_chat(
    model: str,
    messages: list[dict],
    temperature: float = 0.3,
    max_retries: int = 3
) -> Optional[str]:
    """Chat with error handling and retries."""
    for attempt in range(max_retries):
        try:
            response = ollama.chat(
                model=model,
                messages=messages,
                stream=False,
                options={'temperature': temperature}
            )
            return response['message']['content']

        except ConnectionError:
            print(f"Connection failed (attempt {attempt + 1})")
            if attempt == max_retries - 1:
                return None

        except ollama.ResponseError as e:
            print(f"API error: {e.error}")
            return None

        except Exception as e:
            print(f"Unexpected error: {e}")
            if attempt == max_retries - 1:
                return None

    return None

# Usage
result = safe_chat('qwen2.5', [
    {'role': 'user', 'content': 'What is your return policy?'}
])
if result:
    print(result)
else:
    print("Failed to get response")

Saída：

TEXT 
Failed to get response

7. Exemplo Abrangente: Wrapper Python do SupportBot V1

PYTHON 
# ============================================
# Comprehensive: SupportBot V1
# Python wrapper for e-commerce customer service
# ============================================

import ollama
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class SupportBot:
    model: str = "qwen2.5"
    temperature: float = 0.4
    max_history: int = 10
    system_prompt: str = (
        "You are SupportBot, an e-commerce customer service agent. "
        "Be polite, concise, and helpful. "
        "If unsure, say 'Let me connect you with a human agent.'"
    )
    messages: list[dict] = field(default_factory=list)

    def __post_init__(self):
        self.messages = [
            {"role": "system", "content": self.system_prompt}
        ]

    def chat(self, user_input: str) -> str:
        self.messages.append({"role": "user", "content": user_input})
        try:
            response = ollama.chat(
                model=self.model,
                messages=self.messages[-self.max_history:],
                stream=False,
                options={"temperature": self.temperature}
            )
            assistant_msg = response["message"]["content"]
            self.messages.append({"role": "assistant", "content": assistant_msg})
            return assistant_msg
        except Exception as e:
            self.messages.pop()  # Remove failed user message
            return f"Error: {str(e)}"

    def stream_chat(self, user_input: str):
        self.messages.append({"role": "user", "content": user_input})
        full_response = []
        try:
            stream = ollama.chat(
                model=self.model,
                messages=self.messages[-self.max_history:],
                stream=True,
                options={"temperature": self.temperature}
            )
            for chunk in stream:
                content = chunk["message"]["content"]
                full_response.append(content)
                print(content, end="", flush=True)
            print()
            self.messages.append({"role": "assistant", "content": "".join(full_response)})
        except Exception as e:
            print(f"\nError: {e}")

    def reset(self):
        self.messages = [{"role": "system", "content": self.system_prompt}]

# Usage
if __name__ == "__main__":
    bot = SupportBot(model="qwen2.5", temperature=0.4)

    print("=== SupportBot V1 ===")
    print(bot.chat("Where is my order #12345?"))
    print()
    print(bot.chat("It has been 7 days since I ordered."))
    print()
    print(bot.chat("Can I get a refund instead?"))
💻 Saída：

TEXT 
=== SupportBot V1 ===
I'd be happy to check on your order #12345. Based on our records, your order is currently in transit and expected to arrive within 2-3 business days. You can track it at our website.

I understand your concern. If you'd prefer a refund instead of waiting, I can initiate that for you. Our refund policy covers orders that haven't been delivered within the estimated timeframe.

Yes, I can process a full refund for order #12345. The refund will be credited to your original payment method within 3-5 business days. Would you like me to proceed?

❓ Perguntas Frequentes

P: pip install ollama falha, o que devo fazer? R: Garanta Python >= 3.8 e pip atualizado: pip install --upgrade pip. Para problemas de rede, use um espelho: pip install ollama -i https://pypi.tuna.tsinghua.edu.cn/simple.

P: Como escolher entre APIs síncronas e assíncronas? R: Use síncrona para scripts e ferramentas simples (mais simples). Use assíncrona para serviços web (FastAPI/Django) para evitar bloquear o loop de eventos. A diferença é mínima para cenários de usuário único.

P: Saída streaming não exibe no Jupyter Notebook, o que devo fazer? R: Jupyter tem suporte limitado para flush=True. Use IPython.display.clear_output com atualizações em loop, ou mude para modo não-streaming.

P: Como especificar o endereço do servidor Ollama? R: Conexão padrão é localhost:11434. Altere via variável de ambiente OLLAMA_HOST, ou instancie Client(host='http://...') no código.

P: O que acontece quando a lista messages fica muito longa? R: Conteúdo que excede a janela de contexto do Modelo (num_ctx) será truncado. Recomendamos implementar uma janela deslizante, mantendo apenas as últimas N rodadas, ou resumindo conversas anteriores.

P: Como obtenho velocidade de Inferência e outras estatísticas? R: Respostas não-streaming contêm campos total_duration, eval_count, prompt_eval_count. O último chunk de respostas streaming contém essas estatísticas.

📖 Resumo

📝 Exercícios

  1. Básico (Dificuldade ⭐): Use o SDK Python para implementar chat() e generate() uma vez cada, comparando diferenças na saída.
  2. Intermediário (Dificuldade ⭐⭐): Implemente uma função de chat streaming com janela deslizante (mantendo as últimas 5 rodadas), descartando automaticamente as mensagens mais antigas quando a conversa excede 5 rodadas.
  3. Avançado (Dificuldade ⭐⭐⭐): Construa um SupportBot V1+ com retry de erros, controle de timeout e saída em formato JSON que salva conversas de atendimento em um arquivo e registra o tempo de Inferência de cada chamada.
Web-Tutorial.com

Equipe Técnica Web-Tutorial

Uma plataforma de tutoriais mantida por diversos desenvolvedores. Cada tutorial é escrito e revisado por profissionais da área correspondente. Trabalhamos para manter nosso conteúdo preciso e confiável — se encontrar algum problema, avise-nos.

100%