feat(transcription): [2f388f42] integrate prompt database and AI insights

Implements the core functionality for the AI-powered analysis of meeting transcripts in the Transcription Tool.

This commit introduces a new 'AI Insights' feature that allows users to generate various summaries and analyses from a transcript on demand.

- Creates a  to manage and version different AI prompts for tasks like generating meeting minutes, extracting action items, and creating sales summaries.
- Adds a new  responsible for orchestrating the analysis process: fetching the transcript, calling the Gemini API with the appropriate prompt, and caching the results in the database.
- Extends the FastAPI backend with a new endpoint  to trigger the insight generation.
- Updates the React frontend () with a new 'AI Insights' panel, including buttons to trigger the analyses and a modal to display the results.
- Updates the documentation () to reflect the new features, API endpoints, and version.
This commit is contained in:
2026-01-26 07:43:24 +00:00
parent f96235c607
commit e427ec19f2
5 changed files with 348 additions and 10 deletions

View File

@@ -0,0 +1,110 @@
import sys
import os
from sqlalchemy.orm import Session
from .. import database
from .. import prompt_library
# Add project root to path to allow importing from 'helpers'
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', '..')))
from helpers import call_gemini_flash
def _format_transcript(chunks: list[database.TranscriptChunk]) -> str:
"""
Formats the transcript chunks into a single, human-readable string.
Example: "[00:00:01] Speaker A: Hello world."
"""
full_transcript = []
# Sort chunks by their index to ensure correct order
sorted_chunks = sorted(chunks, key=lambda c: c.chunk_index)
for chunk in sorted_chunks:
if not chunk.json_content:
continue
for item in chunk.json_content:
# json_content can be a list of dicts
if isinstance(item, dict):
speaker = item.get('speaker', 'Unknown')
start_time = item.get('start', 0)
text = item.get('line', '')
# Format timestamp from seconds to HH:MM:SS
hours, remainder = divmod(int(start_time), 3600)
minutes, seconds = divmod(remainder, 60)
timestamp = f"{hours:02}:{minutes:02}:{seconds:02}"
full_transcript.append(f"[{timestamp}] {speaker}: {text}")
return "\n".join(full_transcript)
def get_prompt_by_type(insight_type: str) -> str:
"""
Returns the corresponding prompt from the prompt_library based on the type.
"""
if insight_type == "meeting_minutes":
return prompt_library.MEETING_MINUTES_PROMPT
elif insight_type == "action_items":
return prompt_library.ACTION_ITEMS_PROMPT
elif insight_type == "sales_summary":
return prompt_library.SALES_SUMMARY_PROMPT
else:
raise ValueError(f"Unknown insight type: {insight_type}")
def generate_insight(db: Session, meeting_id: int, insight_type: str) -> database.AnalysisResult:
"""
Generates a specific insight for a meeting, stores it, and returns it.
Checks for existing analysis to avoid re-generating.
"""
# 1. Check if the insight already exists
existing_insight = db.query(database.AnalysisResult).filter(
database.AnalysisResult.meeting_id == meeting_id,
database.AnalysisResult.prompt_key == insight_type
).first()
if existing_insight:
return existing_insight
# 2. Get the meeting and its transcript
meeting = db.query(database.Meeting).filter(database.Meeting.id == meeting_id).first()
if not meeting:
raise ValueError(f"Meeting with id {meeting_id} not found.")
if not meeting.chunks:
raise ValueError(f"Meeting with id {meeting_id} has no transcript chunks.")
# 3. Format the transcript and select the prompt
transcript_text = _format_transcript(meeting.chunks)
if not transcript_text.strip():
raise ValueError(f"Transcript for meeting {meeting_id} is empty.")
prompt_template = get_prompt_by_type(insight_type)
final_prompt = prompt_template.format(transcript_text=transcript_text)
# 4. Call the AI model
# Update meeting status
meeting.status = "ANALYZING"
db.commit()
try:
generated_text = call_gemini_flash(prompt=final_prompt, temperature=0.5)
# 5. Store the new insight
new_insight = database.AnalysisResult(
meeting_id=meeting_id,
prompt_key=insight_type,
result_text=generated_text
)
db.add(new_insight)
meeting.status = "COMPLETED"
db.commit()
db.refresh(new_insight)
return new_insight
except Exception as e:
meeting.status = "ERROR"
db.commit()
# Log the error properly in a real application
print(f"Error generating insight for meeting {meeting_id}: {e}")
raise