Call Transcript-to-PRD-to-Ticket Agent: Converting Meeting Transcripts to Linear Tickets using Mistral AI LLMs

from mistralai import Mistral
from gql import gql, Client
from gql.transport.requests import RequestsHTTPTransport
from pydantic import BaseModel
from typing import List, Dict, Optional, Any
from dataclasses import dataclass
from pypdf import PdfReader
import json

@dataclass
class Config:
    """Configuration settings for the application."""
    LINEAR_API_KEY: str # OAuth token for Linear API authentication
    LINEAR_TEAM_ID: str # Unique identifier for your Linear team/project
    LINEAR_GRAPHQL_URL: str # Linear's GraphQL API endpoint (usually "https://api.linear.app/graphql")
    MISTRAL_API_KEY: str # API Key for accessing Mistral LLMs
    MISTRAL_MODEL: str # Specific Mistral model to use (e.g., "mistral-large-latest")

config = Config(
    LINEAR_API_KEY = "YOUR API KEY ON LINEAR",
    LINEAR_TEAM_ID = "YOUR TEAM ID ON LINEAR",
    LINEAR_GRAPHQL_URL = "https://api.linear.app/graphql",
    MISTRAL_API_KEY = "YOUR MISTRAL API KEY", # Get your API key from https://console.mistral.ai/api-keys/
    MISTRAL_MODEL = "ministral-large-latest",
)

class FeaturesList(BaseModel):
    """Pydantic model for structured feature data."""
    Features: List[str]
    DescriptionOfFeatures: List[str]

class PRDAgent:
    """Agent responsible for generating and refining PRD from transcripts."""

    def __init__(self, transcript: str, mistral_client: Mistral, model: str = "mistral-large-latest"):
        """
        Initialize PRD agent.

        Args:
            transcript (str): Call transcript text
            mistral_client (Mistral): Initialized Mistral client
            model (str): Model name to use
        """
        self.transcript: str = transcript
        self.prd: Optional[str] = None
        self.feedback: Optional[str] = None
        self.client: Mistral = mistral_client
        self.model: str = model

    def generate_initial_prd(self) -> str:
        """
        Generate initial PRD from transcript.

        Returns:
            str: Generated PRD text
        """
        prompt = f"""
        Based on the following call transcript, create an initial Product Requirements Document (PRD) with some or all of these sections:
        1. Title
        2. Purpose
        3. Scope
        4. Features and Requirements
        5. User Personas
        6. Technical Requirements
        7. Constraints
        8. Success Metrics
        9. Timeline and Milestones

        Transcript:
        {self.transcript}

        Align everything only with the information provided in the transcript. If any section is not present in the transcript, you can skip it in the PRD.

        PRD:
        """
        response = self.client.chat.complete(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.1
        )
        self.prd = response.choices[0].message.content
        return self.prd

    def get_feedback(self) -> str:
        """
        Get feedback on current PRD.

        Returns:
            str: Feedback text
        """
        prompt = f"""
            Review the following Product Requirements Document (PRD) based on the original call transcript. Provide feedback on:
            - Missing information in PRD that are present in the transcript.
            - Inconsistencies in the PRD that are not aligned with the transcript.

            Transcript:
            {self.transcript}

            Current PRD:
            {self.prd}

            Align the feedback only with the information provided in the transcript. We are not looking for additional information based on your knowledge.

            If no feedback is required, respond with "None." and don't provide any further feedback. Your task is only to review the alignment between the PRD and the transcript and provide feedback based on that. Don't refine the PRD at this stage.

            Feedback:
            """
        response = self.client.chat.complete(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.1
        )
        self.feedback = response.choices[0].message.content
        return self.feedback

    def refine_prd(self) -> str:
        """
        Refine PRD based on feedback.

        Returns:
            str: Refined PRD text
        """
        prompt = f"""
        Refine the PRD based on the provided feedback and aligning it with the transcript:

        Current PRD:
        {self.prd}

        Feedback:
        {self.feedback}

        Transcript:
        {self.transcript}
        """
        response = self.client.chat.complete(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.1
        )
        self.prd = response.choices[0].message.content
        return self.prd

    def run(self, max_iterations: int = 3) -> str:
        """
        Run the PRD generation and refinement process.

        Args:
            max_iterations (int): Maximum number of refinement iterations

        Returns:
            str: Final PRD text
        """
        print("Generating initial PRD...")
        self.generate_initial_prd()
        print(f"Initial PRD:\n{self.prd}")

        for iteration in range(max_iterations):
            print(f"\nIteration {iteration}: Requesting feedback...")
            feedback = self.get_feedback()
            print(f"Feedback:\n{feedback}")

            if "none" in feedback.strip().lower():
                print("\nNo further feedback. Finalizing PRD...")
                break

            print("\nRefining PRD...")
            self.refine_prd()
            print(f"Refined PRD:\n{self.prd}")

        return self.prd

class TicketCreationAgent:
    """Agent responsible for creating Linear tickets from PRD."""

    def __init__(self, api_key: str, team_id: str, mistral_client: Mistral, graphql_url: str):
        """
        Initialize Linear ticket agent.

        Args:
            api_key (str): Linear API key
            team_id (str): Linear team ID
            mistral_client (Mistral): Initialized Mistral client
            graphql_url (str): Linear GraphQL API URL
        """
        self.client = Client(
            transport=RequestsHTTPTransport(
                url=graphql_url,
                headers={'Authorization': api_key},
                verify=True,
                retries=3
            ),
            fetch_schema_from_transport=True
        )
        self.team_id = team_id
        self.mistral_client = mistral_client

    def parse_prd(self, prd_text: str) -> Dict[str, List[str]]:
        """
        Parse PRD into structured feature data.

        Args:
            prd_text (str): PRD text to parse

        Returns:
            Dict[str, List[str]]: Structured feature data
        """
        messages = [
            {
                "role": "system",
                "content": (
                    "You are an AI assistant helping to create Features list and their descriptions from a Product Requirements Document (PRD)."
                    "The description should contain a brief explanation of the feature that includes Technical requirements (if any), Constraints (if any), Success metrics (if any), User personas (if any), and Timeline and Milestones (if any)."
                )
            },
            {
                "role": "user",
                "content": f"PRD:\n\n{prd_text}"
            }
        ]

        chat_response = self.mistral_client.chat.parse(
            model="mistral-large-latest",
            messages=messages,
            response_format=FeaturesList,
            max_tokens=2048,
            temperature=0.1
        )

        return json.loads(chat_response.choices[0].message.content)

    def create_ticket(self, title: str, description: str) -> Dict[str, Any]:
        """
        Create a single Linear ticket.

        Args:
            title (str): Ticket title
            description (str): Ticket description

        Returns:
            Dict[str, Any]: Creation result from Linear API
        """
        mutation = gql("""
        mutation CreateIssue($title: String!, $description: String!, $teamId: String!) {
            issueCreate(
                input: {
                    title: $title,
                    description: $description,
                    teamId: $teamId
                }
            ) {
                success
                issue {
                    id
                    url
                }
            }
        }
        """)

        variables = {
            "title": title,
            "description": description,
            "teamId": self.team_id
        }

        result = self.client.execute(mutation, variable_values=variables)
        print(f"Created ticket: {result['issueCreate']['issue']['url']}")
        return result

    def create_tickets_from_prd(self, parsed_items: Dict[str, List[str]]) -> List[Dict[str, Any]]:
        """
        Create Linear tickets from parsed PRD items.

        Args:
            parsed_items (Dict[str, List[str]]): Parsed feature data

        Returns:
            List[Dict[str, Any]]: List of ticket creation results
        """
        results = []
        for title, description in zip(
            parsed_items['Features'],
            parsed_items['DescriptionOfFeatures']
        ):
            result = self.create_ticket(title, description)
            results.append(result)
        return results

class WorkflowOrchestrator:
    """Orchestrates the entire workflow from transcript to Linear tickets."""

    def __init__(self, config: Config, transcript: str):
        """
        Initialize workflow orchestrator.

        Args:
            config (Config): Application configuration
            transcript (str): Call transcript text
        """
        mistral_client = Mistral(api_key=config.MISTRAL_API_KEY)
        self.prd_agent = PRDAgent(
            transcript=transcript,
            mistral_client=mistral_client
        )
        self.linear_agent = TicketCreationAgent(
            api_key=config.LINEAR_API_KEY,
            team_id=config.LINEAR_TEAM_ID,
            mistral_client=mistral_client,
            graphql_url=config.LINEAR_GRAPHQL_URL
        )

    def run(self) -> Dict[str, Any]:
        """
        Run the complete workflow.

        Returns:
            Dict[str, Any]: Workflow results including PRD and ticket data
        """
        print("Generating and finalizing PRD...")
        prd = self.prd_agent.run()

        print("\nParsing PRD into actionable items...")
        parsed_items = self.linear_agent.parse_prd(prd)

        print("\nCreating Linear tickets...")
        ticket_results = self.linear_agent.create_tickets_from_prd(parsed_items)

        return {
            "prd": prd,
            "parsed_items": parsed_items,
            "ticket_results": ticket_results
        }

def parse_transcript(config: Config, file_path: str) -> str:
  """Parse a transcriot PDF file and extract text from all pages using Mistral OCR."""

  mistral_client = Mistral(api_key=config.MISTRAL_API_KEY)

  uploaded_pdf = mistral_client.files.upload(
      file={
          "file_name": file_path,
          "content": open(file_path, "rb"),
      },
      purpose="ocr"
  )

  signed_url = mistral_client.files.get_signed_url(file_id=uploaded_pdf.id)

  ocr_response = mistral_client.ocr.process(
      model="mistral-ocr-latest",
      document={
          "type": "document_url",
          "document_url": signed_url.url,
      }
  )

  text = "\n".join([x.markdown for x in (ocr_response.pages)])

  return text

transcript = parse_transcript(config, "./lechat_product_call_trascript.pdf")