这一章我们用 Ollama 构建一个完整的聊天机器人，支持多轮对话、历史记录等功能。
import ollama
def chatbot():
print("聊天机器人已启动，输入 'exit' 退出")
print("-" * 40)
messages = [
{'role': 'system', 'content': '你是一个友好的助手，回答简洁准确。'}
]
while True:
user_input = input("你: ")
if user_input.lower() in ['exit', 'quit', '退出']:
print("再见！")
break
messages.append({'role': 'user', 'content': user_input})
response = ollama.chat(
model='llama3.2',
messages=messages
)
reply = response['message']['content']
messages.append({'role': 'assistant', 'content': reply})
print(f"助手: {reply}")
print()
if __name__ == "__main__":
chatbot()
import ollama
def streaming_chatbot():
print("聊天机器人已启动（流式输出）")
print("-" * 40)
messages = [
{'role': 'system', 'content': '你是一个友好的助手。'}
]
while True:
user_input = input("你: ")
if user_input.lower() in ['exit', 'quit']:
break
messages.append({'role': 'user', 'content': user_input})
print("助手: ", end="", flush=True)
full_reply = ""
stream = ollama.chat(
model='llama3.2',
messages=messages,
stream=True
)
for chunk in stream:
text = chunk['message']['content']
if text:
print(text, end="", flush=True)
full_reply += text
print("\n")
messages.append({'role': 'assistant', 'content': full_reply})
streaming_chatbot()
import ollama
import json
from datetime import datetime
class ChatBot:
def __init__(self, model='llama3.2', system=None, history_file=None):
self.model = model
self.messages = []
self.history_file = history_file
if system:
self.messages.append({'role': 'system', 'content': system})
if history_file:
self.load_history()
def load_history(self):
try:
with open(self.history_file, 'r', encoding='utf-8') as f:
data = json.load(f)
self.messages = data.get('messages', self.messages)
except FileNotFoundError:
pass
def save_history(self):
if self.history_file:
with open(self.history_file, 'w', encoding='utf-8') as f:
json.dump({'messages': self.messages}, f, ensure_ascii=False, indent=2)
def send(self, content):
self.messages.append({'role': 'user', 'content': content})
response = ollama.chat(
model=self.model,
messages=self.messages
)
reply = response['message']['content']
self.messages.append({'role': 'assistant', 'content': reply})
self.save_history()
return reply
def send_stream(self, content, callback=None):
self.messages.append({'role': 'user', 'content': content})
full_reply = ""
stream = ollama.chat(
model=self.model,
messages=self.messages,
stream=True
)
for chunk in stream:
text = chunk['message']['content']
if text:
full_reply += text
if callback:
callback(text)
self.messages.append({'role': 'assistant', 'content': full_reply})
self.save_history()
return full_reply
def clear(self):
system_msg = next((m for m in self.messages if m['role'] == 'system'), None)
self.messages = [system_msg] if system_msg else []
self.save_history()
def export_chat(self, filename):
with open(filename, 'w', encoding='utf-8') as f:
for msg in self.messages:
role = {'system': '系统', 'user': '用户', 'assistant': '助手'}.get(msg['role'], msg['role'])
f.write(f"[{role}]\n{msg['content']}\n\n")
def main():
bot = ChatBot(
model='llama3.2',
system='你是一个友好的助手，回答简洁准确。',
history_file='chat_history.json'
)
print("聊天机器人已启动（带历史记录）")
print("命令: /clear 清空对话, /export 导出对话, /exit 退出")
print("-" * 40)
while True:
user_input = input("你: ")
if user_input.lower() in ['/exit', 'exit']:
break
elif user_input == '/clear':
bot.clear()
print("对话已清空\n")
continue
elif user_input == '/export':
bot.export_chat(f'chat_{datetime.now().strftime("%Y%m%d_%H%M%S")}.txt')
print("对话已导出\n")
continue
print("助手: ", end="", flush=True)
bot.send_stream(user_input, lambda t: print(t, end="", flush=True))
print("\n")
if __name__ == "__main__":
main()
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import ollama
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
class ChatRequest(BaseModel):
message: str
history: list = []
@app.post("/chat")
async def chat(request: ChatRequest):
messages = request.history + [
{'role': 'user', 'content': request.message}
]
response = ollama.chat(
model='llama3.2',
messages=messages
)
return {
'reply': response['message']['content'],
'history': messages + [{'role': 'assistant', 'content': response['message']['content']}]
}
@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
from fastapi.responses import StreamingResponse
import json
messages = request.history + [
{'role': 'user', 'content': request.message}
]
async def generate():
stream = ollama.chat(
model='llama3.2',
messages=messages,
stream=True
)
for chunk in stream:
if chunk['message']['content']:
yield f"data: {json.dumps({'content': chunk['message']['content']})}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>Ollama 聊天机器人</title>
<style>
* { margin: 0; padding: 0; box-sizing: border-box; }
body { font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; background: #f5f5f5; }
.container { max-width: 800px; margin: 0 auto; height: 100vh; display: flex; flex-direction: column; }
.header { padding: 20px; background: #4a90d9; color: white; }
.chat-container { flex: 1; overflow-y: auto; padding: 20px; }
.message { margin-bottom: 15px; }
.message.user { text-align: right; }
.message .bubble { display: inline-block; padding: 10px 15px; border-radius: 15px; max-width: 70%; }
.message.user .bubble { background: #4a90d9; color: white; }
.message.assistant .bubble { background: white; }
.input-container { padding: 20px; background: white; border-top: 1px solid #ddd; }
.input-container form { display: flex; gap: 10px; }
.input-container input { flex: 1; padding: 10px 15px; border: 1px solid #ddd; border-radius: 20px; outline: none; }
.input-container button { padding: 10px 20px; background: #4a90d9; color: white; border: none; border-radius: 20px; cursor: pointer; }
</style>
</head>
<body>
<div class="container">
<div class="header">
<h1>Ollama 聊天机器人</h1>
</div>
<div class="chat-container" id="chat"></div>
<div class="input-container">
<form id="form">
<input type="text" id="input" placeholder="输入消息..." autocomplete="off">
<button type="submit">发送</button>
</form>
</div>
</div>
<script>
const chat = document.getElementById('chat');
const form = document.getElementById('form');
const input = document.getElementById('input');
let history = [];
function addMessage(role, content) {
const div = document.createElement('div');
div.className = `message ${role}`;
div.innerHTML = `<div class="bubble">${content}</div>`;
chat.appendChild(div);
chat.scrollTop = chat.scrollHeight;
}
form.addEventListener('submit', async (e) => {
e.preventDefault();
const message = input.value.trim();
if (!message) return;
addMessage('user', message);
input.value = '';
const response = await fetch('/chat/stream', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ message, history })
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
let reply = '';
let assistantDiv = null;
while (true) {
const { done, value } = await reader.read();
if (done) break;
const lines = decoder.decode(value).split('\n');
for (const line of lines) {
if (line.startsWith('data: ') && line !== 'data: [DONE]') {
const data = JSON.parse(line.slice(6));
reply += data.content;
if (!assistantDiv) {
assistantDiv = document.createElement('div');
assistantDiv.className = 'message assistant';
assistantDiv.innerHTML = '<div class="bubble"></div>';
chat.appendChild(assistantDiv);
}
assistantDiv.querySelector('.bubble').textContent = reply;
chat.scrollTop = chat.scrollHeight;
}
}
}
history.push(
{ role: 'user', content: message },
{ role: 'assistant', content: reply }
);
});
</script>
</body>
</html>
import ollama
import argparse
from rich.console import Console
from rich.markdown import Markdown
from rich.panel import Panel
console = Console()
def rich_chatbot(model, system):
console.print(Panel.fit("Ollama 聊天机器人", style="bold blue"))
console.print("输入 /exit 退出, /clear 清空对话\n")
messages = []
if system:
messages.append({'role': 'system', 'content': system})
while True:
try:
user_input = console.input("[bold green]你:[/] ").strip()
except KeyboardInterrupt:
console.print("\n再见！")
break
if not user_input:
continue
if user_input == '/exit':
console.print("再见！")
break
elif user_input == '/clear':
messages = [m for m in messages if m['role'] == 'system']
console.print("[yellow]对话已清空[/]\n")
continue
messages.append({'role': 'user', 'content': user_input})
console.print("[bold blue]助手:[/]", end=" ")
full_reply = ""
stream = ollama.chat(model=model, messages=messages, stream=True)
for chunk in stream:
text = chunk['message']['content']
if text:
console.print(text, end="")
full_reply += text
console.print("\n")
messages.append({'role': 'assistant', 'content': full_reply})
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument('--model', default='llama3.2')
parser.add_argument('--system', default='你是一个友好的助手。')
args = parser.parse_args()
rich_chatbot(args.model, args.system)