Meeting Transcript Processing

import fenic as fc
from pydantic import BaseModel, Field
from typing import List, Optional

# Configure session
config = fc.SessionConfig(
    app_name="meeting_transcript_processing",
    semantic=fc.SemanticConfig(
        language_models={
            "nano": fc.OpenAILanguageModel(
                model_name="gpt-4.1-nano",
                rpm=500,
                tpm=200_000,
            )
        }
    ),
)

session = fc.Session.get_or_create(config)

# Parse transcripts using native function
parsed_transcripts_df = transcripts_df.with_column(
    "structured_transcript",
    fc.text.parse_transcript(fc.col("transcript"), "generic"),
)

# Explode structured transcript into individual segments
segments_df = (
    parsed_transcripts_df.explode("structured_transcript")
    .unnest("structured_transcript")
    .select(
        fc.col("meeting_id"),
        fc.col("meeting_type"),
        fc.col("speaker"),
        fc.col("start_time"),
        fc.col("content"),
    )
)

print("Individual speaking segments:")
segments_df.show(5)

# Technical entities schema
class TechnicalEntitiesSchema(BaseModel):
    services: List[str] = Field(
        description="Technical services or systems mentioned (e.g., user-service, auth-service)"
    )
    technologies: List[str] = Field(
        description="Technologies, databases, or tools mentioned (e.g., Redis, PostgreSQL, JWT)"
    )
    metrics: List[str] = Field(
        description="Performance metrics, numbers, or measurements mentioned"
    )
    incident_references: List[str] = Field(
        description="Incident IDs, ticket numbers, or reference numbers mentioned"
    )

# Action items schema
class ActionItemSchema(BaseModel):
    has_action_item: str = Field(
        description="Whether this segment contains an action item (yes/no)"
    )
    assignee: Optional[str] = Field(
        default=None,
        description="Person assigned to the action item (if any)"
    )
    task_description: str = Field(
        description="Description of the task or action to be completed"
    )
    deadline: Optional[str] = Field(
        default=None,
        description="When the task should be completed (if mentioned)"
    )

# Decisions schema
class DecisionSchema(BaseModel):
    has_decision: str = Field(
        description="Whether this segment contains a decision (yes/no)"
    )
    decision_summary: str = Field(
        description="Summary of the decision made"
    )
    decision_rationale: Optional[str] = Field(
        default=None,
        description="Why this decision was made (if mentioned)"
    )

# Extract all information from each segment
enriched_df = (
    segments_df.with_column(
        "technical_entities",
        fc.semantic.extract(fc.col("content"), TechnicalEntitiesSchema),
    )
    .with_column(
        "action_items",
        fc.semantic.extract(fc.col("content"), ActionItemSchema)
    )
    .with_column(
        "decisions",
        fc.semantic.extract(fc.col("content"), DecisionSchema)
    )
    .cache()
)

# Flatten the extracted data
insights_df = (
    enriched_df.unnest("technical_entities")
    .unnest("action_items")
    .unnest("decisions")
    .select(
        fc.col("meeting_id"),
        fc.col("meeting_type"),
        fc.col("speaker"),
        fc.col("content"),
        fc.col("services"),
        fc.col("technologies"),
        fc.col("has_action_item"),
        fc.col("assignee"),
        fc.col("task_description"),
        fc.col("deadline"),
        fc.col("has_decision"),
        fc.col("decision_summary"),
    )
)

# Extract action items summary
action_items_summary = insights_df.filter(
    fc.col("has_action_item") == "yes"
).select(
    fc.col("meeting_id"),
    fc.col("meeting_type"),
    fc.col("assignee"),
    fc.col("task_description"),
    fc.col("deadline"),
)

print("Action Items Summary:")
action_items_summary.show()

# Extract decisions summary
decisions_summary = insights_df.filter(
    fc.col("has_decision") == "yes"
).select(
    fc.col("meeting_id"),
    fc.col("meeting_type"),
    fc.col("decision_summary"),
    fc.col("decision_rationale"),
)

print("Decisions Summary:")
decisions_summary.show()

# Technology mentions across meetings
all_technologies = (
    insights_df.select(fc.col("meeting_id"), fc.col("technologies"))
    .explode("technologies")
    .filter(fc.col("technologies").is_not_null() & (fc.col("technologies") != ""))
    .group_by("technologies")
    .agg(fc.count(fc.col("meeting_id")).alias("mention_count"))
    .sort("mention_count", ascending=False)
)

print("Most Mentioned Technologies:")
all_technologies.show()

# Team member workload
assignee_workload = (
    insights_df.filter(fc.col("has_action_item") == "yes")
    .group_by("assignee")
    .agg(fc.count("*").alias("assigned_tasks"))
    .order_by(fc.col("assigned_tasks").desc())
)

print("Team Member Workload (Action Items):")
assignee_workload.show()

# Set your API key
export OPENAI_API_KEY="your-api-key-here"

# Run the pipeline
python transcript_processing.py