멀티 에이전트 실적 발표 분석 시스템
멀티 에이전트 실적 발표 분석 시스템 (Multi-Agent Earnings Call Analysis System, MAECAS)
기업의 실적 발표(earnings call)는 회사의 성과, 전략, 미래 전망에 대한 중요한 통찰을 담고 있어요. 하지만 이런 대본(transcript)은 길고 밀도가 높으며 다양한 주제를 다루기 때문에, 원하는 통찰을 효율적으로 뽑아내기가 쉽지 않아요. 이 노트북에서는 여러 전문 에이전트를 협력시켜 분기별 실적 발표를 포괄적으로 분석하는 MAECAS 시스템을 만들어 볼 거예요.
출처: 문서
본문
기업의 실적 발표는 회사의 성과, 전략, 미래 전망에 대한 중요한 통찰을 제공해요. 하지만 이런 대본은 길고 밀도가 높으며 다양한 주제를 다루기 때문에, 목표로 하는 통찰을 효율적으로 추출하는 것이 어려운 과제예요.
문제점 (The Problem)
분기별 실적 발표는 회사 성과, 전략, 전망에 대한 중요한 통찰을 제공하지만, 의미 있는 분석을 추출하는 데는 큰 어려움이 있어요.
- 실적 발표 대본은 길고 밀도가 높아서, 종종 20쪽이 넘는 복잡한 재무 논의로 이어져요.
- 핵심 통찰이 텍스트 곳곳에 명확한 정리 없이 흩어져 있어요.
- 이해관계자마다 필요로 하는 정보 유형이 달라요 (재무 지표, 전략 이니셔티브, 위험 요소 등).
- 분기 간 분석은 여러 콜에 걸쳐 진화하는 서사를 수동으로 추적해야 해요.
- 전통적인 수동 분석은 시간이 많이 들고 일관성이 없으며, 중요한 세부사항을 놓치기 쉬워요.
왜 중요한가 (Why This Matters)
투자자, 애널리스트, 비즈니스 리더에게 포괄적인 실적 발표 분석은 큰 가치를 제공해요.
- 시간 효율성 (Time Efficiency): 분석 시간을 며칠에서 몇 분으로 줄여요.
- 의사결정 지원 (Decision Support): 투자 및 전략 결정을 위한 구조화된 통찰을 제공해요.
- 포괄적 커버리지 (Comprehensive Coverage): 중요한 통찰을 놓치지 않도록 보장해요.
- 일관된 분석 (Consistent Analysis): 모든 대본에 동일한 분석 엄격성을 적용해요.
- 추세 탐지 (Trend Detection): 분기 간 패턴을 찾아내서 눈에 띄지 않았을 부분을 발견해요.
우리의 솔루션 (Our Solution)
**실적 발표 분석 오케스트레이터(Earnings Call Analysis Orchestrator)**는 멀티 에이전트 워크플로우를 통해 실적 발표 처리를 혁신해요. 이 워크플로우는:
- 전문 분석 에이전트를 사용해 분기별 대본에서 통찰을 추출해요.
- 포괄적인 보고서와 목표 지향적인 쿼리 응답을 모두 제공해요.
- 분기 간의 추세와 패턴을 식별해요.
- 실적 통찰의 구조화된 지식 베이스를 유지해요.
전문 분석 에이전트 (Specialized Analysis Agents)
우리 시스템은 조정된 전문 에이전트들을 사용해서 포괄적인 분석을 제공해요.
- 재무 에이전트 (Financial Agent): 매출 수치, 이익률, 성장 지표 등 정량적 성과 지표를 추출해요.
- 전략 에이전트 (Strategic Agent): 제품 로드맵, 시장 확장, 파트너십, 장기 비전 이니셔티브를 식별해요.
- 감정 에이전트 (Sentiment Agent): 여러 비즈니스 세그먼트에 걸쳐 경영진의 자신감, 어조, 열정을 평가해요.
- 위험 에이전트 (Risk Agent): 공급망, 시장, 규제상의 과제를 탐지하고 그 심각성과 완화 계획을 평가해요.
- 경쟁사 에이전트 (Competitor Agent): 경쟁 포지셔닝, 시장 점유율 논의, 차별화 전략을 추적해요.
- 시간 흐름 에이전트 (Temporal Agent): 분기 간 추세를 분석해서 비즈니스 궤적과 진화하는 우선순위를 파악해요.
워크플로우 오케스트레이션 (Workflow Orchestration)
오케스트레이터는 다음과 같은 중앙 조정자 역할을 해요.
- 고급 OCR을 사용해 대본 텍스트를 효율적으로 처리하고 캐싱해요.
- 분석 필요에 따라 전문 에이전트를 활성화해요.
- 구조화된 통찰을 중앙 지식 베이스에 저장해요.
- 경영 요약, 부문별 분석, 전망이 포함된 포괄적인 보고서를 생성해요.
- 분기별 관련 통찰을 활용해 특정 쿼리에 답변해요.
데이터셋 (Dataset)
데모 목적으로 2025년 NVIDIA의 분기별 실적 발표 대본을 사용해요.
- Q1 2025 Earnings Call Transcript
- Q2 2025 Earnings Call Transcript
- Q3 2025 Earnings Call Transcript
- Q4 2025 Earnings Call Transcript
이 대본들은 NVIDIA 경영진과 금융 애널리스트들의 상호작용, 재무 실적, 전략 이니셔티브, 시장 상황, 미래 예측 발언에 대한 논의를 담고 있어요.
Mistral AI 모델 (Mistral AI Models)
우리 구현에서는 Mistral AI의 LLM을 사용해요.
mistral-small-latest: 일반 분석과 응답 생성에 사용해요.mistral-large-latest: 구조화된 출력 생성에 사용해요.mistral-ocr-latest: PDF 대본 추출과 처리에 사용해요.
이 모듈식 접근 방식은 선택적 에이전트 활성화와 통찰 재사용을 통해 효율성을 유지하면서, 깊이 있는 보고서 생성과 목표 지향적인 질의응답을 모두 가능하게 해요.
솔루션 아키텍처 (Solution Architecture)
설치 (Installation)
LLM 사용을 위해 mistralai가 필요해요.
Python
!pip install mistralai
임포트 (Imports)
import os
import json
import hashlib
from datetime import datetime
from pathlib import Path
from typing import List, Dict, Any, Literal, Optional, Union
from abc import ABC, abstractmethod
from pydantic import BaseModel, Field
from mistralai.client import Mistral
from IPython.display import display, Markdown
API 키 설정 (Setup API Keys)
여기서 MistralAI API 키를 설정해요.
os.environ['MISTRAL_API_KEY'] = '<YOUR MISTRALAI API KEY>' # Get your API key from https://console.mistral.ai/api-keys/
api_key = os.environ.get('MISTRAL_API_KEY')
Mistral 클라이언트 초기화 (Initialize Mistral client)
여기서 Mistral 클라이언트를 초기화해요.
mistral_client = Mistral(api_key=os.environ.get("MISTRAL_API_KEY"))
데이터 다운로드 (Download Data)
2025년 NVIDIA의 분기별 실적 발표 대본을 사용할 거예요.
- Q1 2025 Earnings Call Transcript
- Q2 2025 Earnings Call Transcript
- Q3 2025 Earnings Call Transcript
- Q4 2025 Earnings Call Transcript
이 대본들은 NVIDIA 경영진의 재무 실적, 전략 이니셔티브, 시장 상황, 미래 예측 발언과 금융 애널리스트와의 상호작용을 담고 있어요.
Python
!wget "https://github.com/mistralai/cookbook/blob/main/mistral/agents/non_framework/earnings_calls/data/nvidia_earnings_2025_Q1.pdf" -O "nvidia_earnings_2025_Q1.pdf"
!wget "https://github.com/mistralai/cookbook/blob/main/mistral/agents/non_framework/earnings_calls/data/nvidia_earnings_2025_Q2.pdf" -O "nvidia_earnings_2025_Q2.pdf"
!wget "https://github.com/mistralai/cookbook/blob/main/mistral/agents/non_framework/earnings_calls/data/nvidia_earnings_2025_Q3.pdf" -O "nvidia_earnings_2025_Q3.pdf"
!wget "https://github.com/mistralai/cookbook/blob/main/mistral/agents/non_framework/earnings_calls/data/nvidia_earnings_2025_Q4.pdf" -O "nvidia_earnings_2025_Q4.pdf"
모델 초기화 (Initiate Models)
DEFAULT_MODEL- 일반 분석용STRUCTURED_MODEL- 구조화된 출력용OCR_MODEL- 실적 발표 문서 파싱용
DEFAULT_MODEL = "mistral-small-latest"
STRUCTURED_MODEL = "mistral-large-latest"
OCR_MODEL = "mistral-ocr-latest"
데이터 모델 (Data Models)
이 솔루션은 통찰을 구조화하고 추출하기 위해 전문화된 Pydantic 모델을 사용해요.
핵심 분석 모델 (Core Analysis Models)
- FinancialInsight: 재무 성과의 지표, 값, 신뢰도 점수를 담아요.
- StrategicInsight: 이니셔티브, 설명, 시간대, 중요도 등급을 나타내요.
- SentimentInsight: 주제 감정, 근거, 발언자 귀속을 추적해요.
- RiskInsight: 위험, 영향, 완화책, 심각도 점수를 문서화해요.
- CompetitorInsight: 시장 세그먼트, 포지셔닝, 경쟁 역학을 기록해요.
- TemporalInsight: 분기 간 추세, 패턴, 뒷받침 근거를 식별해요.
워크플로우 모델 (Workflow Models)
- QueryAnalysis: 사용자 쿼리에서 필요한 분기, 에이전트 유형, 분석 차원을 결정해요.
- ReportSection: 제목, 본문, 선택적 하위 섹션으로 보고서 콘텐츠를 구성해요.
응답 래퍼 (Response Wrappers)
각 분석 모델에는 FinancialInsightsResponse와 같은 대응하는 응답 래퍼가 있어서, 통찰을 Mistral API 파싱 기능과 호환되는 구조화된 형식으로 묶어줘요.
이 모델들은 감정 수준이나 추세 유형 같은 분류 필드에 Python의 Literal 타입을 사용해 엄격한 검증과 일관된 용어를 보장해요. 덕분에 분기 간 비교가 신뢰할 수 있고, 포괄적인 보고서와 목표 지향적인 쿼리 모두에 대해 여러 분석 차원에서 일관된 지식 추출, 저장, 검색이 가능해져요.
재무 통찰 (Financial Insight)
class FinancialInsight(BaseModel):
"""Financial insights extracted from transcript"""
metric_name: str = Field(description="Name of the financial metric")
value: Optional[str] = Field(description="Numerical or textual value of the metric")
context: str = Field(description="Surrounding context for the metric")
quarter: Literal["Q1", "Q2", "Q3", "Q4"] = Field(description="Quarter the insight relates to (e.g., Q1, Q2)")
confidence: float = Field(description="Confidence score for the insight with limits ge=0.0, le=1.0")
class FinancialInsightsResponse(BaseModel):
"""Wrapper for list of financial insights"""
insights: List[FinancialInsight] = Field(description="Collection of financial insights")
전략 통찰 (Strategic Insight)
class StrategicInsight(BaseModel):
"""Strategic insights about business direction"""
initiative: str = Field(description="Name of the strategic initiative")
description: str = Field(description="Details about the strategic initiative")
timeframe: Optional[str] = Field(description="Expected timeline for implementation")
quarter: Literal["Q1", "Q2", "Q3", "Q4"] = Field(description="Quarter the insight relates to (e.g., Q1, Q2, Q3, Q4)")
importance: int = Field(description="Importance rating with limits ge=1, le=5")
class StrategicInsightsResponse(BaseModel):
"""Wrapper for list of strategic insights"""
insights: List[StrategicInsight] = Field(description="Collection of strategic insights")
감정 통찰 (Sentiment Insight)
class SentimentInsight(BaseModel):
"""Insights about management sentiment"""
topic: str = Field(description="Subject matter being discussed")
sentiment: Literal["very negative", "negative", "neutral", "positive", "very positive"] = Field(description="Tone expressed by management")
evidence: str = Field(description="Quote or context supporting the sentiment analysis")
speaker: str = Field(description="Person who expressed the sentiment")
quarter: Literal["Q1", "Q2", "Q3", "Q4"] = Field(description="Quarter the insight relates to (e.g., Q1, Q2, Q3, Q4)")
class SentimentInsightsResponse(BaseModel):
"""Wrapper for list of sentiment insights"""
insights: List[SentimentInsight] = Field(description="Collection of sentiment insights")
위험 통찰 (Risk Insight)
class RiskInsight(BaseModel):
"""Identified risks or challenges"""
risk_factor: str = Field(description="Name or type of risk identified")
description: str = Field(description="Details about the risk")
potential_impact: str = Field(description="Possible consequences of the risk")
mitigation_mentioned: Optional[str] = Field(description="Strategies to address the risk")
quarter: Literal["Q1", "Q2", "Q3", "Q4"] = Field(description="Quarter the insight relates to (e.g., Q1, Q2, Q3, Q4)")
severity: int = Field(description="Severity rating with limits ge=1, le=5")
class RiskInsightsResponse(BaseModel):
"""Wrapper for list of risk insights"""
insights: List[RiskInsight] = Field(description="Collection of risk insights")
경쟁사 통찰 (Competitor Insight)
class CompetitorInsight(BaseModel):
"""Insights about competitive positioning"""
competitor: Optional[str] = Field(description="Name of the competitor company")
market_segment: str = Field(description="Specific market area being discussed")
positioning: str = Field(description="Competitive stance or market position")
quarter: Literal["Q1", "Q2", "Q3", "Q4"] = Field(description="Quarter the insight relates to (e.g., Q1, Q2, Q3, Q4)")
mentioned_by: str = Field(description="Person who mentioned the competitive information")
class CompetitorInsightsResponse(BaseModel):
"""Wrapper for list of competitor insights"""
insights: List[CompetitorInsight] = Field(description="Collection of competitor insights")
시간 흐름 통찰 (Temporal Insight)
class TemporalInsight(BaseModel):
"""Insights about trends across quarters"""
trend_type: Literal["growth", "decline", "stable", "volatile", "emerging", "fading"] = Field(description="Direction or pattern of the trend")
topic: str = Field(description="Subject matter of the trend")
description: str = Field(description="Explanation of the trend's significance")
quarters_observed: List[Literal["Q1", "Q2", "Q3", "Q4"]] = Field(description="Quarters where the trend appears")
supporting_evidence: str = Field(description="Data or quotes supporting the trend identification")
class TemporalInsightsResponse(BaseModel):
"""Wrapper for list of temporal insights"""
insights: List[TemporalInsight] = Field(description="Collection of temporal insights")
쿼리 분석 (Query Analysis)
class QueryAnalysis(BaseModel):
"""Analysis of user query to determine required components"""
quarters: List[str] = Field(description="List of quarters to analyze")
agent_types: List[str] = Field(description="List of agent types to use")
temporal_analysis_required: bool = Field(description="Whether temporal analysis across quarters is needed")
query_intent: str = Field(description="Brief description of user's intent")
보고서 섹션 (Report Section)
class ReportSection(BaseModel):
"""Section of the final report"""
title: str = Field(description="Heading for the report section")
content: str = Field(description="Main text content of the section")
subsections: Optional[List["ReportSection"]] = Field(description="Nested sections within this section.")
PDF 파서 (PDF Parser)
우리의 PDF 파서는 Mistral의 OCR 기능을 사용해 실적 발표 대본에서 고품질 텍스트를 추출하고, 파일 기반 캐싱 시스템을 구현해서 성능을 개선해요. 이 접근 방식은 반복 분석 시 최소한의 처리 오버헤드로 정확한 텍스트 추출을 가능하게 해요.
class PDFParser:
"""Parse a transcript PDF file and extract text from all pages using Mistral OCR."""
CACHE_DIR = Path("transcript_cache")
@staticmethod
def _ensure_cache_dir():
"""Make sure cache directory exists"""
PDFParser.CACHE_DIR.mkdir(exist_ok=True)
@staticmethod
def _get_cache_path(file_path: str) -> Path:
"""Get the path for a cached transcript file"""
# Create a hash of the file path to use as the cache filename
file_hash = hashlib.md5(file_path.encode()).hexdigest()
return PDFParser.CACHE_DIR / f"{file_hash}.txt"
@staticmethod
def read_transcript(file_path: str, mistral_client: Mistral) -> str:
"""Extract text from PDF transcript using Mistral OCR"""
print(f"Processing PDF file: {file_path}")
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=OCR_MODEL,
document={
"type": "document_url",
"document_url": signed_url.url,
}
)
text = "\n".join([x.markdown for x in (ocr_response.pages)])
return text
@staticmethod
def get_transcript_by_quarter(company: str, quarter: str, year: str, mistral_client: Mistral) -> str:
"""Get the transcript for a specific quarter"""
company_lower = company.lower()
file_path = f"{company_lower}_earnings_{year}_{quarter}.pdf"
PDFParser._ensure_cache_dir()
cache_path = PDFParser._get_cache_path(file_path)
# Check if transcript is in cache
if cache_path.exists():
print(f"Using cached transcript for {company} {year} {quarter}")
with open(cache_path, "r", encoding="utf-8") as f:
return f.read()
else:
try:
print(f"Parsing transcript for {company} {year} {quarter}")
transcript = PDFParser.read_transcript(file_path, mistral_client)
# Store in cache for future use
with open(cache_path, "w", encoding="utf-8") as f:
f.write(transcript)
print(f"Cached transcript for {company} {year} {quarter}")
return transcript
except Exception as e:
print(f"Error processing transcript: {str(e)}")
raise
통찰 저장소 (Insights Storage)
시스템에는 중앙 집중식 InsightsStore 컴포넌트가 포함돼 있어요.
- 추출된 모든 통찰의 영구 JSON 데이터베이스를 유지해요.
- 통찰을 유형(재무, 전략 등)과 분기별로 정리해요.
- 보고서 생성과 쿼리 응답 모두를 위한 효율적인 검색을 제공해요.
- 분석 결과를 캐싱해 중복 처리를 없애요.
class InsightsStore:
"""Centralized storage for insights across all quarters and analysis types"""
def __init__(self, company: str, year: str):
self.company = company.lower()
self.year = year
self.db_path = Path(f"{self.company}_{self.year}_insights.json")
self.insights = self._load_insights()
def _load_insights(self) -> Dict:
"""Load insights from database file or initialize if not exists"""
if self.db_path.exists():
with open(self.db_path, "r", encoding="utf-8") as f:
return json.load(f)
else:
return {
"financial": {},
"strategic": {},
"sentiment": {},
"risk": {},
"competitor": {},
"temporal": {}
}
def save_insights(self):
"""Save insights to database file"""
with open(self.db_path, "w", encoding="utf-8") as f:
json.dump(self.insights, f, indent=2)
def add_insights(self, insight_type: str, quarter: str, insights: List):
"""Add insights for a specific type and quarter"""
if quarter not in self.insights[insight_type]:
self.insights[insight_type][quarter] = []
# Convert insights to dictionaries for storage
insight_dicts = []
for insight in insights:
if hasattr(insight, "dict"):
insight_dicts.append(insight.dict())
elif isinstance(insight, dict):
insight_dicts.append(insight)
else:
insight_dicts.append({"content": str(insight)})
# Append new insights
self.insights[insight_type][quarter] = insight_dicts
self.save_insights()
def get_insights(self, insight_type=None, quarters=None):
"""Retrieve insights, optionally filtered by type and quarters"""
if insight_type is None:
return self.insights
if quarters is None:
return self.insights[insight_type]
filtered = {}
for q in quarters:
if q in self.insights[insight_type]:
filtered[q] = self.insights[insight_type][q]
return filtered
전문 에이전트 (Specialised Agents)
우리 분석은 각각 특정 통찰을 추출하는 다섯 개의 도메인 중심 에이전트에 의존해요.
- Financial Agent - 지표, 매출 수치, 마진, 성장률을 분석해요.
- Strategic Agent - 제품 로드맵, 시장 확장, R&D 투자를 식별해요.
- Sentiment Agent - 주제 전반에 걸친 경영진의 어조, 자신감 수준, 열정을 평가해요.
- Risk Agent - 심각도 등급과 함께 과제, 불확실성, 잠재적 위협을 탐지해요.
- Competitor Agent - 경쟁 포지셔닝, 시장 점유율 논의, 차별화 전략을 추적해요.
각 에이전트는 전문화된 프롬프트로 대본을 처리해서, 전체 분석에 들어가는 구조화된 통찰을 생성해요.
전문 에이전트의 베이스 클래스 (base class for specialised agents)
Agent
class Agent(ABC):
"""Base class for all specialized agents"""
def __init__(self):
self.client = mistral_client
@abstractmethod
def analyze(self, transcript: str, quarter: str) -> Any:
"""Analyze the transcript and return insights"""
pass
재무 에이전트 (Financial Agent)
class FinancialAgent(Agent):
"""Agent for financial analysis of earnings call transcripts"""
def analyze(self, transcript: str, quarter: str) -> List[FinancialInsight]:
"""Extract financial insights from transcript"""
system_prompt = """
You are a financial analyst focused on extracting key financial metrics and performance
indicators from the earnings call transcripts.
Focus on:
- Revenue figures (overall and by segment)
- Profit margins
- Growth rates
- Forward guidance
- Capital expenditures
- Cash flow metrics
- Any financial KPIs mentioned
Extract only facts that are explicitly stated in the transcript, with their proper context.
Keep the insights as short as possible.
"""
print(f"Extracting financial insights for quarter {quarter}...")
response = self.client.chat.parse(
model=STRUCTURED_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"Extract financial insights from this earnings call transcript for {quarter}:\n\n{transcript}"}
],
response_format=FinancialInsightsResponse,
temperature=0.1
)
print(f"Financial agent completed for {quarter}")
parsed_response = json.loads(response.choices[0].message.content)
return parsed_response['insights']
전략 에이전트 (Strategic Agent)
class StrategicAgent(Agent):
"""Agent for strategic analysis of earnings call transcripts"""
def analyze(self, transcript: str, quarter: str) -> List[StrategicInsight]:
"""Extract strategic insights from transcript"""
system_prompt = """
You are a business strategy analyst focused on the company's strategic direction.
Extract insights about:
- Product roadmaps
- Market expansions
- Strategic partnerships
- R&D investments
- Long-term vision
- Business model changes
- Market segments of focus
Focus on extracting concrete strategic initiatives and plans, not general statements.
Assign an importance score (1-5) based on how central it appears to the company's strategy.
"""
print(f"Extracting strategic insights for quarter {quarter}...")
response = self.client.chat.parse(
model=STRUCTURED_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"Extract strategic insights from this earnings call transcript for {quarter}:\n\n{transcript}"}
],
response_format=StrategicInsightsResponse,
temperature=0.1
)
print(f"Strategic agent completed for {quarter}")
parsed_response = json.loads(response.choices[0].message.content)
return parsed_response['insights']
감정 에이전트 (Sentiment Agent)
class SentimentAgent(Agent):
"""Agent for sentiment analysis of earnings call transcripts"""
def analyze(self, transcript: str, quarter: str) -> List[SentimentInsight]:
"""Extract sentiment insights from transcript"""
system_prompt = """
You are an expert in analyzing sentiment and tone in corporate communications.
Focus on:
- Management's confidence level
- Tone when discussing different business segments
- Enthusiasm for future prospects
- Concerns or hesitations
- Changes in sentiment when answering analyst questions
Extract specific topics and the sentiment expressed about them by specific speakers.
Use the transcript to find evidence of the sentiment you identify.
"""
print(f"Extracting sentiment insights for quarter {quarter}...")
response = self.client.chat.parse(
model=STRUCTURED_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"Extract sentiment insights from this earnings call transcript for {quarter}:\n\n{transcript}"}
],
response_format=SentimentInsightsResponse,
temperature=0.1
)
print(f"Sentiment agent completed for {quarter}")
parsed_response = json.loads(response.choices[0].message.content)
return parsed_response['insights']
위험 에이전트 (Risk Agent)
class RiskAgent(Agent):
"""Agent for risk analysis of earnings call transcripts"""
def analyze(self, transcript: str, quarter: str) -> List[RiskInsight]:
"""Extract risk insights from transcript"""
system_prompt = """
You are a risk analyst specialized in identifying challenges, uncertainties, and risk factors
mentioned in earnings calls.
Focus on:
- Supply chain challenges
- Market uncertainties
- Competitive pressures
- Regulatory concerns
- Technical challenges
- Execution risks
- Macroeconomic factors
For each risk, identify its potential impact and any mentioned mitigation strategies.
Assign a severity score (1-5) based on how serious the risk appears from the transcript.
"""
print(f"Extracting risk insights for quarter {quarter}...")
response = self.client.chat.parse(
model=STRUCTURED_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"Extract risk insights from this earnings call transcript for {quarter}:\n\n{transcript}"}
],
response_format=RiskInsightsResponse,
temperature=0.1
)
print(f"Risk agent completed for {quarter}")
parsed_response = json.loads(response.choices[0].message.content)
return parsed_response['insights']
경쟁사 에이전트 (Competitor Agent)
class CompetitorAgent(Agent):
"""Agent for competitive analysis of earnings call transcripts"""
def analyze(self, transcript: str, quarter: str) -> List[CompetitorInsight]:
"""Extract competitor insights from transcript"""
system_prompt = """
You are a competitive intelligence analyst focused on the company's positioning relative to competitors.
Focus on:
- Direct mentions of competitors
- Market share discussions
- Competitive advantages or disadvantages
- Differentiation strategies
- Responses to competitive threats
- Emerging competition
Extract specific insights about the company's competitive positioning in different market segments.
Note who mentioned the competitive information (CEO, CFO, analyst, etc.)
"""
print(f"Extracting competitor insights for quarter {quarter}...")
response = self.client.chat.parse(
model=STRUCTURED_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"Extract competitive insights from this earnings call transcript for {quarter}:\n\n{transcript}"}
],
response_format=CompetitorInsightsResponse,
temperature=0.1
)
print(f"Competitor agent completed for {quarter}")
parsed_response = json.loads(response.choices[0].message.content)
return parsed_response['insights']
시간 흐름 분석 에이전트 (Temporal Analysis Agent)
class TemporalAnalysisAgent(Agent):
"""Agent for analyzing trends across quarters"""
def analyze(self, all_insights: Dict) -> List[TemporalInsight]:
"""Analyze trends and patterns across quarters"""
system_prompt = """
You are a trend analyst specialized in identifying patterns, changes, and developments
across multiple quarters of earnings calls.
Focus on:
- Growing or declining emphasis on specific topics
- Evolving business priorities
- Shifts in competitive positioning
- Changes in risk factors
- Sentiment trends
Identify meaningful patterns that show how the business is evolving over time.
Use specific evidence from multiple quarters to support each trend you identify.
"""
print("Running temporal analysis across quarters...")
# Format insights for analysis
formatted_insights = self._format_insights_for_analysis(all_insights)
response = self.client.chat.parse(
model=STRUCTURED_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"Analyze these insights across quarters to identify trends and patterns:\n\n{formatted_insights}"}
],
response_format=TemporalInsightsResponse,
temperature=0.2
)
print("Temporal analysis completed")
parsed_response = json.loads(response.choices[0].message.content)
return parsed_response['insights']
def _format_insights_for_analysis(self, all_insights: Dict) -> str:
"""Format all insights for temporal analysis"""
formatted = ""
for agent_type, quarters_data in all_insights.items():
formatted += f"\n## {agent_type.capitalize()} Insights Across Quarters:\n"
for quarter, insights in quarters_data.items():
formatted += f"\n### {quarter}:\n"
if isinstance(insights, list):
for insight in insights:
# Convert insight object to string representation
if isinstance(insight, dict):
insight_str = json.dumps(insight)
else:
insight_str = str(insight)
formatted += f"- {insight_str}\n"
else:
formatted += f"{insights}\n"
return formatted
쿼리 프로세서 (Query Processor)
Query Processor는 사용자 질문을 분석해서 필요한 특정 컴포넌트를 결정해요.
- NVIDIA 실적 발표에 대한 쿼리를 해석해요.
- 문제와 관련된 분기(Q1-Q4)를 식별해요.
- 쿼리 내용에 따라 활성화할 에이전트 유형을 결정해요.
- 분기 간 시간 흐름 분석이 필요한지 판단해요.
- 사용자 의도를 명확하게 해석해요.
이 컴포넌트는 워크플로우가 필요한 분석 경로만 활성화하도록 보장해서, 포괄적인 답변을 유지하면서 효율성을 높여요.
class QueryProcessor:
"""Processes user queries to determine workflow requirements"""
def __init__(self):
self.client = mistral_client
def analyze_query(self, query: str, company: str) -> QueryAnalysis:
"""Analyze user query to determine required components"""
system_prompt = f"""
You are a query analyzer for {company} earnings call transcripts.
Extract key information about which quarters and agent types are needed.
For agent_types, select from these options: Financial, Strategic, Sentiment, Risk, Competitor
For quarters, select from these options: Q1, Q2, Q3, Q4
Determine if temporal analysis is needed (comparing across quarters).
Provide a brief description of the user's intent.
"""
print(f"Analyzing query: {query}")
response = self.client.chat.parse(
model=STRUCTURED_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"Analyze this query about {company}: {query}"}
],
response_format=QueryAnalysis,
temperature=0
)
print("Query analysis completed")
parsed_response = json.loads(response.choices[0].message.content)
# Display query analysis results
print(f"Quarters needed: {parsed_response['quarters']}")
print(f"Agent types needed: {parsed_response['agent_types']}")
print(f"Temporal analysis required: {parsed_response['temporal_analysis_required']}")
print(f"Query intent: {parsed_response['query_intent']}")
return parsed_response
오케스트레이션 레이어 (Orchestration Layer)
EarningsCallAnalysisOrchestrator가 전체 분석 워크플로우를 핵심 함수들과 함께 조율해요.
process_transcript(): 모든 전문 에이전트로 분기별 대본을 분석해요.generate_comprehensive_report(): 선택한 분기들에 대한 상세 보고서를 만들어요.answer_query(): 특정 실적 발표 질문에 목표 지향적인 응답을 제공해요._generate_report_sections(): 구조화된 섹션(재무, 전략 등)을 생성해요._generate_query_response(): 관련 통찰에서 집중된 답변을 만들어요._compile_report(): 모든 섹션을 하나의 응집된 마크다운 문서로 조립해요.
이 오케스트레이션은 깊이 있는 분석 보고서와 정확한 쿼리 응답을 모두 제공하면서 효율적인 리소스 사용을 보장해요.
class EarningsCallAnalysisOrchestrator:
"""Agentic workflow orchestrator for earnings call analysis combining report generation and query capabilities"""
def __init__(self, company: str, year: str, mistral_client: Mistral):
self.company = company
self.year = year
self.insights_store = InsightsStore(company, year)
# Initialize agents
self.financial_agent = FinancialAgent()
self.strategic_agent = StrategicAgent()
self.sentiment_agent = SentimentAgent()
self.risk_agent = RiskAgent()
self.competitor_agent = CompetitorAgent()
self.temporal_agent = TemporalAnalysisAgent()
# Initialize query processor
self.query_processor = QueryProcessor()
# Initialize Mistral client
self.client = mistral_client
def process_transcript(self, quarter: str):
"""Process a transcript and store all insights"""
print(f"\n=== Processing {self.company} {self.year} {quarter} transcript ===\n")
try:
# Get transcript for this quarter
transcript = PDFParser.get_transcript_by_quarter(self.company, quarter, self.year, self.client)
# Run all agents on the transcript
financial_insights = self.financial_agent.analyze(transcript, quarter)
self.insights_store.add_insights("financial", quarter, financial_insights)
strategic_insights = self.strategic_agent.analyze(transcript, quarter)
self.insights_store.add_insights("strategic", quarter, strategic_insights)
sentiment_insights = self.sentiment_agent.analyze(transcript, quarter)
self.insights_store.add_insights("sentiment", quarter, sentiment_insights)
risk_insights = self.risk_agent.analyze(transcript, quarter)
self.insights_store.add_insights("risk", quarter, risk_insights)
competitor_insights = self.competitor_agent.analyze(transcript, quarter)
self.insights_store.add_insights("competitor", quarter, competitor_insights)
print(f"\n=== Completed processing {self.company} {self.year} {quarter} transcript ===\n")
return True
except Exception as e:
print(f"Error processing transcript for {quarter}: {str(e)}")
return False
def process_all_transcripts(self):
"""Process all quarterly transcripts for the year"""
all_success = True
for quarter in ["Q1", "Q2", "Q3", "Q4"]:
success = self.process_transcript(quarter)
all_success = all_success and success
return all_success
def generate_comprehensive_report(self, quarters=None):
"""Generate a comprehensive report for specified quarters or all quarters"""
if quarters is None:
quarters = ["Q1", "Q2", "Q3", "Q4"]
print(f"\n=== Generating comprehensive report for {self.company} {self.year} {', '.join(quarters)} ===\n")
# Ensure all needed transcripts are processed
for quarter in quarters:
if quarter not in self.insights_store.get_insights("financial"):
print(f"Processing missing transcript for {quarter}...")
self.process_transcript(quarter)
# Get all insights for the specified quarters
all_insights = {
"financial": self.insights_store.get_insights("financial", quarters),
"strategic": self.insights_store.get_insights("strategic", quarters),
"sentiment": self.insights_store.get_insights("sentiment", quarters),
"risk": self.insights_store.get_insights("risk", quarters),
"competitor": self.insights_store.get_insights("competitor", quarters)
}
# Run temporal analysis if multiple quarters
if len(quarters) > 1:
temporal_insights = self.temporal_agent.analyze(all_insights)
quarters_key = "_".join(sorted(quarters))
self.insights_store.add_insights("temporal", quarters_key, temporal_insights)
else:
temporal_insights = []
# Generate report sections
report_sections = self._generate_report_sections(quarters, all_insights, temporal_insights)
# Compile final report
report_content = self._compile_report(report_sections, quarters)
# Save report to file
output_file = f"{self.company}_{self.year}_{'_'.join(quarters)}_Analysis.md"
with open(output_file, "w", encoding="utf-8") as f:
f.write(report_content)
print(f"\n=== Report saved to {output_file} ===\n")
return output_file, report_content
def answer_query(self, query: str):
"""Answer a specific query about earnings calls"""
print(f"\n=== Processing query: {query} ===\n")
# Analyze the query to determine which quarters and agents to use
query_analysis = self.query_processor.analyze_query(query, self.company)
# Ensure we have the necessary insights
for quarter in query_analysis["quarters"]:
if quarter not in self.insights_store.get_insights("financial"):
print(f"Processing missing transcript for {quarter}...")
self.process_transcript(quarter)
# Collect relevant insights based on the query
relevant_insights = {}
for agent_type in query_analysis["agent_types"]:
agent_key = agent_type.lower()
relevant_insights[agent_key] = self.insights_store.get_insights(agent_key, query_analysis["quarters"])
# Get temporal insights if needed
temporal_insights = None
if query_analysis["temporal_analysis_required"] and len(query_analysis["quarters"]) > 1:
# Either use existing temporal insights or generate new ones
quarters_key = "_".join(sorted(query_analysis["quarters"]))
if quarters_key in self.insights_store.get_insights("temporal"):
temporal_insights = self.insights_store.get_insights("temporal")[quarters_key]
else:
temporal_insights = self.temporal_agent.analyze(relevant_insights)
self.insights_store.add_insights("temporal", quarters_key, temporal_insights)
# Generate response to the query
response = self._generate_query_response(query, query_analysis, relevant_insights, temporal_insights)
print("\n=== Query processing completed ===\n")
return response
def _generate_report_sections(self, quarters, all_insights, temporal_insights):
"""Generate all sections for the comprehensive report"""
print("Generating report sections...")
report_sections = {}
# Executive Summary
report_sections["executive_summary"] = self._generate_executive_summary(quarters, all_insights, temporal_insights)
# Financial Performance
report_sections["financial_performance"] = self._generate_financial_section(quarters, all_insights)
# Strategic Initiatives
report_sections["strategic_initiatives"] = self._generate_strategic_section(quarters, all_insights)
# Market Positioning
report_sections["market_positioning"] = self._generate_market_section(quarters, all_insights)
# Risk Assessment
report_sections["risk_assessment"] = self._generate_risk_section(quarters, all_insights)
# Quarterly Trends
if len(quarters) > 1:
report_sections["quarterly_trends"] = self._generate_trends_section(temporal_insights)
# Outlook and Projections
if "Q4" in quarters or len(quarters) > 2:
report_sections["outlook"] = self._generate_outlook_section(quarters, all_insights, temporal_insights)
print("Report sections generated")
return report_sections
def _generate_executive_summary(self, quarters, all_insights, temporal_insights):
"""Generate executive summary from all insights"""
system_prompt = f"""
You are a senior financial analyst creating an executive summary for a comprehensive report on {self.company}'s
performance across {', '.join(quarters)} of {self.year}.
Create a concise, high-level overview that captures:
1. Key financial performance highlights
2. Major strategic developments
3. Notable shifts in market positioning
4. Significant risks or challenges
5. Overall business trajectory
Keep the summary professional, balanced, and data-driven.
IMPORTANT FORMATTING INSTRUCTIONS:
- Use bullet points (not numbers) for lists with only one item
- Only use numbered lists when there are multiple items that need to be ordered
- Format subheadings as bold text using ** for emphasis
"""
# Format insights for the summary
formatted_insights = ""
for insight_type, quarters_data in all_insights.items():
formatted_insights += f"\n## {insight_type.capitalize()} Insights:\n"
for quarter, insights in quarters_data.items():
formatted_insights += f"\n### {quarter}:\n"
for insight in insights[:5]: # Limit to 5 insights per type per quarter
formatted_insights += f"- {json.dumps(insight)}\n"
# Add temporal insights if available
if temporal_insights:
formatted_insights += "\n## Temporal Trends:\n"
for insight in temporal_insights[:10]: # Limit to 10 temporal insights
if isinstance(insight, dict):
formatted_insights += f"- Trend: {insight.get('trend_type', 'N/A')}, Topic: {insight.get('topic', 'N/A')}\n"
formatted_insights += f" Description: {insight.get('description', 'N/A')}\n"
else:
formatted_insights += f"- {str(insight)}\n"
response = self.client.chat.complete(
model=DEFAULT_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"Generate an executive summary for {self.company}'s {self.year} performance based on these insights:\n\n{formatted_insights}"}
],
temperature=0.3
)
return {
"title": "Executive Summary",
"content": response.choices[0].message.content
}
def _generate_financial_section(self, quarters, all_insights):
"""Generate financial performance section"""
system_prompt = f"""
You are a financial analyst creating a detailed report section on {self.company}'s financial performance
across {', '.join(quarters)} of {self.year}.
Create a comprehensive analysis that includes:
1. Quarter-by-quarter revenue analysis (overall and by segment)
2. Profitability metrics and trends
3. Cash flow and balance sheet highlights
4. Key performance indicators and their trajectories
5. Comparison of actual results vs. guidance
Use subsections with clear headings, and include specific figures whenever available.
IMPORTANT FORMATTING INSTRUCTIONS:
- Use bullet points (not numbers) for lists with only one item
- Only use numbered lists when there are multiple items that need to be ordered
- Format subheadings as bold text using ** for emphasis
"""
# Format financial insights for all quarters
financial_insights = ""
for quarter, insights in all_insights["financial"].items():
financial_insights += f"\n## {quarter} Financial Insights:\n"
for insight in insights:
if isinstance(insight, dict):
financial_insights += f"- Metric: {insight.get('metric_name', 'N/A')}, Value: {insight.get('value', 'N/A')}\n"
financial_insights += f" Context: {insight.get('context', 'N/A')}\n"
else:
financial_insights += f"- {str(insight)}\n"
response = self.client.chat.complete(
model=DEFAULT_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"Generate a comprehensive financial performance section for {self.company}'s {self.year} based on these insights:\n\n{financial_insights}"}
],
temperature=0.3
)
return {
"title": "Financial Performance Analysis",
"content": response.choices[0].message.content
}
def _generate_strategic_section(self, quarters, all_insights):
"""Generate strategic initiatives section"""
system_prompt = f"""
You are a business strategy analyst creating a detailed report section on {self.company}'s strategic initiatives
across {', '.join(quarters)} of {self.year}.
Create a comprehensive analysis that includes:
1. Key strategic priorities and how they evolved
2. Product roadmap developments
3. Major partnerships and acquisitions
4. R&D focus areas and investments
5. Market expansion efforts
Organize by major strategic themes, highlighting changes in emphasis over time.
IMPORTANT FORMATTING INSTRUCTIONS:
- Use bullet points (not numbers) for lists with only one item
- Only use numbered lists when there are multiple items that need to be ordered
- Format subheadings as bold text using ** for emphasis
"""
# Format strategic insights for all quarters
strategic_insights = ""
for quarter, insights in all_insights["strategic"].items():
strategic_insights += f"\n## {quarter} Strategic Insights:\n"
for insight in insights:
if isinstance(insight, dict):
strategic_insights += f"- Initiative: {insight.get('initiative', 'N/A')}, Importance: {insight.get('importance', 'N/A')}/5\n"
strategic_insights += f" Description: {insight.get('description', 'N/A')}\n"
if insight.get('timeframe'):
strategic_insights += f" Timeframe: {insight.get('timeframe')}\n"
else:
strategic_insights += f"- {str(insight)}\n"
response = self.client.chat.complete(
model=DEFAULT_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"Generate a comprehensive strategic initiatives section for {self.company}'s {self.year} based on these insights:\n\n{strategic_insights}"}
],
temperature=0.3
)
return {
"title": "Strategic Initiatives Analysis",
"content": response.choices[0].message.content
}
def _generate_market_section(self, quarters, all_insights):
"""Generate market positioning section"""
system_prompt = f"""
You are a market analyst creating a detailed report section on {self.company}'s competitive positioning
across {', '.join(quarters)} of {self.year}.
Create a comprehensive analysis that includes:
1. {self.company}'s position in key market segments
2. Competitive dynamics with major rivals
3. Market share developments
4. Differentiation strategies
5. Emerging competition and responses
Organize by major market segments, analyzing competitive position in each.
IMPORTANT FORMATTING INSTRUCTIONS:
- Use bullet points (not numbers) for lists with only one item
- Only use numbered lists when there are multiple items that need to be ordered
- Format subheadings as bold text using ** for emphasis
"""
# Format competitor insights for all quarters
competitor_insights = ""
for quarter, insights in all_insights["competitor"].items():
competitor_insights += f"\n## {quarter} Competitive Insights:\n"
for insight in insights:
if isinstance(insight, dict):
competitor_insights += f"- Market Segment: {insight.get('market_segment', 'N/A')}\n"
if insight.get('competitor'):
competitor_insights += f" Competitor: {insight.get('competitor')}\n"
competitor_insights += f" Positioning: {insight.get('positioning', 'N/A')}\n"
competitor_insights += f" Mentioned by: {insight.get('mentioned_by', 'N/A')}\n"
else:
competitor_insights += f"- {str(insight)}\n"
# Also include sentiment insights as they relate to market positioning
for quarter, insights in all_insights["sentiment"].items():
competitor_insights += f"\n## {quarter} Sentiment Insights (Market Related):\n"
for insight in insights:
if isinstance(insight, dict) and any(market_term in insight.get('topic', '').lower() for market_term in
['market', 'competitor', 'competition', 'position', 'share']):
competitor_insights += f"- Topic: {insight.get('topic', 'N/A')}, Sentiment: {insight.get('sentiment', 'N/A')}\n"
competitor_insights += f" Speaker: {insight.get('speaker', 'N/A')}\n"
competitor_insights += f" Evidence: {insight.get('evidence', 'N/A')}\n"
response = self.client.chat.complete(
model=DEFAULT_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"Generate a comprehensive market positioning section for {self.company}'s {self.year} based on these insights:\n\n{competitor_insights}"}
],
temperature=0.3
)
return {
"title": "Market Positioning Analysis",
"content": response.choices[0].message.content
}
def _generate_risk_section(self, quarters, all_insights):
"""Generate risk assessment section"""
system_prompt = f"""
You are a risk analyst creating a detailed report section on {self.company}'s risk factors and challenges
across {', '.join(quarters)} of {self.year}.
Create a comprehensive analysis that includes:
1. Major risk categories (supply chain, competition, regulatory, etc.)
2. Evolution of key risks throughout the year
3. Mitigation strategies mentioned by management
4. Emerging vs. declining risk factors
5. Assessment of risk management effectiveness
Organize by risk categories, with severity assessments and trends over time.
IMPORTANT FORMATTING INSTRUCTIONS:
- Use bullet points (not numbers) for lists with only one item
- Only use numbered lists when there are multiple items that need to be ordered
- Format subheadings as bold text using ** for emphasis
"""
# Format risk insights for all quarters
risk_insights = ""
for quarter, insights in all_insights["risk"].items():
risk_insights += f"\n## {quarter} Risk Insights:\n"
for insight in insights:
if isinstance(insight, dict):
risk_insights += f"- Risk Factor: {insight.get('risk_factor', 'N/A')}, Severity: {insight.get('severity', 'N/A')}/5\n"
risk_insights += f" Description: {insight.get('description', 'N/A')}\n"
risk_insights += f" Potential Impact: {insight.get('potential_impact', 'N/A')}\n"
if insight.get('mitigation_mentioned'):
risk_insights += f" Mitigation: {insight.get('mitigation_mentioned')}\n"
else:
risk_insights += f"- {str(insight)}\n"
response = self.client.chat.complete(
model=DEFAULT_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"Generate a comprehensive risk assessment section for {self.company}'s {self.year} based on these insights:\n\n{risk_insights}"}
],
temperature=0.3
)
return {
"title": "Risk Assessment",
"content": response.choices[0].message.content
}
def _generate_trends_section(self, temporal_insights):
"""Generate quarterly trends section"""
system_prompt = f"""
You are a business analyst creating a detailed report section on {self.company}'s quarter-to-quarter trends
across multiple dimensions.
Create a comprehensive analysis that includes:
1. Major trends across all analysis dimensions (financial, strategic, etc.)
2. Inflection points or significant shifts during the year
3. Business cycle position and momentum
4. Management focus evolution
5. Market reception changes
Highlight the most significant developments and their implications.
IMPORTANT FORMATTING INSTRUCTIONS:
- Use bullet points (not numbers) for lists with only one item
- Only use numbered lists when there are multiple items that need to be ordered
- Format subheadings as bold text using ** for emphasis
"""
# Format temporal insights
formatted_temporal_insights = ""
for insight in temporal_insights:
if isinstance(insight, dict):
formatted_temporal_insights += f"- Trend: {insight.get('trend_type', 'N/A')}, Topic: {insight.get('topic', 'N/A')}\n"
formatted_temporal_insights += f" Description: {insight.get('description', 'N/A')}\n"
formatted_temporal_insights += f" Quarters: {', '.join(insight.get('quarters_observed', ['N/A']))}\n"
formatted_temporal_insights += f" Evidence: {insight.get('supporting_evidence', 'N/A')}\n\n"
else:
formatted_temporal_insights += f"- {str(insight)}\n"
response = self.client.chat.complete(
model=DEFAULT_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"Generate a comprehensive quarterly trends section for {self.company}'s {self.year} based on these insights:\n\n{formatted_temporal_insights}"}
],
temperature=0.3
)
return {
"title": "Quarterly Trends Analysis",
"content": response.choices[0].message.content
}
def _generate_outlook_section(self, quarters, all_insights, temporal_insights):
"""Generate outlook and projections section"""
system_prompt = f"""
You are a forward-looking analyst creating a detailed outlook section for {self.company}
based on earnings call insights across multiple quarters.
Create a comprehensive outlook that includes:
1. Forward guidance from management
2. Key initiatives to watch in the coming year
3. Potential challenges and opportunities
4. Market segment outlooks
5. Long-term strategic trajectory
Focus particularly on the most recent quarter and guidance, but incorporate the full context.
IMPORTANT FORMATTING INSTRUCTIONS:
- Use bullet points (not numbers) for lists with only one item
- Only use numbered lists when there are multiple items that need to be ordered
- Format subheadings as bold text using ** for emphasis
"""
# Get the latest quarter's insights
latest_quarter = sorted(quarters)[-1]
# Financial insights from latest quarter
latest_insights = f"\n## {latest_quarter} Financial Insights:\n"
if latest_quarter in all_insights["financial"]:
for insight in all_insights["financial"][latest_quarter]:
if isinstance(insight, dict):
latest_insights += f"- Metric: {insight.get('metric_name', 'N/A')}, Value: {insight.get('value', 'N/A')}\n"
latest_insights += f" Context: {insight.get('context', 'N/A')}\n"
else:
latest_insights += f"- {str(insight)}\n"
# Strategic insights from latest quarter
latest_insights += f"\n## {latest_quarter} Strategic Insights:\n"
if latest_quarter in all_insights["strategic"]:
for insight in all_insights["strategic"][latest_quarter]:
if isinstance(insight, dict):
latest_insights += f"- Initiative: {insight.get('initiative', 'N/A')}, Importance: {insight.get('importance', 'N/A')}/5\n"
latest_insights += f" Description: {insight.get('description', 'N/A')}\n"
else:
latest_insights += f"- {str(insight)}\n"
# Add sentiment insights about future outlook
latest_insights += f"\n## {latest_quarter} Sentiment on Future Outlook:\n"
if latest_quarter in all_insights["sentiment"]:
for insight in all_insights["sentiment"][latest_quarter]:
if isinstance(insight, dict) and any(future_term in insight.get('topic', '').lower() for future_term in
['outlook', 'future', 'guidance', 'next quarter', 'next year', 'projection']):
latest_insights += f"- Topic: {insight.get('topic', 'N/A')}, Sentiment: {insight.get('sentiment', 'N/A')}\n"
latest_insights += f" Speaker: {insight.get('speaker', 'N/A')}\n"
latest_insights += f" Evidence: {insight.get('evidence', 'N/A')}\n"
# Add temporal insights if available
if temporal_insights:
latest_insights += "\n## Overall Trends:\n"
for insight in temporal_insights:
if isinstance(insight, dict):
latest_insights += f"- Trend: {insight.get('trend_type', 'N/A')}, Topic: {insight.get('topic', 'N/A')}\n"
latest_insights += f" Description: {insight.get('description', 'N/A')}\n"
else:
latest_insights += f"- {str(insight)}\n"
response = self.client.chat.complete(
model=DEFAULT_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"Generate a comprehensive outlook and projections section for {self.company} based on their {self.year} earnings calls:\n\n{latest_insights}"}
],
temperature=0.3
)
return {
"title": "Outlook and Projections",
"content": response.choices[0].message.content
}
def _compile_report(self, report_sections, quarters):
"""Compile all sections into a final comprehensive report"""
print("Compiling final comprehensive report...")
# Assemble full report content
report_content = f"# {self.company} {self.year} Earnings Call Analysis\n\n"
if len(quarters) == 4:
report_content += f"## Annual Comprehensive Analysis Report\n\n"
else:
report_content += f"## Analysis Report for {', '.join(quarters)}\n\n"
# Add date generated
report_content += f"*Generated on: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}*\n\n"
# Add executive summary
report_content += f"# {report_sections['executive_summary']['title']}\n\n"
report_content += f"{report_sections['executive_summary']['content']}\n\n"
# Add financial performance
report_content += f"# {report_sections['financial_performance']['title']}\n\n"
report_content += f"{report_sections['financial_performance']['content']}\n\n"
# Add strategic initiatives
report_content += f"# {report_sections['strategic_initiatives']['title']}\n\n"
report_content += f"{report_sections['strategic_initiatives']['content']}\n\n"
# Add market positioning
report_content += f"# {report_sections['market_positioning']['title']}\n\n"
report_content += f"{report_sections['market_positioning']['content']}\n\n"
# Add risk assessment
report_content += f"# {report_sections['risk_assessment']['title']}\n\n"
report_content += f"{report_sections['risk_assessment']['content']}\n\n"
# Add quarterly trends if available
if 'quarterly_trends' in report_sections:
report_content += f"# {report_sections['quarterly_trends']['title']}\n\n"
report_content += f"{report_sections['quarterly_trends']['content']}\n\n"
# Add outlook if available
if 'outlook' in report_sections:
report_content += f"# {report_sections['outlook']['title']}\n\n"
report_content += f"{report_sections['outlook']['content']}\n\n"
return report_content
def _generate_query_response(self, query, query_analysis, relevant_insights, temporal_insights):
"""Generate response to a specific query"""
system_prompt = f"""
You are an expert analyst of {self.company} earnings calls.
Provide a clear, concise response to the user's query based on the insights provided.
Focus only on answering what was asked, using the most relevant insights.
Include specific data points and evidence from the earnings calls.
"""
# Format insights for prompt
insights_formatted = ""
for insight_type, quarters_data in relevant_insights.items():
insights_formatted += f"\n## {insight_type.capitalize()} Insights:\n"
for quarter, insights in quarters_data.items():
insights_formatted += f"\n### {quarter}:\n"
for insight in insights:
if isinstance(insight, dict):
insight_formatted = json.dumps(insight)
else:
insight_formatted = str(insight)
insights_formatted += f"- {insight_formatted}\n"
# Add temporal insights if available
temporal_formatted = ""
if temporal_insights:
temporal_formatted += "\n## Temporal Trends:\n"
for insight in temporal_insights:
if isinstance(insight, dict):
temporal_formatted += f"- Trend: {insight.get('trend_type', 'N/A')}, Topic: {insight.get('topic', 'N/A')}\n"
temporal_formatted += f" Description: {insight.get('description', 'N/A')}\n"
temporal_formatted += f" Quarters: {', '.join(insight.get('quarters_observed', ['N/A']))}\n"
temporal_formatted += f" Evidence: {insight.get('supporting_evidence', 'N/A')}\n"
else:
temporal_formatted += f"- {str(insight)}\n"
user_prompt = f"""
Query: {query}
Quarters analyzed: {', '.join(query_analysis['quarters'])}
Agent types used: {', '.join(query_analysis['agent_types'])}
Insights collected:
{insights_formatted}
"""
if temporal_formatted:
user_prompt += f"""
Temporal insights:
{temporal_formatted}
"""
response = self.client.chat.complete(
model=DEFAULT_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
temperature=0.3
)
return response.choices[0].message.content
NVIDIA 2025 실적 발표를 위한 시스템 초기화 (Initialize the system for NVIDIA 2025 earnings calls)
company = "NVIDIA"
year = "2025"
orchestrator = EarningsCallAnalysisOrchestrator(company, year, mistral_client)
모든 분기별 대본 처리 (Process All Quarterly Transcripts)
모든 분기별 대본을 한 번에 처리해서 다양한 통찰을 생성해요. 이렇게 하면 보고서 생성과 쿼리 응답 모두가 더 효율적이 돼요.
Python
print("Processing all quarterly transcripts...")
quarters = ["Q1", "Q2", "Q3", "Q4"]
for quarter in quarters:
success = orchestrator.process_transcript(quarter)
if success:
print(f"✓ Successfully processed {quarter} transcript")
else:
print(f"✗ Failed to process {quarter} transcript")
보고서 생성 (Report Generation)
분기별 통찰을 경영 요약, 재무 분석, 전략 이니셔티브, 시장 포지셔닝, 위험 평가, 미래 전망 등의 구조화된 섹션으로 정리해서 포괄적인 보고서를 생성해요.
Python
print("\nGenerating comprehensive annual report...")
report_file, report_content = orchestrator.generate_comprehensive_report(quarters)
Python
print(f"\nDisplaying report saved to: {report_file}")
display(Markdown(report_content))
쿼리 응답 (Query Answering)
우리 시스템은 사용자 질문을 분석해서 관련 분기, 에이전트 유형, 분석 차원을 결정한 뒤, 가장 적절한 통찰만 사용해서 목표 지향적인 응답을 제공해요.
Query-1
- Query - What were NVIDIA's key financial metrics in Q1 and Q2 2025?
- Agents Used - Financial Agent
- Quarters - Q1, Q2
- Temporal Analysis Required - False
- Financial Year - 2025
Python
query = "What were the key financial metrics in Q1 and Q2?"
answer = orchestrator.answer_query(query)
Python
display(Markdown(answer))
Query-2
- Query - Identify strategic shifts in NVIDIA's automotive business across 2025
- Agents Used - Strategic Agent
- Quarters - Q1, Q2, Q3, Q4
- Temporal Analysis Required - True
- Financial Year - 2025
Python
query = "Identify strategic shifts in NVIDIA's automotive business across 2025"
answer = orchestrator.answer_query(query)
Python
display(Markdown(answer))
Query-3
- Query - What risks did NVIDIA highlight in their Q4 earnings call, and how do they compare to those mentioned in Q3?
- Agents Used - Risk Agent
- Quarters - Q3, Q4
- Temporal Analysis Required - True
- Financial Year - 2025
Python
query = "What risks did NVIDIA highlight in their Q4 earnings call, and how do they compare to those mentioned in Q3?"
answer = orchestrator.answer_query(query)
Python
display(Markdown(answer))