From f407a5118f0b81bfa85c8b1e1b2a7d4da5d7be32 Mon Sep 17 00:00:00 2001 From: furyhawk Date: Fri, 12 Jun 2026 20:54:57 +0800 Subject: [PATCH] feat: implement chat message persistence with Valkey; add session management and history retrieval --- backend/core/database.py | 80 ++++++++++++++++++++++++++++++++++++++++ backend/routes/chat.py | 76 +++++++++++++++++++++++++++++++++++--- frontend/src/App.tsx | 57 ++++++++++++++++++++++++---- frontend/src/api.ts | 25 ++++++++++++- 4 files changed, 225 insertions(+), 13 deletions(-) diff --git a/backend/core/database.py b/backend/core/database.py index e0cc97d..182f0a9 100644 --- a/backend/core/database.py +++ b/backend/core/database.py @@ -9,10 +9,20 @@ Usage:: valkey = await get_valkey() await valkey.set("key", "value") + + +Chat persistence (Valkey):: + + from backend.core.database import save_message, get_session_messages + + await save_message("session-1", "user", "Hello!") + messages = await get_session_messages("session-1") """ from __future__ import annotations +import json +from datetime import datetime, timezone from typing import AsyncIterator from redis.asyncio import Redis @@ -88,3 +98,73 @@ async def close_valkey() -> None: async def close_engine() -> None: """Dispose the SQLAlchemy engine (call on shutdown).""" await _engine.dispose() + + +# ── Chat persistence (Valkey) ───────────────────────────────────────────── + +_MESSAGES_KEY = "chat:{session_id}:messages" +_SESSIONS_SET = "chat:sessions" + + +async def save_message( + session_id: str, + role: str, + content: str, + valkey: Redis | None = None, +) -> None: + """Append a chat message to the session history in Valkey.""" + if valkey is None: + valkey = await get_valkey() + + msg = { + "role": role, + "content": content, + "timestamp": datetime.now(timezone.utc).isoformat(), + } + key = _MESSAGES_KEY.format(session_id=session_id) + async with valkey.pipeline(transaction=True) as pipe: + pipe.rpush(key, json.dumps(msg)) + pipe.sadd(_SESSIONS_SET, session_id) + await pipe.execute() + + +async def get_session_messages( + session_id: str, + valkey: Redis | None = None, +) -> list[dict[str, str]]: + """Return all messages for a session as a list of dicts. + + Each message has keys ``role``, ``content``, ``timestamp``. + Returns an empty list if the session does not exist. + """ + if valkey is None: + valkey = await get_valkey() + + key = _MESSAGES_KEY.format(session_id=session_id) + raw = await valkey.lrange(key, 0, -1) + return [json.loads(item) for item in raw] + + +async def list_sessions( + valkey: Redis | None = None, +) -> list[str]: + """Return all known session IDs.""" + if valkey is None: + valkey = await get_valkey() + + return sorted(await valkey.smembers(_SESSIONS_SET)) + + +async def delete_session( + session_id: str, + valkey: Redis | None = None, +) -> None: + """Delete a session and its messages from Valkey.""" + if valkey is None: + valkey = await get_valkey() + + key = _MESSAGES_KEY.format(session_id=session_id) + async with valkey.pipeline(transaction=True) as pipe: + pipe.delete(key) + pipe.srem(_SESSIONS_SET, session_id) + await pipe.execute() diff --git a/backend/routes/chat.py b/backend/routes/chat.py index 63bc0a3..403c911 100644 --- a/backend/routes/chat.py +++ b/backend/routes/chat.py @@ -1,11 +1,21 @@ -"""Chat API endpoint — sends prompts to the agent and returns replies.""" +"""Chat API endpoint — sends prompts to the agent and returns replies. + +Messages are persisted in Valkey so chat history survives server restarts. +""" from __future__ import annotations +import uuid + from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from backend.core.agent import AgentService +from backend.core.database import ( + get_session_messages, + list_sessions, + save_message, +) from backend.core.dependencies import get_agent_service router = APIRouter(prefix="/api/chat", tags=["chat"]) @@ -18,7 +28,21 @@ class ChatRequest(BaseModel): class ChatResponse(BaseModel): reply: str - session_id: str | None = None + session_id: str + + +class MessageOut(BaseModel): + role: str + content: str + timestamp: str + + +class SessionOut(BaseModel): + session_id: str + message_count: int + + +# ── POST /api/chat ──────────────────────────────────────────────────────── @router.post("", response_model=ChatResponse) @@ -26,9 +50,51 @@ async def chat_endpoint( body: ChatRequest, agent: AgentService = Depends(get_agent_service), ) -> ChatResponse: - """Send a user message to the agent and return its reply.""" + """Send a user message to the agent and return its reply. + + Saves both the user message and the assistant reply to Valkey. + Assigns a new UUID ``session_id`` if none was provided. + """ + session_id = body.session_id or uuid.uuid4().hex + try: - output = await agent.ask(body.message, session_id=body.session_id) - return ChatResponse(reply=output, session_id=body.session_id) + # Persist the user message. + await save_message(session_id, "user", body.message) + + # Ask the agent. + output = await agent.ask(body.message, session_id=session_id) + + # Persist the assistant reply. + await save_message(session_id, "assistant", output) + + return ChatResponse(reply=output, session_id=session_id) except Exception as exc: raise HTTPException(status_code=500, detail=str(exc)) + + +# ── GET /api/chat/history ───────────────────────────────────────────────── + + +class HistoryParams(BaseModel): + session_id: str + + +@router.get("/history", response_model=list[MessageOut]) +async def get_history(session_id: str) -> list[MessageOut]: + """Return all messages for a given session.""" + messages = await get_session_messages(session_id) + return [MessageOut(**msg) for msg in messages] + + +# ── GET /api/chat/sessions ──────────────────────────────────────────────── + + +@router.get("/sessions", response_model=list[SessionOut]) +async def sessions_list() -> list[SessionOut]: + """Return all known session IDs with their message counts.""" + ids = await list_sessions() + result: list[SessionOut] = [] + for sid in ids: + msgs = await get_session_messages(sid) + result.append(SessionOut(session_id=sid, message_count=len(msgs))) + return result diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index b1eab07..6bb9127 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -1,5 +1,5 @@ import { useState, useRef, useEffect } from "react"; -import { sendMessage, type ChatResponse } from "./api"; +import { sendMessage, getHistory, type MessageData } from "./api"; /* ------------------------------------------------------------------ */ /* Types */ @@ -10,19 +10,60 @@ interface Message { content: string; } +const STORAGE_KEY = "agent_alpha_session_id"; + /* ------------------------------------------------------------------ */ /* Component */ /* ------------------------------------------------------------------ */ export default function App() { - const [messages, setMessages] = useState([ - { role: "assistant", content: "Hello! I'm **Agent Alpha**. How can I help you today?" }, - ]); + const [messages, setMessages] = useState([]); const [input, setInput] = useState(""); const [loading, setLoading] = useState(false); - const [sessionId, setSessionId] = useState(); + const [sessionId, setSessionId] = useState(() => { + return localStorage.getItem(STORAGE_KEY) ?? ""; + }); const bottomRef = useRef(null); + /* Restore chat history on mount */ + useEffect(() => { + const sid = localStorage.getItem(STORAGE_KEY); + if (sid) { + getHistory(sid) + .then((history: MessageData[]) => { + if (history.length > 0) { + setMessages( + history.map((m) => ({ role: m.role, content: m.content })), + ); + } else { + setMessages([ + { + role: "assistant", + content: + "Hello! I'm **Agent Alpha**. How can I help you today?", + }, + ]); + } + }) + .catch(() => { + setMessages([ + { + role: "assistant", + content: + "Hello! I'm **Agent Alpha**. How can I help you today?", + }, + ]); + }); + } else { + setMessages([ + { + role: "assistant", + content: "Hello! I'm **Agent Alpha**. How can I help you today?", + }, + ]); + } + }, []); + /* Auto-scroll on new messages */ useEffect(() => { bottomRef.current?.scrollIntoView({ behavior: "smooth" }); @@ -38,8 +79,10 @@ export default function App() { setLoading(true); try { - const data: ChatResponse = await sendMessage(text, sessionId); - setSessionId(data.session_id ?? undefined); + const data = await sendMessage(text, sessionId || undefined); + // Persist session_id so future refreshes restore history. + localStorage.setItem(STORAGE_KEY, data.session_id); + setSessionId(data.session_id); setMessages((prev) => [ ...prev, { role: "assistant", content: data.reply }, diff --git a/frontend/src/api.ts b/frontend/src/api.ts index 731dabb..edf5848 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -1,10 +1,20 @@ const API_BASE = "/api"; +/* ── Types ──────────────────────────────────────────────────────── */ + export interface ChatResponse { reply: string; - session_id: string | null; + session_id: string; } +export interface MessageData { + role: "user" | "assistant"; + content: string; + timestamp: string; +} + +/* ── Send a message ────────────────────────────────────────────── */ + export async function sendMessage( message: string, sessionId?: string, @@ -20,3 +30,16 @@ export async function sendMessage( } return res.json(); } + +/* ── Load session history ──────────────────────────────────────── */ + +export async function getHistory(sessionId: string): Promise { + const res = await fetch( + `${API_BASE}/chat/history?session_id=${encodeURIComponent(sessionId)}`, + ); + if (!res.ok) { + const body = await res.json().catch(() => null); + throw new Error(body?.detail ?? `HTTP ${res.status}`); + } + return res.json(); +}