财务文档的AI处理管线

APPLICATION Jan 1, 2025

推荐： ComfyUI-PandasAI | ComfyUI-Autodistill | ComfyUI节点速查手册

SEC 10-K 报告等财务文件通常处理起来复杂而繁琐，但有了正确的工作流程，整个流程就可以改变。

我开发了一个集成 Phidata、n8n 和 Qdrant 的简化解决方案，以自动执行财务文件分析。此工作流程无缝处理报告，提取精确且结构化的见解，并将其存储在强大的矢量数据库 Qdrant 中，从而实现高级搜索和检索功能。结果是一种更快、更有效地处理财务数据的方式，使决策者可以访问和采取行动。

通过这种方法，管理大规模财务运营不仅变得更简单，而且更智能，利用人工智能和自动化来推动有影响力的结果。

1、架构

该架构集成了各种组件，以创建高效且简化的流程来处理 SEC 10-K 财务文件。它从本地文件触发器开始，该触发器监视本地系统上的指定目录以查找新的财务文件。一旦检测到文件，它就会启动工作流程。系统的核心是使用 phidata 设计的 Sec10k 代理，phidata 是一个定制的财务分析代理，由 Claude 3.5 Sonnet 模型提供支持。该代理专门用于分析财务文件并生成结构化的 JSON 输出，使用 PDF 分析工具和精确指令集的组合。该代理还配置了调试和流式传输功能，以确保可靠且动态的分析过程。

然后，处理后的数据流入 Qdrant Vector Store，这是一个强大的矢量数据库，用于存储和管理嵌入。这些嵌入由 Embedding Ollama 模块生成，该模块将提取的内容转换为适合高级搜索和检索的矢量化表示。为了确保文档得到有效处理，默认数据加载器负责文档的准备工作，确保它们满足后续操作的必要要求。内容通过递归字符文本分割器进一步细化，它将文本分解为可管理的块，同时保留其语义完整性。

2、测试文档

为了撰写本文，我考虑了一份 sec 10-Q 文档，你可以下载 pdf 进行实验。导入以下 JSON 文件以创建 n8n 工作流。

{
  "name": "filechange2qdrant",
  "nodes": [
    {
      "parameters": {
        "mode": "insert",
        "qdrantCollection": {
          "__rl": true,
          "value": "multi_document_agent",
          "mode": "list",
          "cachedResultName": "multi_document_agent"
        },
        "options": {}
      },
      "type": "@n8n/n8n-nodes-langchain.vectorStoreQdrant",
      "typeVersion": 1,
      "position": [
        60,
        -100
      ],
      "id": "988b5468-8483-49db-832f-15d77333f391",
      "name": "Qdrant Vector Store",
      "credentials": {
        "qdrantApi": {
          "id": "jbqGna16O2L9iR8V",
          "name": "QdrantApi account"
        }
      }
    },
    {
      "parameters": {
        "method": "POST",
        "url": "https://f309-2401-4900-889d-f100-6539-35f9-5e06-1638.ngrok-free.app/api/v1/analyze",
        "sendBody": true,
        "bodyParameters": {
          "parameters": [
            {
              "name": "query",
              "value": "analyse the financial statement and provide the final response as a structured JSON"
            },
            {
              "name": "file_path",
              "value": "={{ $json.path }}"
            }
          ]
        },
        "options": {}
      },
      "type": "n8n-nodes-base.httpRequest",
      "typeVersion": 4.2,
      "position": [
        -160,
        -100
      ],
      "id": "1657cd1d-246f-4b28-975b-fcc5b1a6edfc",
      "name": "sec10k agent"
    },
    {
      "parameters": {
        "triggerOn": "folder",
        "path": "YOUR_DATA_PATH",
        "events": [
          "add"
        ],
        "options": {
          "usePolling": true
        }
      },
      "type": "n8n-nodes-base.localFileTrigger",
      "typeVersion": 1,
      "position": [
        -440,
        -100
      ],
      "id": "bd57a74a-4d96-4d86-97f2-b376505da7ad",
      "name": "Local File Trigger"
    },
    {
      "parameters": {
        "jsonMode": "expressionData",
        "jsonData": "={{ $('sec10k agent').item.json }}",
        "options": {}
      },
      "type": "@n8n/n8n-nodes-langchain.documentDefaultDataLoader",
      "typeVersion": 1,
      "position": [
        240,
        100
      ],
      "id": "e4e22dd9-cac9-4bcc-a893-d5339f99a49c",
      "name": "Default Data Loader"
    },
    {
      "parameters": {
        "options": {}
      },
      "type": "@n8n/n8n-nodes-langchain.textSplitterRecursiveCharacterTextSplitter",
      "typeVersion": 1,
      "position": [
        340,
        260
      ],
      "id": "bcbfa95e-a1c2-4144-8525-c7e43b231b3c",
      "name": "Recursive Character Text Splitter"
    },
    {
      "parameters": {
        "model": "nomic-embed-text:latest"
      },
      "type": "@n8n/n8n-nodes-langchain.embeddingsOllama",
      "typeVersion": 1,
      "position": [
        100,
        100
      ],
      "id": "e634d2e6-c6ca-4e31-9dfd-f63d182731c5",
      "name": "Embeddings Ollama",
      "credentials": {
        "ollamaApi": {
          "id": "3fAFU0fFchwovvbD",
          "name": "Ollama account"
        }
      }
    }
  ],
  "pinData": {},
  "connections": {
    "sec10k agent": {
      "main": [
        [
          {
            "node": "Qdrant Vector Store",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Local File Trigger": {
      "main": [
        [
          {
            "node": "sec10k agent",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Default Data Loader": {
      "ai_document": [
        [
          {
            "node": "Qdrant Vector Store",
            "type": "ai_document",
            "index": 0
          }
        ]
      ]
    },
    "Recursive Character Text Splitter": {
      "ai_textSplitter": [
        [
          {
            "node": "Default Data Loader",
            "type": "ai_textSplitter",
            "index": 0
          }
        ]
      ]
    },
    "Embeddings Ollama": {
      "ai_embedding": [
        [
          {
            "node": "Qdrant Vector Store",
            "type": "ai_embedding",
            "index": 0
          }
        ]
      ]
    }
  },
  "active": true,
  "settings": {
    "executionOrder": "v1"
  },
  "versionId": "e4f7ee4a-b657-46a5-b2ff-82b7cd6c3aad",
  "meta": {
    "instanceId": "e711fbe877d128d86a078d3ddcaeb0c456781dc70945c5f7c313501777f80a45"
  },
  "id": "FHzKIgnbgnbIcZu8",
  "tags": []
}

3、代理实现

首先，项目脚手架如下：

├── api_server.py
├── data
│   └── 0001437749-24-035313.pdf
├── phidata
│   ├── __init__.py
│   ├── anthropic_utility.py
│   ├── financial_agent.py
│   └── financial_models.py
└── requirements.txt

requirements.txt 文件如下所示，该文件具有使用 fastapi 将代理公开为 API 的依赖模块。

phidata==2.7.5
anthropic==0.42.0
openai==1.58.1
python-dotenv==1.0.1
pypdf==5.0.1
fastapi==0.115.6
uvicorn==0.34.0

代理需要 Anthropic API 密钥才能与 Anthropic API 交互。在根文件夹中创建一个 .env 文件并将 API 密钥保存在那里。我的情况是，我同时尝试了 OpenAI 和 Anthropic，所以我两者都有。

OPENAI_API_KEY=sk-proj-****
ANTHROPIC_API_KEY=sk-ant-****

现在让我们从如何构建 Agentic 工具的提示开始。整个系统依赖于 2 个提示 system_prompt 和 user_prompt

def _create_user_prompt(self, file_type: str, page_count: int) -> str:
        """Create the user prompt for the financial analysis."""
        return f"""I'm sending you a {file_type} document as {page_count} images. Please analyze all pages and extract the following financial metrics:
        - EBIT
        - EBITDA
        - Net Income
        - Revenue
        - Currency
        - Units: (Actuals | Thousands)
        - Depreciation
        - Amortization
        - Filing Date
        - Fiscal Year End
        - Language
        - Country
        
        For each metric, please provide:
        1. The exact value
        2. Where it was found (coordinates/page numbers)
        3. The relevant snippet of text
        4. Your reasoning for the extraction
        5. A confidence score
        6. Whether the value was derived or directly extracted
        7. Any calculations performed
        8. Detect the language and country and populate the fields accordingly.
        
        Please provide the output in valid JSON format matching the provided class structure. No other text is needed just the JSON is sufficient
        """

def _create_system_prompt(self) -> str:
        """Create the system prompt for the financial analysis."""
        return """You are a specialized financial analyst with deep expertise in interpreting 
        corporate filings across multiple languages. Your strength lies in identifying, extracting, and 
        validating financial metrics like EBIT, EBITDA, Net Income, and other key performance indicators. 
        You have been trained to provide detailed documentation of your findings, including coordinate references 
        and contextual snippets. You are meticulous about explaining your reasoning and providing confidence scores 
        for each extraction. No other text is needed just the JSON is sufficient.
        
        You MUST provide your analysis in the following JSON structure:
        {
            "company_name": str,
            "filing_date": str,
            "filing_type": str,
            "currency": str,
            "fiscal_year_end": str,
            "language": str,
            "country": str,
            "unity": str
            "metrics": [
                {
                    "attribute": str,
                    "value": float,
                    "coordinates": str | null,
                    "snippet": str,
                    "reasoning": str,
                    "confidence_score": float,
                    "translation": str | null,
                    "is_derived": bool,
                    "calculation_details": {str: float} | null,
                    "unit": str
                }
            ],
            "confidence_summary": float
        }    
        """

现在，该工具将考虑上述提示，使用下面的pydantic模型制作更结构化的输出：

from pydantic import BaseModel
from typing import Optional, Dict, List


class FinancialMetric(BaseModel):
    attribute: str
    value: float
    coordinates: Optional[str]
    snippet: str
    reasoning: str
    confidence_score: float
    translation: Optional[str]
    is_derived: bool
    calculation_details: Optional[Dict[str, float]]
    unit: str


class FinancialAnalysis(BaseModel):
    company_name: str
    filing_date: str
    filing_type: str
    currency: str
    country: str
    language: str
    unit: str
    fiscal_year_end: str
    metrics: List[FinancialMetric]
    confidence_summary: float

实际调用anthropic的工具以及 pdf 上传，如下所示：

def analyze_financial_filing_pdf(self):
        """Analyze financial filings PDF file and return the financial analysis."""
        try:
            # Load and encode the PDF
            with open(self.pdf_path, "rb") as f:
                pdf_data = base64.b64encode(f.read()).decode("utf-8")
                self.logger.info(f"{self.pdf_path} converted to base64")

            reader = PdfReader(stream=self.pdf_path)
            self.logger.info("creating the Anthropic message contract and calling API")

            message = self.client.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1024,
                system=self._create_system_prompt(),
                messages=[
                    {
                        "role": "user",
                        "content": [
                            {
                                "type": "document",
                                "source": {
                                    "type": "base64",
                                    "media_type": "application/pdf",
                                    "data": pdf_data
                                },
                                "cache_control": {"type": "ephemeral"}
                            },
                            {
                                "type": "text",
                                "text": self._create_user_prompt("PDF", len(reader.pages))
                            }
                        ]
                    }
                ],
            )

            try:
                response_text = message.content[0].text
                print(response_text)
                response_dict = json.loads(response_text)
                analysis = FinancialAnalysis(**response_dict)
                self.logger.info(json.loads(analysis.model_dump_json(indent=2)))
                return analysis.model_dump_json(indent=2)
            except Exception as e:
                self.logger.error(f"Error parsing response: {e}")
                self.logger.info("Raw response:")
                self.logger.info(message.content
                raise

        except Exception as e:
            self.logger.error(f"Error analyzing PDF: {e}")
            raise

现在让我们创建代理，使用上述工具正确分析财务文件（本例中为 sec-10Q）并生成指定的输出：

from phi.agent import Agent, RunResponse  # noqa
from phi.model.anthropic import Claude

from phidata.anthropic_utility import FinancialAnalyzer
from phidata.financial_models import FinancialAnalysis


def init_financial_analyzer():
    analyzer = FinancialAnalyzer()
    return analyzer


def financial_agent(analyzer: FinancialAnalyzer):
    # Agent that uses JSON mode
    json_mode_agent = Agent(
        model=Claude(id="claude-3-5-sonnet-20241022"),
        name="financial Agent",
        description="financial filings Analysis Agent",
        response_model=FinancialAnalysis,
        tools=[analyzer.analyze_financial_filing_pdf],
        show_tool_calls=True,
        tool_call_limit=5,
        reasoning=False,
        instructions=["Your task is to get the financial analysis in a specified json format"],
        stream=True,
        debug_mode=True,
        structured_outputs=False
    )
    return json_mode_agent