Add LLM chat integration with streaming responses via Ollama
Phase 4: Full chat system with SSE streaming, note-aware context, and conversation persistence. Backend: - Migration 0005: conversations + messages tables with FKs and indexes - Conversation/Message SQLAlchemy models with relationships - LLM service: ensure_model (auto-pull on startup), stream_chat (NDJSON), generate_completion, fetch_url_content (HTML stripping), build_context (keyword extraction, related note search, URL content injection) - Chat service: conversation CRUD, save_response_as_note, summarize_conversation_as_note - Chat routes blueprint: 9 endpoints including SSE streaming for messages, save/summarize as note, Ollama model listing - Auto-pull llama3.1 model on app startup (non-blocking) Frontend: - apiStreamPost: SSE client using fetch + ReadableStream - Chat Pinia store with streaming state management - ChatView: dedicated /chat page with conversation sidebar + message thread - ChatPanel: slide-out panel with contextNoteId from current route - ChatMessage: markdown-rendered message bubble with "Save as Note" action - Updated AppHeader with Chat nav link + panel toggle button - Updated App.vue to mount ChatPanel with route-derived context - Added /chat and /chat/:id routes Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,155 @@
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
from fabledassistant.models import async_session
|
||||
from fabledassistant.models.conversation import Conversation, Message
|
||||
from fabledassistant.services.llm import generate_completion
|
||||
from fabledassistant.services.notes import create_note
|
||||
|
||||
|
||||
async def create_conversation(title: str = "", model: str = "") -> Conversation:
|
||||
async with async_session() as session:
|
||||
conv = Conversation(title=title, model=model)
|
||||
session.add(conv)
|
||||
await session.commit()
|
||||
await session.refresh(conv)
|
||||
# Initialize empty messages list for to_dict()
|
||||
conv.messages = []
|
||||
return conv
|
||||
|
||||
|
||||
async def get_conversation(conversation_id: int) -> Conversation | None:
|
||||
async with async_session() as session:
|
||||
result = await session.execute(
|
||||
select(Conversation)
|
||||
.options(selectinload(Conversation.messages))
|
||||
.where(Conversation.id == conversation_id)
|
||||
)
|
||||
return result.scalars().first()
|
||||
|
||||
|
||||
async def list_conversations(
|
||||
limit: int = 50, offset: int = 0
|
||||
) -> tuple[list[Conversation], int]:
|
||||
async with async_session() as session:
|
||||
total = await session.scalar(
|
||||
select(func.count(Conversation.id))
|
||||
) or 0
|
||||
|
||||
result = await session.execute(
|
||||
select(Conversation)
|
||||
.options(selectinload(Conversation.messages))
|
||||
.order_by(Conversation.updated_at.desc())
|
||||
.limit(limit)
|
||||
.offset(offset)
|
||||
)
|
||||
conversations = list(result.scalars().all())
|
||||
return conversations, total
|
||||
|
||||
|
||||
async def delete_conversation(conversation_id: int) -> bool:
|
||||
async with async_session() as session:
|
||||
conv = await session.get(Conversation, conversation_id)
|
||||
if conv is None:
|
||||
return False
|
||||
await session.delete(conv)
|
||||
await session.commit()
|
||||
return True
|
||||
|
||||
|
||||
async def update_conversation_title(
|
||||
conversation_id: int, title: str
|
||||
) -> Conversation | None:
|
||||
async with async_session() as session:
|
||||
conv = await session.get(Conversation, conversation_id)
|
||||
if conv is None:
|
||||
return None
|
||||
conv.title = title
|
||||
conv.updated_at = datetime.now(timezone.utc)
|
||||
await session.commit()
|
||||
await session.refresh(conv)
|
||||
return conv
|
||||
|
||||
|
||||
async def add_message(
|
||||
conversation_id: int,
|
||||
role: str,
|
||||
content: str,
|
||||
context_note_id: int | None = None,
|
||||
) -> Message:
|
||||
async with async_session() as session:
|
||||
msg = Message(
|
||||
conversation_id=conversation_id,
|
||||
role=role,
|
||||
content=content,
|
||||
context_note_id=context_note_id,
|
||||
)
|
||||
session.add(msg)
|
||||
# Touch conversation updated_at
|
||||
conv = await session.get(Conversation, conversation_id)
|
||||
if conv:
|
||||
conv.updated_at = datetime.now(timezone.utc)
|
||||
await session.commit()
|
||||
await session.refresh(msg)
|
||||
return msg
|
||||
|
||||
|
||||
async def get_message(message_id: int) -> Message | None:
|
||||
async with async_session() as session:
|
||||
return await session.get(Message, message_id)
|
||||
|
||||
|
||||
async def save_response_as_note(message_id: int) -> dict:
|
||||
"""Create a note from an assistant message. Returns the new note dict."""
|
||||
msg = await get_message(message_id)
|
||||
if msg is None:
|
||||
raise ValueError("Message not found")
|
||||
if msg.role != "assistant":
|
||||
raise ValueError("Can only save assistant messages as notes")
|
||||
|
||||
# Use first line as title, rest as body
|
||||
lines = msg.content.strip().split("\n", 1)
|
||||
title = lines[0].strip().lstrip("# ")[:100]
|
||||
body = msg.content
|
||||
|
||||
note = await create_note(title=title, body=body)
|
||||
return note.to_dict()
|
||||
|
||||
|
||||
async def summarize_conversation_as_note(
|
||||
conversation_id: int, model: str
|
||||
) -> dict:
|
||||
"""Summarize a conversation using the LLM and save as a note."""
|
||||
conv = await get_conversation(conversation_id)
|
||||
if conv is None:
|
||||
raise ValueError("Conversation not found")
|
||||
|
||||
# Build the conversation text
|
||||
conv_text = []
|
||||
for msg in conv.messages:
|
||||
if msg.role == "system":
|
||||
continue
|
||||
label = "User" if msg.role == "user" else "Assistant"
|
||||
conv_text.append(f"{label}: {msg.content}")
|
||||
|
||||
prompt_messages = [
|
||||
{
|
||||
"role": "system",
|
||||
"content": (
|
||||
"Summarize the following conversation into a concise note. "
|
||||
"Include key points, decisions, and any action items. "
|
||||
"Format the summary in markdown."
|
||||
),
|
||||
},
|
||||
{"role": "user", "content": "\n\n".join(conv_text)},
|
||||
]
|
||||
|
||||
summary = await generate_completion(prompt_messages, model)
|
||||
|
||||
title = conv.title or "Conversation Summary"
|
||||
title = f"Summary: {title}"
|
||||
|
||||
note = await create_note(title=title, body=summary)
|
||||
return note.to_dict()
|
||||
@@ -0,0 +1,194 @@
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
from collections.abc import AsyncGenerator
|
||||
|
||||
import httpx
|
||||
|
||||
from fabledassistant.config import Config
|
||||
from fabledassistant.services.notes import get_note, list_notes
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
STOP_WORDS = frozenset({
|
||||
"a", "an", "the", "is", "it", "to", "in", "for", "of", "and", "or",
|
||||
"on", "at", "by", "with", "from", "as", "be", "was", "were", "been",
|
||||
"are", "am", "do", "does", "did", "have", "has", "had", "will", "would",
|
||||
"can", "could", "shall", "should", "may", "might", "must", "that",
|
||||
"this", "these", "those", "i", "me", "my", "you", "your", "he", "she",
|
||||
"we", "they", "them", "his", "her", "its", "our", "their", "what",
|
||||
"which", "who", "whom", "how", "when", "where", "why", "not", "no",
|
||||
"but", "if", "so", "than", "too", "very", "just", "about", "up",
|
||||
})
|
||||
|
||||
|
||||
async def ensure_model(model: str) -> None:
|
||||
"""Check if model exists in Ollama, pull if missing."""
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||||
resp = await client.get(f"{Config.OLLAMA_URL}/api/tags")
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
names = {m["name"] for m in data.get("models", [])}
|
||||
# Check both with and without :latest tag
|
||||
if model in names or f"{model}:latest" in names:
|
||||
logger.info("Model '%s' already available", model)
|
||||
return
|
||||
except Exception:
|
||||
logger.warning("Failed to check Ollama models, attempting pull anyway")
|
||||
|
||||
logger.info("Pulling model '%s' from Ollama...", model)
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=600.0) as client:
|
||||
async with client.stream(
|
||||
"POST",
|
||||
f"{Config.OLLAMA_URL}/api/pull",
|
||||
json={"name": model},
|
||||
) as resp:
|
||||
resp.raise_for_status()
|
||||
async for line in resp.aiter_lines():
|
||||
if line.strip():
|
||||
status = json.loads(line)
|
||||
if "status" in status:
|
||||
logger.info("Pull %s: %s", model, status["status"])
|
||||
logger.info("Model '%s' pulled successfully", model)
|
||||
except Exception:
|
||||
logger.warning("Failed to pull model '%s' — chat may not work", model, exc_info=True)
|
||||
|
||||
|
||||
async def stream_chat(
|
||||
messages: list[dict], model: str
|
||||
) -> AsyncGenerator[str, None]:
|
||||
"""Stream chat completion from Ollama, yielding content chunks."""
|
||||
async with httpx.AsyncClient(timeout=httpx.Timeout(300.0, connect=10.0)) as client:
|
||||
async with client.stream(
|
||||
"POST",
|
||||
f"{Config.OLLAMA_URL}/api/chat",
|
||||
json={"model": model, "messages": messages, "stream": True},
|
||||
) as resp:
|
||||
resp.raise_for_status()
|
||||
async for line in resp.aiter_lines():
|
||||
if not line.strip():
|
||||
continue
|
||||
data = json.loads(line)
|
||||
chunk = data.get("message", {}).get("content", "")
|
||||
if chunk:
|
||||
yield chunk
|
||||
if data.get("done"):
|
||||
break
|
||||
|
||||
|
||||
async def generate_completion(messages: list[dict], model: str) -> str:
|
||||
"""Non-streaming chat completion, returns full response text."""
|
||||
async with httpx.AsyncClient(timeout=httpx.Timeout(300.0, connect=10.0)) as client:
|
||||
resp = await client.post(
|
||||
f"{Config.OLLAMA_URL}/api/chat",
|
||||
json={"model": model, "messages": messages, "stream": False},
|
||||
)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
return data.get("message", {}).get("content", "")
|
||||
|
||||
|
||||
async def fetch_url_content(url: str) -> str:
|
||||
"""Fetch a URL and return text content (HTML tags stripped)."""
|
||||
try:
|
||||
async with httpx.AsyncClient(
|
||||
timeout=15.0, follow_redirects=True, headers={"User-Agent": "FabledAssistant/1.0"}
|
||||
) as client:
|
||||
resp = await client.get(url)
|
||||
resp.raise_for_status()
|
||||
text = resp.text
|
||||
# Strip HTML tags
|
||||
text = re.sub(r"<script[^>]*>.*?</script>", "", text, flags=re.DOTALL)
|
||||
text = re.sub(r"<style[^>]*>.*?</style>", "", text, flags=re.DOTALL)
|
||||
text = re.sub(r"<[^>]+>", " ", text)
|
||||
# Collapse whitespace
|
||||
text = re.sub(r"\s+", " ", text).strip()
|
||||
# Truncate to reasonable size
|
||||
if len(text) > 4000:
|
||||
text = text[:4000] + "..."
|
||||
return text
|
||||
except Exception as e:
|
||||
logger.warning("Failed to fetch URL %s: %s", url, e)
|
||||
return f"[Failed to fetch URL: {url}]"
|
||||
|
||||
|
||||
def _extract_keywords(text: str) -> list[str]:
|
||||
"""Extract meaningful keywords from text for note search."""
|
||||
words = re.findall(r"\b[a-zA-Z]{3,}\b", text.lower())
|
||||
keywords = [w for w in words if w not in STOP_WORDS]
|
||||
# Deduplicate while preserving order
|
||||
seen: set[str] = set()
|
||||
unique = []
|
||||
for w in keywords:
|
||||
if w not in seen:
|
||||
seen.add(w)
|
||||
unique.append(w)
|
||||
return unique[:5]
|
||||
|
||||
|
||||
def _find_urls(text: str) -> list[str]:
|
||||
"""Find URLs in text."""
|
||||
return re.findall(r"https?://[^\s<>\"')\]]+", text)
|
||||
|
||||
|
||||
async def build_context(
|
||||
history: list[dict],
|
||||
current_note_id: int | None,
|
||||
user_message: str,
|
||||
) -> list[dict]:
|
||||
"""Build messages array for Ollama with system prompt and context."""
|
||||
system_parts = [
|
||||
"You are a helpful assistant integrated into a note-taking and task-tracking app called Fabled Assistant. "
|
||||
"Help users with their notes, tasks, and general questions. "
|
||||
"When note context is provided, use it to give relevant answers."
|
||||
]
|
||||
|
||||
# Include current note context if provided
|
||||
if current_note_id:
|
||||
note = await get_note(current_note_id)
|
||||
if note:
|
||||
system_parts.append(
|
||||
f"\n\n--- Current Note ---\n"
|
||||
f"Title: {note.title}\n"
|
||||
f"Content:\n{note.body}\n"
|
||||
f"--- End Note ---"
|
||||
)
|
||||
|
||||
# Search notes by keywords from user message
|
||||
keywords = _extract_keywords(user_message)
|
||||
if keywords:
|
||||
search_q = " ".join(keywords[:3])
|
||||
try:
|
||||
notes, _ = await list_notes(q=search_q, limit=3)
|
||||
if notes:
|
||||
snippets = []
|
||||
for n in notes:
|
||||
# Skip the current note (already included)
|
||||
if current_note_id and n.id == current_note_id:
|
||||
continue
|
||||
body_preview = n.body[:300] if n.body else ""
|
||||
snippets.append(f"- {n.title}: {body_preview}")
|
||||
if snippets:
|
||||
system_parts.append(
|
||||
"\n\n--- Related Notes ---\n"
|
||||
+ "\n".join(snippets)
|
||||
+ "\n--- End Related Notes ---"
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Fetch URL content from user message
|
||||
urls = _find_urls(user_message)
|
||||
for url in urls[:2]: # Limit to 2 URLs
|
||||
content = await fetch_url_content(url)
|
||||
if content and not content.startswith("[Failed"):
|
||||
system_parts.append(
|
||||
f"\n\n--- Content from {url} ---\n{content}\n--- End URL Content ---"
|
||||
)
|
||||
|
||||
messages = [{"role": "system", "content": "".join(system_parts)}]
|
||||
messages.extend(history)
|
||||
messages.append({"role": "user", "content": user_message})
|
||||
return messages
|
||||
Reference in New Issue
Block a user