5f3169bad4
Adds ResolutionOutputGenerator service that generates PSA ticket notes, knowledge base article draft, and client summary on session resolve, plus integration tests for generate_all and edit_output. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
119 lines
4.4 KiB
Python
119 lines
4.4 KiB
Python
"""Resolution output generator — three deliverables on session resolve."""
|
|
import logging
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.ai_session import AISession
|
|
from app.models.session_resolution_output import SessionResolutionOutput
|
|
from app.services.assistant_chat_service import _call_ai
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
RESOLUTION_MODEL = "claude-sonnet-4-6"
|
|
|
|
|
|
class ResolutionOutputGenerator:
|
|
def __init__(self, db: AsyncSession):
|
|
self.db = db
|
|
|
|
async def generate_all(self, session_id: UUID) -> list[SessionResolutionOutput]:
|
|
result = await self.db.execute(
|
|
select(AISession).where(AISession.id == session_id)
|
|
)
|
|
session = result.scalar_one_or_none()
|
|
if not session:
|
|
raise ValueError(f"Session {session_id} not found")
|
|
|
|
context = self._build_session_context(session)
|
|
|
|
outputs = []
|
|
for output_type, prompt in [
|
|
("psa_ticket_notes", self._psa_notes_prompt(context)),
|
|
("knowledge_base", self._kb_article_prompt(context)),
|
|
("client_summary", self._client_summary_prompt(context)),
|
|
]:
|
|
content, _, _ = await _call_ai(
|
|
system_base="You are a technical documentation assistant for MSP teams.",
|
|
rag_context="",
|
|
history=[],
|
|
new_message=prompt,
|
|
max_tokens=2048,
|
|
)
|
|
|
|
output = SessionResolutionOutput(
|
|
session_id=session_id,
|
|
output_type=output_type,
|
|
generated_content=content,
|
|
status="draft",
|
|
generated_by_model=RESOLUTION_MODEL,
|
|
)
|
|
self.db.add(output)
|
|
outputs.append(output)
|
|
|
|
await self.db.flush()
|
|
return outputs
|
|
|
|
async def edit_output(self, output_id: UUID, edited_content: str) -> SessionResolutionOutput:
|
|
result = await self.db.execute(
|
|
select(SessionResolutionOutput).where(SessionResolutionOutput.id == output_id)
|
|
)
|
|
output = result.scalar_one_or_none()
|
|
if not output:
|
|
raise ValueError(f"Output {output_id} not found")
|
|
output.edited_content = edited_content
|
|
await self.db.flush()
|
|
return output
|
|
|
|
async def push_output(self, output_id: UUID, destination: str) -> SessionResolutionOutput:
|
|
result = await self.db.execute(
|
|
select(SessionResolutionOutput).where(SessionResolutionOutput.id == output_id)
|
|
)
|
|
output = result.scalar_one_or_none()
|
|
if not output:
|
|
raise ValueError(f"Output {output_id} not found")
|
|
|
|
from datetime import datetime, timezone
|
|
output.status = "pushed"
|
|
output.pushed_to = destination
|
|
output.pushed_at = datetime.now(timezone.utc)
|
|
await self.db.flush()
|
|
return output
|
|
|
|
def _build_session_context(self, session: AISession) -> str:
|
|
parts = [
|
|
f"Problem: {session.problem_summary or 'Unknown'}",
|
|
f"Domain: {session.problem_domain or 'Unknown'}",
|
|
f"Resolution: {session.resolution_summary or 'Not specified'}",
|
|
f"Steps taken: {session.step_count}",
|
|
]
|
|
msgs = session.conversation_messages or []
|
|
if msgs:
|
|
parts.append("\nConversation highlights:")
|
|
for msg in msgs[-10:]:
|
|
role = msg.get("role", "unknown")
|
|
content = msg.get("content", "")[:200]
|
|
parts.append(f" [{role}]: {content}")
|
|
return "\n".join(parts)
|
|
|
|
def _psa_notes_prompt(self, context: str) -> str:
|
|
return (
|
|
f"Generate professional PSA ticket notes for this resolved troubleshooting session.\n"
|
|
f"Format as structured markdown with: Problem, Diagnostic Steps, Resolution, Recommendations.\n\n{context}"
|
|
)
|
|
|
|
def _kb_article_prompt(self, context: str) -> str:
|
|
return (
|
|
f"Generate a knowledge base article draft from this resolved session.\n"
|
|
f"Include: Symptoms, Root Cause, Resolution Steps, Things to Rule Out First.\n\n{context}"
|
|
)
|
|
|
|
def _client_summary_prompt(self, context: str) -> str:
|
|
return (
|
|
f"Generate a non-technical summary for the end user/client.\n"
|
|
f"Explain what was wrong and what was done to fix it in plain language.\n"
|
|
f"No jargon. 2-3 paragraphs max.\n\n{context}"
|
|
)
|