Compare companies’ 10-K filings¶
An analyst wants to know which of several companies carries the most concentrated risk and which is growing fastest, grounded in the filings rather than the model’s memory. A weekly Dag fetches each latest 10-K from SEC EDGAR and indexes it. An on-demand Dag has the model split the question per company, retrieves against each index in its own task so a missing index retries alone, and writes a structured report. A human edits the question on the way in and approves the report on the way out.
Two variants build the same graph: LlamaIndex operators, or LangChain with FAISS.
What this demonstrates¶
Embed documents: LlamaIndexEmbeddingOperator and Retrieve context: LlamaIndexRetrievalOperator – index per ticker on a schedule, retrieve per sub-question on demand.
Single prompts: LLMOperator and @task.llm –
@task.llmwith structured output for decomposition, andLLMOperatorbounded byUsageLimitsfor synthesis.Dynamic Task Mapping – the model decides how many sub-questions there are; the Dag maps over them at runtime.
Human in the Loop (HITL) Operators –
HITLEntryOperatoron the way in,ApprovalOperatoron the way out.LangChain models: LangChainHook – the LangChain variant does indexing and retrieval in plain
@taskfunctions overLangChainHookand FAISS.
Run it¶
Install the provider with the LlamaIndex extra (or
langchainfor the other variant):pip install "apache-airflow-providers-common-ai[openai,llamaindex]"
Create a
llamaindexconnection namedllamaindex_defaultfor embedding and retrieval (see LlamaIndex connection);pydanticai_defaultdoes decomposition and synthesis.Set
EDGAR_USER_AGENTin the file to your name and email. SEC requires a contact address on every EDGAR request. No API key is needed.Run the indexing Dag once, then trigger the analysis:
airflow dags test example_llamaindex_10k_index airflow dags test example_llamaindex_10k_analysis
The run pauses at analyst_input to confirm the question and tickers, and at
review_report. Answer both from Required Actions in the UI (see the note on
What you can build). The synthesize_report XCom holds the AnalysisReport.
The Dags share only the index path INDEX_BASE_DIR/<lowercased ticker>, so index a
ticker before analyzing it.
The indexing Dag¶
One mapped LlamaIndexEmbeddingOperator per ticker, weekly:
@dag(
schedule="@weekly",
catchup=False,
params={
"tickers": Param(
DEFAULT_TICKERS,
type="string",
description="Comma-separated stock tickers to fetch and index (e.g. AAPL,MSFT,UBER)",
),
},
tags=["example", "llamaindex", "10k"],
)
def example_llamaindex_10k_index():
"""
Fetch 10-K filings from SEC EDGAR and build per-company vector indexes.
Runs weekly to refresh indexes when new filings arrive. Each company
gets its own persisted index via Dynamic Task Mapping.
Task graph::
fetch_filings (@task, live from SEC EDGAR)
-> build_index (LlamaIndexEmbeddingOperator x N companies)
"""
# [START 10k_index_dtm]
@task
def fetch_filings(params: dict) -> list[dict]:
tickers = [t.strip().upper() for t in params["tickers"].split(",") if t.strip()]
result = []
for ticker in tickers:
cik, company_name = _resolve_ticker(ticker)
doc_url, filing_date = _find_latest_10k(cik)
html = _edgar_get_text(doc_url)
plain_text = _strip_html_tags(html)
documents = _extract_filing_sections(plain_text, ticker, company_name, filing_date)
result.append(
{
"documents": documents,
"persist_dir": f"{INDEX_BASE_DIR}/{ticker.lower()}",
}
)
return result
LlamaIndexEmbeddingOperator.partial(
task_id="build_index",
embed_model="text-embedding-3-small",
llm_conn_id=LLAMAINDEX_CONN_ID,
chunk_size=512,
chunk_overlap=50,
).expand_kwargs(fetch_filings())
# [END 10k_index_dtm]
The analysis Dag¶
The output types come first. DecomposedQuestion is what the model returns from the
decomposition step, and AnalysisReport is what the reviewer approves:
class SubQuestion(BaseModel):
"""One sub-question targeting a specific company."""
sub_question: str
ticker: str
class DecomposedQuestion(BaseModel):
"""LLM-produced decomposition of the analyst's question."""
sub_questions: list[SubQuestion]
class AnalysisReport(BaseModel):
"""Structured financial comparison report."""
executive_summary: str
company_findings: list[dict]
key_risks: list[str]
recommendations: list[str]
The Dag itself:
@dag(
schedule=None,
catchup=False,
params={
"tickers": Param(
DEFAULT_TICKERS,
type="string",
description="Comma-separated stock tickers (must match indexed companies)",
),
},
tags=["example", "llamaindex", "10k"],
)
def example_llamaindex_10k_analysis():
"""
Multi-company financial comparison via LLM-driven sub-question decomposition.
An analyst submits a comparison question. The LLM decomposes it into
company-specific sub-questions (N decided at runtime), each sub-question
retrieves from the appropriate company's vector index in parallel via
Dynamic Task Mapping, and the results are synthesized into a structured
report for human review.
Task graph::
analyst_input (HITLEntryOperator, tickers + question)
-> get_question (@task)
-> get_tickers (@task)
-> decompose_question (@task.llm, structured output)
-> extract_sub_questions (@task)
-> build_retrieval_kwargs (@task)
-> retrieve (LlamaIndexRetrievalOperator x N, DTM)
-> collect_results (@task)
-> synthesize_report (LLMOperator, UsageLimits + AnalysisReport)
-> format_report (@task, readable text for reviewer)
-> review_report (ApprovalOperator)
"""
# ------------------------------------------------------------------
# Step 1: Analyst submits the comparison question via HITL.
# ------------------------------------------------------------------
# [START 10k_hitl_entry]
analyst_input = HITLEntryOperator(
task_id="analyst_input",
subject="Enter a 10-K comparison question",
params={
"tickers": Param(
DEFAULT_TICKERS,
type="string",
description="Comma-separated stock tickers to compare (must match indexed companies)",
),
"question": Param(
DEFAULT_QUESTION,
type="string",
description="Financial comparison question across the selected companies",
),
},
response_timeout=timedelta(hours=1),
)
# [END 10k_hitl_entry]
@task
def get_question(hitl_response: dict) -> str:
return hitl_response["params_input"]["question"]
question = get_question(analyst_input.output)
@task
def get_tickers(hitl_response: dict) -> str:
return hitl_response["params_input"]["tickers"]
tickers = get_tickers(analyst_input.output)
# ------------------------------------------------------------------
# Step 2: LLM decomposes the question into company-specific
# sub-questions. N is decided by the LLM at runtime -- this is the
# dynamic adaptation that a static Dag cannot express.
# ------------------------------------------------------------------
# [START 10k_decompose]
@task.llm(
llm_conn_id=LLM_CONN_ID,
system_prompt=DECOMPOSE_SYSTEM_PROMPT,
output_type=DecomposedQuestion,
# Push the structured output to XCom as a dict so the example runs on
# every supported Airflow version (the model-instance form needs 3.3+).
serialize_output=True,
)
def decompose_question(question: str, tickers: str) -> str:
return (
f"Decompose this question into company-specific sub-questions.\n"
f"Available companies (by ticker): {tickers}\n\n"
f"Question: {question}"
)
decomposed = decompose_question(question, tickers)
# [END 10k_decompose]
@task
def extract_sub_questions(decomposed: dict) -> list[dict]:
return decomposed["sub_questions"]
sub_questions = extract_sub_questions(decomposed)
# ------------------------------------------------------------------
# Step 3: Map sub-questions to LlamaIndexRetrievalOperator kwargs.
# Each sub-question targets a specific company's pre-built index.
# ------------------------------------------------------------------
@task
def build_retrieval_kwargs(sub_questions: list[dict]) -> list[dict]:
return [
{
"query": sq["sub_question"],
"index_persist_dir": f"{INDEX_BASE_DIR}/{sq['ticker'].lower()}",
}
for sq in sub_questions
]
retrieval_kwargs = build_retrieval_kwargs(sub_questions)
# ------------------------------------------------------------------
# Step 4: Retrieve relevant chunks for each sub-question.
# Dynamic Task Mapping fans out one LlamaIndexRetrievalOperator
# per sub-question, each targeting the company's vector index.
# ------------------------------------------------------------------
# [START 10k_dtm_retrieval]
retrieval_results = LlamaIndexRetrievalOperator.partial(
task_id="retrieve",
embed_model="text-embedding-3-small",
llm_conn_id=LLAMAINDEX_CONN_ID,
top_k=5,
).expand_kwargs(retrieval_kwargs)
# [END 10k_dtm_retrieval]
# ------------------------------------------------------------------
# Step 5: Collect all retrieval results into a single context.
# Mapped outputs preserve input order, so zip with sub_questions
# re-associates each result with its company.
# ------------------------------------------------------------------
@task
def collect_results(sub_questions: list[dict], results: list[dict]) -> str:
sections = []
for sq, r in zip(sub_questions, results):
chunks_text = "\n".join(
f" [{i + 1}] (score {c.get('score') or 0.0:.2f}) {c['text']}"
for i, c in enumerate(r["chunks"])
)
sections.append(f"## {sq['ticker']} -- {sq['sub_question']}\n{chunks_text}")
return "\n\n".join(sections)
collected = collect_results(sub_questions, retrieval_results.output)
# ------------------------------------------------------------------
# Step 6: Synthesize a structured comparison report.
# UsageLimits caps the token spend; output_type=AnalysisReport
# enforces the Pydantic schema on the LLM response.
# ------------------------------------------------------------------
# [START 10k_synthesis]
synthesize = LLMOperator(
task_id="synthesize_report",
llm_conn_id=LLM_CONN_ID,
system_prompt=SYNTHESIS_SYSTEM_PROMPT,
prompt="""\
Synthesize a cross-company financial comparison from these retrieval results.
Cite specific data points and scores.
{{ ti.xcom_pull(task_ids='collect_results') }}""",
output_type=AnalysisReport,
serialize_output=True,
usage_limits=UsageLimits(
request_limit=10,
input_tokens_limit=50_000,
output_tokens_limit=16_000,
),
)
# [END 10k_synthesis]
collected >> synthesize
# ------------------------------------------------------------------
# Step 7: Format the structured report into readable text for the
# human reviewer. The LLM produced a dict (via output_type=
# AnalysisReport with serialize_output); this task renders it as clean prose.
# ------------------------------------------------------------------
@task
def format_report(report: dict) -> str:
lines = [f"# Executive Summary\n\n{report['executive_summary']}"]
if report.get("company_findings"):
lines.append("\n# Company Findings")
for finding in report["company_findings"]:
company = finding.get("company") or finding.get("ticker", "Unknown")
lines.append(f"\n## {company}")
for key, value in finding.items():
if key not in ("company", "ticker"):
lines.append(f"- **{key}**: {value}")
if report.get("key_risks"):
lines.append("\n# Key Risks")
for risk in report["key_risks"]:
lines.append(f"- {risk}")
if report.get("recommendations"):
lines.append("\n# Recommendations")
for rec in report["recommendations"]:
lines.append(f"- {rec}")
return "\n".join(lines)
review_body = format_report(synthesize.output)
# ------------------------------------------------------------------
# Step 8: Analyst reviews the report before it reaches the
# investment committee.
# ------------------------------------------------------------------
# [START 10k_hitl_approval]
ApprovalOperator(
task_id="review_report",
subject="Review 10-K comparison report before sharing",
body=review_body,
response_timeout=timedelta(hours=24),
)
# [END 10k_hitl_approval]
Decomposition is the step to notice: the model returns a list and the Dag maps retrieval over it, so the model decides at runtime how many tasks run, each with its own log and retries.
Adapting it¶
Change
DEFAULT_TICKERSor passtickersas a Dag param. Any US-listed company works; EDGAR resolves the ticker.Replace
fetch_filingswith your own document source; nothing downstream cares where the text came from.Tighten
UsageLimitsonsynthesize_reportto cap spend per run.The LangChain build is
example_langchain_10k.py(Dag idsexample_langchain_10k_indexandexample_langchain_10k_analysis); it adds alangchain_defaultconnection for embeddings.