finpy1789's picture
Agentic Financial Document Analyst: multi-agent RAG + MCP, agent-coloured UI
aa4269d verified
Raw
History Blame Contribute Delete
3.13 kB
"""Structured professional report generation (Advanced Analysis output).
Assembles the one-click report: executive summary, financial analysis,
ratio analysis, risk assessment, audit observations, potential
inconsistencies, recommendations, limitations.
"""
from __future__ import annotations
import json
from datetime import date
from langchain_core.messages import HumanMessage, SystemMessage
from src.analysis.risk import RiskAssessment
from src.llm import get_llm
SYSTEM = """You are drafting a professional financial analysis report for a reviewer.
Write in a measured, audit-adjacent register. Ground every claim in the material
provided and keep citations in square brackets exactly as given. Where the inputs
flag potential inconsistencies, present them neutrally as items for review with
their confidence levels — never as established errors. Output clean Markdown with
exactly these section headings (## level):
Executive Summary, Financial Analysis, Ratio Analysis, Risk Assessment,
Audit Observations, Potential Inconsistencies, Recommendations, Limitations."""
def generate(question: str,
analysis_text: str,
verification_text: str,
findings: list[dict],
risk: RiskAssessment,
ratio_results: list[dict] | None = None,
doc_ids: list[str] | None = None) -> str:
llm = get_llm("analyst")
payload = {
"analysis_scope": question,
"documents_reviewed": doc_ids or [],
"analyst_findings": analysis_text,
"verification_summary": verification_text,
"evidence_matrix": findings,
"ratio_results": ratio_results or [],
"risk_items": risk.heatmap_rows(),
"reliability_score": risk.reliability_score,
}
resp = llm.invoke([
SystemMessage(content=SYSTEM),
HumanMessage(content=json.dumps(payload, indent=2, default=str)),
])
header = (
f"# Financial Document Analysis Report\n\n"
f"**Date:** {date.today().isoformat()} \n"
f"**Documents:** {', '.join(doc_ids or []) or 'n/a'} \n"
f"**Document-set reliability score:** {risk.reliability_score}/100\n\n---\n\n"
)
return header + resp.content
def evidence_matrix_markdown(findings: list[dict]) -> str:
"""Render the evidence matrix as a Markdown table for the UI/report."""
if not findings:
return "_No cross-checkable claims found._"
lines = ["| Claim | Topic | Status | Confidence | Sources | Note |",
"|---|---|---|---|---|---|"]
for f in findings:
sources = "<br>".join(
f"{s.get('source', '?')} p.{s.get('page', '?')}: \"{s.get('statement', '')[:80]}\""
for s in f.get("sources", [])
)
status = {"agree": "✅ agree", "differ": "⚠️ differ",
"single_source": "ℹ️ single source"}.get(f.get("status"), f.get("status", "?"))
lines.append(
f"| {f.get('claim', '')} | {f.get('topic', '')} | {status} "
f"| {f.get('confidence', '')}% | {sources} | {f.get('note', '')} |"
)
return "\n".join(lines)