EAP – How You Handle Conversation Memory?
Table Of Contents:
app/api/routes.py
from uuid import uuid4
from fastapi import APIRouter, Depends
from app.api.dependencies import get_agent_graph, get_memory
from app.core.config import get_settings
from app.middleware.auth import require_user
from app.schemas.chat import ChatRequest, ChatResponse
from app.schemas.incident import IncidentCreate
from app.schemas.response import HealthResponse
from app.tools.incident_tools import create_incident
router = APIRouter()
@router.get("/health", response_model=HealthResponse)
async def health():
settings = get_settings()
return HealthResponse(status="ok", environment=settings.app_env)
@router.post("/v1/chat", response_model=ChatResponse)
async def chat(request: ChatRequest, user: dict = Depends(require_user)):
conversation_id = request.conversation_id or str(uuid4())
memory = get_memory()
memory.add(conversation_id, "user", request.message)
result = await get_agent_graph().ainvoke({"message": request.message, "conversation_id": conversation_id})
memory.add(conversation_id, "assistant", result["answer"])
return ChatResponse(answer=result["answer"], conversation_id=conversation_id, intent=result["intent"], sources=result.get("sources", []))
@router.post("/v1/incidents")
async def open_incident(payload: IncidentCreate, user: dict = Depends(require_user)):
return await create_incident(**payload.model_dump())app/api/dependencies.py
from functools import lru_cache
from app.graph.graph import build_graph
from app.memory.conversation_memory import ConversationMemory
@lru_cache
def get_agent_graph(): return build_graph()
@lru_cache
def get_memory(): return ConversationMemory()app/graph/graph.py
from langgraph.graph import END, START, StateGraph
from app.graph.state import AgentState
from app.graph.nodes import classify_node, incident_node, knowledge_node, request_node, general_node
from app.graph.router import route_intent
def build_graph():
graph = StateGraph(AgentState)
graph.add_node("classify", classify_node)
graph.add_node("incident", incident_node)
graph.add_node("knowledge", knowledge_node)
graph.add_node("request", request_node)
graph.add_node("general", general_node)
graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_intent, {"incident": "incident", "knowledge": "knowledge", "request": "request", "general": "general"})
for node in ("incident", "knowledge", "request", "general"):
graph.add_edge(node, END)
return graph.compile()app/memory/conversation_memory.py
from app.db.repositories import list_messages, save_message
class ConversationMemory:
def add(self, conversation_id: str, role: str, content: str) -> None:
save_message(conversation_id, role, content)
def history(self, conversation_id: str) -> list[dict]:
return list(reversed(list_messages(conversation_id)))app/db/repositories.py
from datetime import datetime, timezone
from app.db.session import get_connection
def save_message(conversation_id: str, role: str, content: str) -> None:
now = datetime.now(timezone.utc).isoformat()
with get_connection() as conn:
conn.execute("INSERT OR IGNORE INTO conversations(id, created_at) VALUES (?, ?)", (conversation_id, now))
conn.execute("INSERT INTO messages(conversation_id, role, content, created_at) VALUES (?, ?, ?, ?)", (conversation_id, role, content, now))
def list_messages(conversation_id: str, limit: int = 50) -> list[dict]:
with get_connection() as conn:
return [dict(row) for row in conn.execute("SELECT role, content, created_at FROM messages WHERE conversation_id=? ORDER BY id DESC LIMIT ?", (conversation_id, limit))]app/db/session.py
import sqlite3
from contextlib import contextmanager
from app.core.config import get_settings
@contextmanager
def get_connection():
conn = sqlite3.connect(get_settings().database_path)
conn.row_factory = sqlite3.Row
try:
yield conn
conn.commit()
finally:
conn.close()
def init_db() -> None:
with get_connection() as conn:
conn.execute("CREATE TABLE IF NOT EXISTS conversations (id TEXT PRIMARY KEY, created_at TEXT NOT NULL)")
conn.execute("CREATE TABLE IF NOT EXISTS messages (id INTEGER PRIMARY KEY AUTOINCREMENT, conversation_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, created_at TEXT NOT NULL)")app/core/config.py
from pathlib import Path
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
app_env: str = "development"
log_level: str = "INFO"
gcp_project_id: str = ""
gcp_location: str = "us-central1"
vertex_model: str = "gemini-2.0-flash-001"
servicenow_instance: str = ""
servicenow_username: str = ""
servicenow_password: str = ""
jwt_secret: str = "local-development-secret-change-me"
jwt_issuer: str = "servicenow-agentic-ai"
database_path: str = "./data/agent.db"
enable_write_operations: bool = False
def ensure_data_dir(self) -> None:
Path(self.database_path).parent.mkdir(parents=True, exist_ok=True)
@lru_cache
def get_settings() -> Settings:
settings = Settings()
settings.ensure_data_dir()
return settings
