企业年报 AI 解读：让模型看财报，从表格到洞察的工程实践

企业年报 AI 解读：让模型看财报，从表格到洞察的工程实践

一、个性化深度引言

"让大模型看财报"听起来不复杂——把 PDF 丢进去，"告诉我这家公司财务健康吗？"但这行不通。

第一个问题在表格理解上。一份标准年报有30+张表格，其中有合并报表、母公数据、分部业务数据。大模型擅长看文本，但看表格容易把合并口径的数据和母公司口径的数据混为一谈——而这两个口径差了一个量级。

第二个问题在数字串联上。营业收入的增长需要和应收账款、经营现金流、存货周转天数串联起来看。单看一个指标无法形成判断，需要多表联动分析。

第三个问题在幻觉上。让模型直接生成"这家公司财务状况良好"——证据在哪？如果模型编造成长率，财务报表分析就失去了可信度。

这篇文章复盘了一套"让模型看报表而不产生幻觉"的工程实践——核心策略是强制模型使用提取的表格数据而非训练数据，实现"引用即证据"。

二、个性化原理剖析

财报 AI 解读的核心架构：

核心设计思路是"计算归计算，推理归推理"。指标计算（同比增长率、毛利率、资产负债率）必须由代码精确计算，不允许 LLM 介入——因为 LLM 会对数字做近似处理。LLM 只在最后一步负责"从指标到洞察"的语义分析，且必须引用数据来源。

三、个性化代码实践

财报提取和分析的核心实现：

import re
from dataclasses import dataclass, field
from typing import List, Dict, Tuple, Optional
from enum import Enum
from datetime import datetime
import numpy as np
class ReportPeriod(Enum):
    """报告期间——设计原因：统一口径防止年度/季度数据混淆"""
    ANNUAL = "annual"
    Q1 = "q1"
    H1 = "h1"
    Q3 = "q3"
@dataclass
class FinancialMetric:
    """财务指标——设计原因：统一数据结构，所有指标同一格式"""
    name: str
    value: float  # 指标值（需统一单位，如元）
    period: ReportPeriod
    year: int
    caliber: str  # 合并/母公司
    source_table: str  # 来源于哪个表
    source_cell: str   # 来源于哪个单元格
    def format_display(self) -> str:
        """格式化显示——设计原因：展示用，原始值保留精度"""
        if abs(self.value) >= 1e8:
            return f"{self.value/1e8:.2f}亿"
        elif abs(self.value) >= 1e4:
            return f"{self.value/1e4:.2f}万"
        else:
            return f"{self.value:.2f}"
@dataclass  
class FinancialAnalysis:
    """财务分析报告——设计原因：结构化输出，每条洞察有来源"""
    company_name: str
    report_period: str
    metrics: List[FinancialMetric]
    insights: List[Dict]  # 每条洞察: {text, confidence, source_metrics}
class FinancialTableExtractor:
    """财务报表表格提取器——设计原因：财报表格结构复杂，需要专门处理"""
    # 常见财务报表表头模式——设计原因：规则匹配快速且准确
    TABLE_PATTERNS = {
        "balance_sheet": [
            r"(合并|母公司)?.*资产负[债责]表",
            r"资产负债表",
        ],
        "income_statement": [
            r"(合并|母公司)?.*利润表",
            r"损益表",
        ],
        "cash_flow": [
            r"(合并|母公司)?.*现金流量表",
            r"现金流表",
        ],
    }
    # 金额单位映射——设计原因：必须统一单位，否则同比计算会差万倍
    UNIT_MAP = {
        "元": 1,
        "万元": 1e4,
        "百万元": 1e6,
        "亿元": 1e8,
    }
    def extract_table(self, pdf_page_text: str) -> List[Dict]:
        """提取表格——设计原因：识别表头后按行解析，保持行列关系"""
        tables = []
        # 识别表格类型——设计原因：不同类型有不同字段，不能混着用
        table_type = self._identify_table_type(pdf_page_text)
        # 识别金额单位——设计原因：单位不一会导致后续计算全错
        unit = self._identify_unit(pdf_page_text)
        # 识别合并/母公司口径——设计原因：这是财报分析最易错的点
        caliber = self._identify_caliber(pdf_page_text)
        if table_type and unit:
            tables.append({
                "type": table_type,
                "unit": unit,
                "caliber": caliber,
                "rows": []  # 实际需要解析表格行
            })
        return tables
    def _identify_table_type(self, text: str) -> Optional[str]:
        """识别表格类型——设计原因：关键词匹配，覆盖不同写法"""
        for table_type, patterns in self.TABLE_PATTERNS.items():
            for pattern in patterns:
                if re.search(pattern, text):
                    return table_type
        return None
    def _identify_unit(self, text: str) -> float:
        """识别金额单位——设计原因：在表头上方或下方通常会注明"""
        for unit_str, multiplier in self.UNIT_MAP.items():
            if f"单位：{unit_str}" in text or f"单位:{unit_str}" in text:
                return multiplier
        return 1.0  # 默认元
    def _identify_caliber(self, text: str) -> str:
        """识别报告口径——设计原因：合并和母公司数据差一个量级"""
        if "母公司" in text:
            return "parent"
        elif "合并" in text:
            return "consolidated"
        else:
            return "unknown"
class MetricCalculator:
    """指标计算器——设计原因：确定性计算，不依赖LLM"""
    @staticmethod
    def yoy_growth(current: float, previous: float) -> Optional[float]:
        """同比增长率——设计原因：处理除零和负数情况"""
        if previous == 0:
            return None
        return (current - previous) / abs(previous) * 100
    @staticmethod
    def gross_margin(revenue: float, cost: float) -> Optional[float]:
        """毛利率——设计原因：百分比表示，检查分母非零"""
        if revenue == 0:
            return None
        return (revenue - cost) / revenue * 100
    @staticmethod  
    def debt_ratio(total_liabilities: float, total_assets: float) -> Optional[float]:
        """资产负债率——设计原因：指标 > 100% 表示资不抵债"""
        if total_assets == 0:
            return None
        return total_liabilities / total_assets * 100
    @staticmethod
    def current_ratio(current_assets: float, 
                      current_liabilities: float) -> Optional[float]:
        """流动比率——设计原因：标准值2，低于1有短期偿债风险"""
        if current_liabilities == 0:
            return None
        return current_assets / current_liabilities
    def compute_all(self, 
                    current_metrics: Dict[str, FinancialMetric],
                    previous_metrics: Dict[str, FinancialMetric]) -> Dict[str, float]:
        """计算所有指标——设计原因：一次调用输出全部分析指标"""
        results = {}
        # 营业收入同比增长
        if "revenue" in current_metrics and "revenue" in previous_metrics:
            results["revenue_yoy"] = self.yoy_growth(
                current_metrics["revenue"].value,
                previous_metrics["revenue"].value
            )
        # 净利润同比增长
        if "net_profit" in current_metrics and "net_profit" in previous_metrics:
            results["profit_yoy"] = self.yoy_growth(
                current_metrics["net_profit"].value,
                previous_metrics["net_profit"].value
            )
        # 毛利率
        if "revenue" in current_metrics and "cost" in current_metrics:
            results["gross_margin"] = self.gross_margin(
                current_metrics["revenue"].value,
                current_metrics["cost"].value
            )
        # 资产负债率
        if "total_liabilities" in current_metrics and "total_assets" in current_metrics:
            results["debt_ratio"] = self.debt_ratio(
                current_metrics["total_liabilities"].value,
                current_metrics["total_assets"].value
            )
        # 流动比率
        if "current_assets" in current_metrics and "current_liabilities" in current_metrics:
            results["current_ratio"] = self.current_ratio(
                current_metrics["current_assets"].value,
                current_metrics["current_liabilities"].value
            )
        return results
class FinancialInsightGenerator:
    """财务洞察生成器——设计原因：LLM只负责语义分析，数字全部由计算引擎提供"""
    def __init__(self, llm_client=None):
        self.llm = llm_client
        self.calculator = MetricCalculator()
    def generate_analysis(self, 
                          company_name: str,
                          current: Dict[str, FinancialMetric],
                          previous: Dict[str, FinancialMetric]) -> FinancialAnalysis:
        """生成财务分析报告——设计原因：计算+推理的完整管线"""
        # 第一步：精确计算所有指标——设计原因：禁止LLM参与计算
        metrics = self.calculator.compute_all(current, previous)
        # 第二步：构建事实上下文——设计原因：作为LLM推理的唯一信息源
        context = self._build_context(company_name, current, previous, metrics)
        # 第三步：LLM推理生成洞察——设计原因：LLM做语义分析，不做数值计算
        insights = self._generate_insights(context, current)
        # 第四步：交叉校验——设计原因：检查LLM是否引用了计算数据
        verified_insights = self._verify_insights(insights, metrics)
        return FinancialAnalysis(
            company_name=company_name,
            report_period="2024年度",
            metrics=list(current.values()),
            insights=verified_insights
        )
    def _build_context(self, name: str,
                       current: Dict, previous: Dict, 
                       computed: Dict) -> str:
        """构建分析上下文——设计原因：结构化数据转自然语言传LLM"""
        parts = [f"公司：{name}"]
        # 收入相关
        if "revenue" in current:
            cur_rev = current["revenue"]
            parts.append(
                f"营业收入：{cur_rev.format_display()}元 "
                f"({cur_rev.caliber}口径)"
            )
            if "revenue_yoy" in computed and computed["revenue_yoy"] is not None:
                parts.append(
                    f"营业收入同比增长：{computed['revenue_yoy']:.2f}%"
                )
        # 利润相关
        if "net_profit" in current:
            parts.append(
                f"净利润：{current['net_profit'].format_display()}元"
            )
            if "profit_yoy" in computed and computed["profit_yoy"] is not None:
                parts.append(
                    f"净利润同比增长：{computed['profit_yoy']:.2f}%"
                )
        # 盈利能力
        if "gross_margin" in computed and computed["gross_margin"] is not None:
            parts.append(f"毛利率：{computed['gross_margin']:.2f}%")
        # 偿债能力
        if "debt_ratio" in computed and computed["debt_ratio"] is not None:
            parts.append(f"资产负债率：{computed['debt_ratio']:.2f}%")
        if "current_ratio" in computed and computed["current_ratio"] is not None:
            parts.append(f"流动比率：{computed['current_ratio']:.2f}")
        return "n".join(parts)
    def _generate_insights(self, context: str, 
                           metrics: Dict) -> List[Dict]:
        """LLM生成洞察——设计原因：基于计算数据做推理，不是基于训练数据"""
        prompt = f"""请基于以下财务数据分析生成3-5条洞察。
你必须基于提供的数字进行分析，不得编造数据。
每条洞察不超过80字。
{context}
要求：
1. 每条洞察必须引用具体的数字
2. 不能生成未提供的数据
3. 分析应覆盖盈利能力、成长性、偿债能力
4. 如果有异常数据（如负增长、比率过高/过低），需特别指出"""
        # 实际调用LLM
        response = "基于数据分析的洞察结果"  # 占位
        # 解析LLM输出为结构化洞察——设计原因：方便校验和展示
        return [
            {
                "text": insight.strip(),
                "confidence": 0.8,
                "source_metrics": ["revenue_yoy", "gross_margin"]
            }
            for insight in response.split("n") if insight.strip()
        ]
    def _verify_insights(self, insights: List[Dict], 
                         computed: Dict) -> List[Dict]:
        """交叉校验——设计原因：检查LLM的数字是否与计算结果一致"""
        verified = []
        for insight in insights:
            text = insight["text"]
            # 提取LLM输出中的数字——设计原因：检查是否和计算值一致
            numbers_in_text = re.findall(r"[-+]?d+.?d*%?", text)
            # 检查数字是否在计算指标中出现过
            has_verified_source = True
            for num_str in numbers_in_text:
                num_str = num_str.replace("%", "")
                try:
                    num = float(num_str)
                    # 检查该数字是否出现在计算指标中（容差0.1%）
                    found = any(
                        abs(v - num) / max(abs(v), 1) < 0.001
                        for v in computed.values()
                        if v is not None
                    )
                    if not found:
                        has_verified_source = False
                        insight["confidence"] *= 0.5
                except ValueError:
                    continue
            if has_verified_source:
                insight["confidence"] = min(1.0, insight["confidence"] * 1.1)
            verified.append(insight)
        return verified
# 使用示例
def analyze_financial_report():
    """完整财报分析流程——设计原因：端到端演示数据流"""
    extractor = FinancialTableExtractor()
    generator = FinancialInsightGenerator()
    # 模拟提取的财务数据
    current_metrics = {
        "revenue": FinancialMetric(
            "营业收入", 50.5e8, ReportPeriod.ANNUAL, 2024,
            "consolidated", "合并利润表", "B4"
        ),
        "net_profit": FinancialMetric(
            "净利润", 5.2e8, ReportPeriod.ANNUAL, 2024,
            "consolidated", "合并利润表", "B25"
        ),
        "cost": FinancialMetric(
            "营业成本", 32.0e8, ReportPeriod.ANNUAL, 2024,
            "consolidated", "合并利润表", "B6"
        ),
        "total_assets": FinancialMetric(
            "总资产", 120.0e8, ReportPeriod.ANNUAL, 2024,
            "consolidated", "合并资产负债表", "C40"
        ),
        "total_liabilities": FinancialMetric(
            "总负债", 65.0e8, ReportPeriod.ANNUAL, 2024,
            "consolidated", "合并资产负债表", "C60"
        ),
        "current_assets": FinancialMetric(
            "流动资产", 45.0e8, ReportPeriod.ANNUAL, 2024,
            "consolidated", "合并资产负债表", "C15"
        ),
        "current_liabilities": FinancialMetric(
            "流动负债", 30.0e8, ReportPeriod.ANNUAL, 2024,
            "consolidated", "合并资产负债表", "C75"
        ),
    }
    previous_metrics = {
        "revenue": FinancialMetric(
            "营业收入", 42.0e8, ReportPeriod.ANNUAL, 2023,
            "consolidated", "合并利润表", "B4"
        ),
        "net_profit": FinancialMetric(
            "净利润", 4.5e8, ReportPeriod.ANNUAL, 2023,
            "consolidated", "合并利润表", "B25"
        ),
    }
    # 生成分析
    analysis = generator.generate_analysis(
        "示例科技股份有限公司",
        current_metrics,
        previous_metrics
    )
    print(f"分析报告: {analysis.company_name}")
    for i, insight in enumerate(analysis.insights, 1):
        print(f"洞察{i} (置信度{insight['confidence']:.2f}): {insight['text']}")
analyze_financial_report()

核心设计是"计算与推理分离"：指标计算由确定性算法完成，LLM 只负责语义分析。计算层保证数字准确，推理层保证分析有深度。两者之间的桥梁是"数据上下文"——LLM 看到的不是原始报表，而是计算引擎产出的结构化事实。这样从根本上杜绝了幻觉。

四、个性化边界权衡

计算精度 vs 分析灵活性

严格的计算层保证了精度，但牺牲了分析灵活性——比如 LLM 想分析"经营性现金流/净利润"这个比率，但计算引擎没有预设这个指标，LLM 就无法分析。解决方案是允许 LLM 提出需要的指标类型，计算引擎动态计算——相当于给 LLM 一个"计算请求"的 API 而非固定指标清单。

单品分析 vs 多品对比

单家公司年报分析已经比较复杂，行业内多公司横向对比复杂度乘3-5倍——不同公司会计科目名称不统一（"营业收入"vs"主营业务收入"、"研发费用"可能归在管理费用里）。现阶段方案是：制作行业标准科目映射表，先统一科目再对比。

模型知识 vs 数据事实

LLM 的训练数据里可能包含该公司的历史分析文章，在没有当前数据时倾向于用训练知识"补全"。事实校验层必须严格检查——生成的数字不在提取的数据里，就标记为低置信度。宁可信息少，不能信息错。

五、总结

企业年报 AI 解读的核心工程策略是"计算归计算、推理归推理"。指标计算（同比增长率、毛利率、资产负债率、流动比率）全部由确定性算法完成，LLM 只负责语义分析。LLM 的输入不是原始报表文本，而是计算引擎产出的结构化事实数据。输出须经交叉校验——LLM 生成的数字与计算结果不一致时自动降低置信度。实施中需权衡计算精度与分析灵活性的关系、单品深度分析与多品横向对比的能力边界、LLM 训练知识与当前数据事实的冲突管理。核心原则是"引用即证据"，每一条洞察都须有数据来源。