Ollama: Python SDK集成

import ollama

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

# Install the ollama Python package
pip install ollama

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

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())

# 函数定义成功

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'])

chat:
generate:

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

# 执行成功

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())

# 函数定义成功

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")

Failed to get response

# ============================================
# 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?"))

=== 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?