TextIn xParse+LangChain：财务审计 Agent 一键搞定审计全流程

本文以TextIn xParse为数据底座，详解基于LangChain搭建财务审计Agent的全流程，实现财务文档解析、数据提取、合规检查与异常检测自动化，附完整可落地的实操代码与最佳实践。

一、财务审计的核心业务痛点

在企业风控合规与财务审计场景中，审计人员常陷入多重工作困境：

1. 待处理的财务报表、合同、发票等文档量巨大，人力处理效率低下；

2. 非结构化文档的关键信息提取繁琐，耗时且易出错；

3. 合规性检查需对照多类法规政策，操作流程复杂；

4. 金额、日期等数据的异常检测难度高，易遗漏风险点；

5. 发现问题后，追溯原始文档具体位置的操作繁琐，验证成本高。

二、解决方案

以TextIn xParse的财务文档解析能力为数据底座，结合LangChain构建智能财务审计Agent，可实现审计工作的全流程自动化，精准解决上述痛点：

1. 自动化文档解析：通过xParse Pipeline自动解析PDF、Excel、图片等多格式财务文档；

2. 智能信息提取：从解析结果中精准提取金额、日期、合同条款等关键财务数据；

3. 合规性自动检查：依托知识库和历史案例，自动校验文档是否符合法规与内部政策；

4. 异常自动检测：智能识别金额异常、日期冲突、数据不一致等问题；

5. 结果可追溯：解析过程中保留原始元素和坐标信息，便于问题追溯与人工验证。

三、财务审计Agent整体架构设计

TextIn xParse+LangChain 财务审计Agent架构设计的核心链路为：

财务文档（PDF/Excel/图片）
    ↓
[xParse Pipeline]
    ├─ Parse: 解析财务报表、合同、发票
    ├─ Chunk: 按页面分块（保持页面完整性）
    └─ Embed: 向量化
    ↓
向量数据库（Milvus/Zilliz）
    ↓
[LangChain Agent]
    ├─ Tool 1: extract_financial_data（提取财务数据）
    ├─ Tool 2: check_compliance（合规性检查）
    ├─ Tool 3: detect_anomalies（异常检测）
    └─ Tool 4: vector_search（检索历史案例）
    ↓
审计报告（含引用和追溯信息）

四、环境准备

首先搭建Python虚拟环境并安装相关依赖，同时配置TextIn与大模型的密钥信息，命令如下：

python -m venv .venv && source .venv/bin/activate
pip install "xparse-client>=0.2.5" langchain langchain-core \
            langchain_milvus langchain-community \
            pymilvus python-dotenv
export XTI_APP_ID=your-app-id # 在 TextIn 官网注册获取
export XTI_SECRET_CODE=your-secret-code # 在 TextIn 官网注册获取
export DASHSCOPE_API_KEY=your-dashscope-api-key # 本教程使用通义千问大模型，也可以替换成其他大模型

注：XTI_APP_ID与XTI_SECRET_CODE需在TextIn工作台（https://www.textin.com/console/dashboard/setting）获取，示例选用通义千问大模型，也可替换为其他大模型。

五、财务审计Agent的实操构建步骤

基于TextIn xParse完成财务文档解析配置，再逐步构建LangChain工具与Agent，全程代码可直接复用。

Step 1：配置xParse Pipeline

针对财务审计场景，配置专属的xParse Pipeline，我们使用以下配置：

• by_page-按页面分块保持完整性

• include_orig_elements=True-保留原始元素便于追溯

• 优化表格提取保证结构完整

from xparse_client import create_pipeline_from_config
import os
from dotenv import load_dotenv

load_dotenv()

AUDIT_PIPELINE_CONFIG = {
    "source": {
        "type": "local",
        "directory": "./audit_documents",
        "pattern": ["*.pdf", "*.xlsx", "*.xls", "*.png", "*.jpg", "*.txt", "*.docx", "*.doc"]  # 支持多种财务文档格式
    },
    "destination": {
        "type": "milvus",
        "db_path": "./audit_vectors.db",
        "collection_name": "audit_documents",
        "dimension": 1024
    },
    "api_base_url": "https://api.textin.com/api/xparse",
    "api_headers": {
        "x-ti-app-id": os.getenv("XTI_APP_ID"),
        "x-ti-secret-code": os.getenv("XTI_SECRET_CODE")
    },
    "stages": [
        {
            "type": "parse",
            "config": {
                "provider": "textin"# 使用TextIn解析引擎，对表格识别效果好
            }
        },
        {
            "type": "chunk",
            "config": {
                "strategy": "by_page",  # 按页面分块，保持页面完整性
                "include_orig_elements": True,  # 保留原始元素，便于追溯
                "max_characters": 2048,  # 财务文档页面可能较长
                "overlap": 100# 页面间重叠，保持上下文
            }
        },
        {
            "type": "embed",
            "config": {
                "provider": "qwen",
                "model_name": "text-embedding-v3"
            }
        }
    ]
}

def run_audit_pipeline() -> None:
    """运行审计文档处理Pipeline"""
    pipeline = create_pipeline_from_config(AUDIT_PIPELINE_CONFIG)
    pipeline.run()

Step 2：构建LangChain Tools

Tool 1：提取财务数据

提取金额、日期、合同条款、发票信息等关键数据，通过正则匹配从解析后的文档中精准抓取。

from langchain_core.tools import Tool
from langchain_milvus import Milvus
from langchain_community.embeddings import DashScopeEmbeddings
import re
import json
import os

embedding = DashScopeEmbeddings(
    model="text-embedding-v3",
    dashscope_api_key=os.getenv("DASHSCOPE_API_KEY"),
)
vector_store = Milvus(
    embedding_function=embedding,
    collection_name="audit_documents",
    connection_args={"uri": "./audit_vectors.db"},
    primary_field="element_id",
    text_field="text",
    vector_field="embeddings",
    enable_dynamic_field=True,
)

def extract_financial_data(query: str) -> str:
    """
    从文档中提取财务数据
    
    支持提取：
    - 金额（人民币、美元等）
    - 日期（合同签署日期、付款日期等）
    - 合同条款（签约方、违约责任等）
    - 发票信息（发票号、税号等）
    """
    # 检索相关文档片段
    docs = vector_store.similarity_search(query, k=5)
    
    # 提取关键信息
    results = []
    for doc in docs:
        text = doc.page_content
        metadata = doc.metadata
        
        # 提取金额
        amounts = re.findall(r'[¥$€]\s*[\d,]+\.?\d*|[\d,]+\.?\d*\s*[元美元欧元]', text)
        
        # 提取日期
        dates = re.findall(r'\d{4}[-年]\d{1,2}[-月]\d{1,2}[日]?|\d{4}/\d{1,2}/\d{1,2}', text)
        
        # 提取合同相关信息
        contract_info = {}
        if"合同"in text or"协议"in text:
            parties = re.findall(r'(甲方|乙方|签约方)[：:]\s*([^\n]+)', text)
            contract_info["parties"] = dict(parties)
        
        results.append({
            "file": metadata.get("filename", "unknown"),
            "page": metadata.get("page_number", "unknown"),
            "amounts": amounts[:5],  # 限制数量
            "dates": dates[:5],
            "contract_info": contract_info,
            "snippet": text[:200]
        })
    
    return json.dumps(results, ensure_ascii=False, indent=2)

Tool 2：合规性检查

校验合同条款、财务数据、发票信息等是否合规，内置自定义规则如超100万金额需特殊审批。

def check_compliance(query: str) -> str:
    """
    检查文档是否符合合规要求
    
    检查项包括：
    - 合同条款是否符合法规要求
    - 财务数据是否符合会计准则
    - 发票信息是否完整
    - 审批流程是否合规
    """
    # 检索相关文档和历史合规案例
    docs = vector_store.similarity_search(query, k=3)
    
    # 合规性检查规则（示例）
    compliance_checks = []
    
    for doc in docs:
        text = doc.page_content
        metadata = doc.metadata
        
        checks = {
            "file": metadata.get("filename", "unknown"),
            "page": metadata.get("page_number", "unknown"),
            "issues": []
        }
        
        # 检查1: 合同金额是否过大（示例规则）
        amounts = re.findall(r'[\d,]+\.?\d*', text)
        for amount_str in amounts:
            try:
                amount = float(amount_str.replace(',', ''))
                if amount > 1000000:  # 超过100万需要特殊审批
                    checks["issues"].append(f"金额 {amount_str} 超过100万，需要特殊审批")
            except:
                pass
        
        # 检查2: 日期是否合理
        dates = re.findall(r'\d{4}[-年]\d{1,2}[-月]\d{1,2}[日]?', text)
        for date_str in dates:
            # 简单检查：日期不能是未来日期（需要更复杂的逻辑）
            pass
        
        # 检查3: 合同关键条款是否存在
        required_terms = ["违约责任", "争议解决", "合同期限"]
        missing_terms = [term for term in required_terms if term notin text]
        if missing_terms:
            checks["issues"].append(f"缺少关键条款: {', '.join(missing_terms)}")
        
        if checks["issues"]:
            compliance_checks.append(checks)
    
    ifnot compliance_checks:
        return"✅ 未发现合规性问题"
    
    return json.dumps(compliance_checks, ensure_ascii=False, indent=2)

Tool 3：异常检测

检测金额异常、日期冲突、数据不一致等风险，通过数据对比实现智能识别。

def detect_anomalies(query: str) -> str:
    """
    检测财务数据中的异常
    
    检测项包括：
    - 金额异常（过大、过小、负数等）
    - 日期冲突（付款日期早于合同日期等）
    - 数据不一致（同一合同在不同文档中金额不同）
    """
    docs = vector_store.similarity_search(query, k=5)
    
    anomalies = []
    
    # 收集所有金额和日期
    all_amounts = []
    all_dates = []
    
    for doc in docs:
        text = doc.page_content
        metadata = doc.metadata
        
        # 提取金额
        amounts = re.findall(r'[\d,]+\.?\d*', text)
        for amount_str in amounts:
            try:
                amount = float(amount_str.replace(',', ''))
                all_amounts.append({
                    "value": amount,
                    "source": metadata.get("filename", "unknown"),
                    "page": metadata.get("page_number", "unknown")
                })
            except:
                pass
        
        # 提取日期
        dates = re.findall(r'\d{4}[-年]\d{1,2}[-月]\d{1,2}[日]?', text)
        all_dates.extend([{
            "value": date,
            "source": metadata.get("filename", "unknown"),
            "page": metadata.get("page_number", "unknown")
        } for date in dates])
    
    # 检测异常
    # 1. 金额异常
    if all_amounts:
        # 过滤掉0值和负数，只保留正数金额
        positive_amounts = [a for a in all_amounts if a["value"] > 0]
        
        if len(positive_amounts) >= 2:
            amounts_values = [a["value"] for a in positive_amounts]
            max_amount = max(amounts_values)
            min_amount = min(amounts_values)
            
            # 确保最小值不为0，避免除以零
            if min_amount > 0:
                ratio = max_amount / min_amount
                if ratio > 1000:  # 金额差异过大
                    anomalies.append({
                        "type": "金额差异异常",
                        "description": f"金额差异异常：最大金额 {max_amount:,.2f} 元与最小金额 {min_amount:,.2f} 元的比例达到 {ratio:.2f}，超过1000倍",
                        "details": [a for a in positive_amounts if a["value"] in [max_amount, min_amount]]
                    })
    
    # 2. 负数金额（可能是错误）
    negative_amounts = [a for a in all_amounts if a["value"] < 0]
    if negative_amounts:
        anomalies.append({
            "type": "负数金额异常",
            "description": "发现负数金额，可能是录入错误",
            "details": negative_amounts
        })
    
    ifnot anomalies:
        return"✅ 未发现异常"
    
    return json.dumps(anomalies, ensure_ascii=False, indent=2)

Tool 4：检索历史案例

基于语义检索历史审计案例，为当前审计工作提供参考。

def search_historical_cases(query: str) -> str:
    """检索历史审计案例"""
    docs = vector_store.similarity_search(query, k=5)
    
    results = []
    for i, doc in enumerate(docs, 1):
        results.append({
            f"案例 {i}": {
                "文件": doc.metadata.get("filename", "unknown"),
                "页码": doc.metadata.get("page_number", "unknown"),
                "内容": doc.page_content[:300] + "...",
                "相似度": "高"if i <= 2else"中"
            }
        })
    
    return json.dumps(results, ensure_ascii=False, indent=2)

组装所有Tools

tools = [
    Tool(
        name="extract_financial_data",
        description="从财务文档中提取关键财务数据，包括金额、日期、合同条款、发票信息等。输入应为要提取的数据类型描述，如'提取合同金额和签署日期'。",
        func=extract_financial_data
    ),
    Tool(
        name="check_compliance",
        description="检查文档是否符合合规要求，包括合同条款合规性、财务数据合规性、发票信息完整性等。输入应为要检查的合规项描述。",
        func=check_compliance
    ),
    Tool(
        name="detect_anomalies",
        description="检测财务数据中的异常，包括金额异常、日期冲突、数据不一致等。输入应为要检测的异常类型描述。",
        func=detect_anomalies
    ),
    Tool(
        name="search_historical_cases",
        description="检索历史审计案例，用于参考和对比。输入应为要检索的案例类型或关键词。",
        func=search_historical_cases
    ),
    Tool(
        name="vector_search",
        description="基于语义检索相关文档片段。输入应为自然语言查询。",
        func=lambda q: "\n\n".join([
            f"[{i+1}] {doc.metadata.get('filename', 'unknown')}\n{doc.page_content[:300]}..."
            for i, doc in enumerate(vector_store.similarity_search(q, k=3))
        ])
    )
]

Step 3：配置LangChain Agent

把Tools组装好后，结合通义千问大模型配置LangChain Agent，设置较低温度保证输出结果的确定性，同时定义系统提示词，让Agent按专业财务审计逻辑作答，核心为引用具体文档信息、提供详细检查结果、明确问题处理建议。

from langchain.agents import create_agent
from langchain_community.chat_models import ChatTongyi
import os

llm = ChatTongyi(
    model="qwen-max",
    dashscope_api_key=os.getenv("DASHSCOPE_API_KEY"),
    temperature=0,  # 使用较低温度以获得更确定性的输出
)

agent = create_agent(
    model=llm,
    tools=tools,
    debug=True,
    system_prompt="""你是一个专业的财务审计助手。你的任务是帮助审计人员：
1. 从财务文档中提取关键信息
2. 检查文档的合规性
3. 检测数据异常
4. 检索历史审计案例

在回答时，请：
- 引用具体的文档名称和页码
- 提供详细的数据和检查结果
- 如果发现问题，说明问题的严重程度和建议的处理方式
- 使用工具获取准确的信息，不要猜测
"""
)

Step 4：完整示例代码

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
财务审计Agent完整示例
"""

import os
import re
import json
from dotenv import load_dotenv
from xparse_client import create_pipeline_from_config
from langchain_core.tools import Tool
from langchain.agents import create_agent
from langchain_community.chat_models import ChatTongyi
from langchain_milvus import Milvus
from langchain_community.embeddings import DashScopeEmbeddings

load_dotenv()

class AuditAgent:
    """财务审计Agent"""
    
    def __init__(self):
        self.setup_pipeline()
        self.setup_vector_store()
        self.setup_agent()
    
    def setup_pipeline(self):
        """配置Pipeline"""
        self.pipeline_config = {
            "source": {
                "type": "local",
                "directory": "./audit_documents",
                "pattern": ["*.pdf", "*.xlsx", "*.xls", "*.png", "*.jpg", "*.txt", "*.docx", "*.doc"]
            },
            "destination": {
                "type": "milvus",
                "db_path": "./audit_vectors.db",
                "collection_name": "audit_documents",
                "dimension": 1024
            },
            "api_base_url": "https://api.textin.com/api/xparse",
            "api_headers": {
                "x-ti-app-id": os.getenv("XTI_APP_ID"),
                "x-ti-secret-code": os.getenv("XTI_SECRET_CODE")
            },
            "stages": [
                {
                    "type": "parse",
                    "config": {"provider": "textin"}
                },
                {
                    "type": "chunk",
                    "config": {
                        "strategy": "by_page",
                        "include_orig_elements": True,
                        "max_characters": 2048,
                        "overlap": 100
                    }
                },
                {
                    "type": "embed",
                    "config": {
                        "provider": "qwen",
                        "model_name": "text-embedding-v3"
                    }
                }
            ]
        }
    
    def setup_vector_store(self):
        """配置向量数据库"""
        self.embedding = DashScopeEmbeddings(
            model="text-embedding-v3",
            dashscope_api_key=os.getenv("DASHSCOPE_API_KEY"),
        )
        self.vector_store = Milvus(
            embedding_function=self.embedding,
            collection_name="audit_documents",
            connection_args={"uri": self.pipeline_config["destination"]["db_path"]},
            primary_field="element_id",
            text_field="text",
            vector_field="embeddings",
            enable_dynamic_field=True,
        )
    
    def setup_agent(self):
        """配置Agent和Tools"""
        # 定义Tools（简化版，完整版见上文）
        tools = [
            Tool(
                name="extract_financial_data",
                description="提取财务数据：金额、日期、合同条款等",
                func=self.extract_financial_data
            ),
            Tool(
                name="check_compliance",
                description="检查合规性：合同条款、财务数据合规性等",
                func=self.check_compliance
            ),
            Tool(
                name="detect_anomalies",
                description="检测异常：金额异常、日期冲突等",
                func=self.detect_anomalies
            ),
            Tool(
                name="vector_search",
                description="语义检索相关文档",
                func=lambda q: "\n\n".join([
                    f"[{i+1}] {doc.metadata.get('filename', 'unknown')}\n{doc.page_content[:300]}..."
                    for i, doc in enumerate(self.vector_store.similarity_search(q, k=3))
                ])
            )
        ]
        
        llm = ChatTongyi(
            model="qwen-max",
            dashscope_api_key=os.getenv("DASHSCOPE_API_KEY"),
            temperature=0,  # 使用较低温度以获得更确定性的输出
        )
        
        self.agent = create_agent(
            model=llm,
            tools=tools,
            debug=True
        )
    
    def extract_financial_data(self, query: str) -> str:
        """提取财务数据"""
        docs = self.vector_store.similarity_search(query, k=5)
        results = []
        for doc in docs:
            text = doc.page_content
            amounts = re.findall(r'[¥$€]\s*[\d,]+\.?\d*|[\d,]+\.?\d*\s*[元美元欧元]', text)
            dates = re.findall(r'\d{4}[-年]\d{1,2}[-月]\d{1,2}[日]?', text)
            results.append({
                "file": doc.metadata.get("filename", "unknown"),
                "page": doc.metadata.get("page_number", "unknown"),
                "amounts": amounts[:5],
                "dates": dates[:5],
                "snippet": text[:200]
            })
        return json.dumps(results, ensure_ascii=False, indent=2)
    
    def check_compliance(self, query: str) -> str:
        """合规性检查"""
        docs = self.vector_store.similarity_search(query, k=3)
        issues = []
        for doc in docs:
            text = doc.page_content
            amounts = re.findall(r'[\d,]+\.?\d*', text)
            for amount_str in amounts:
                try:
                    amount = float(amount_str.replace(',', ''))
                    if amount > 1000000:
                        issues.append({
                            "file": doc.metadata.get("filename", "unknown"),
                            "page": doc.metadata.get("page_number", "unknown"),
                            "issue": f"金额 {amount_str} 超过100万，需要特殊审批"
                        })
                except:
                    pass
        return json.dumps(issues, ensure_ascii=False, indent=2) if issues else"✅ 未发现合规性问题"
    
    def detect_anomalies(self, query: str) -> str:
        """异常检测"""
        docs = self.vector_store.similarity_search(query, k=5)
        anomalies = []
        all_amounts = []
        for doc in docs:
            amounts = re.findall(r'[\d,]+\.?\d*', doc.page_content)
            for amount_str in amounts:
                try:
                    amount = float(amount_str.replace(',', ''))
                    # 过滤掉0值和负数，只保留正数金额
                    if amount > 0:
                        all_amounts.append(amount)
                except:
                    pass
        
        # 检查是否有足够的金额数据进行比较，并避免除以零
        if len(all_amounts) >= 2:
            min_amount = min(all_amounts)
            max_amount = max(all_amounts)
            # 确保最小值不为0，避免除以零
            if min_amount > 0:
                ratio = max_amount / min_amount
                if ratio > 1000:
                    anomalies.append(f"金额差异异常：最大金额 {max_amount:,.2f} 元与最小金额 {min_amount:,.2f} 元的比例达到 {ratio:.2f}，超过1000倍")
            elif max_amount > 0:
                # 如果最小值为0但最大值不为0，这也是一种异常
                anomalies.append(f"发现零金额异常：存在金额为0的记录，同时存在金额为 {max_amount:,.2f} 元的记录")
        
        return json.dumps(anomalies, ensure_ascii=False, indent=2) if anomalies else"✅ 未发现异常"
    
    def process_documents(self):
        """处理文档"""
        print("=" * 60)
        print("开始处理审计文档...")
        print("=" * 60)
        
        pipeline = create_pipeline_from_config(self.pipeline_config)
        pipeline.run()
        
        print("\n文档处理完成！")
    
    def query(self, question: str) -> str:
        """查询Agent"""
        from langchain_core.messages import HumanMessage
        response = self.agent.invoke({
            "messages": [HumanMessage(content=question)]
        })
        return response["messages"][-1].content

def main():
    """主函数"""
    agent = AuditAgent()
    
    # 1. 处理文档（首次运行）
    agent.process_documents()
    
    # 2. 查询示例
    questions = [
        "提取所有合同中的金额和签署日期",
        "检查这些合同是否符合合规要求",
        "检测是否有金额异常的情况",
        "检索类似的历史审计案例"
    ]
    
    for question in questions:
        print(f"\n{'='*60}")
        print(f"问题: {question}")
        print(f"{'='*60}")
        answer = agent.query(question)
        print(f"\n回答:\n{answer}")

if __name__ == "__main__":
    main()

六、使用示例

TextIn xParse+LangChain 财务审计Agent调用简单，仅需实例化后通过query方法输入自然语言指令，即可完成各类审计操作，核心示例如下：

1. 提取财务数据：

agent = AuditAgent()
response = agent.query("从财务报表中提取所有超过10万的金额和对应的日期")
print(response)

2. 合规性检查：

response = agent.query("检查所有合同是否符合以下要求：1) 金额超过100万需要特殊审批 2) 必须包含违约责任条款")
print(response)

3. 异常检测：

response = agent.query("检测财务报表中是否有异常：金额为负数、日期不合理、同一合同金额不一致等")
print(response)

七、财务审计Agent的最佳实践

为保证TextIn xParse+LangChain 财务审计Agent的运行效率和财务文档解析的准确性，落地时需遵循这些实践原则：

1. 文档预处理：保证文档格式统一、命名规范，便于批量解析；

2. 分块策略：使用by_page模式，保持页面完整性，便于问题追溯；

3.  原始元素保留：开启include_orig_elements参数，便于后续人工验证；

4. 合规规则配置：将合规规则单独存储在配置文件，便于更新维护；

5. 异常阈值设置：根据企业业务需求，自定义合理的异常检测阈值；

6. 结果追溯：让Agent回答包含文档名称和页码，降低人工验证成本。

八、常见问题解答

Q：如何处理加密的PDF文档？
A：在xParse配置中添加pdf_pwd参数，或在Pipeline运行前先完成文档解密，再进行文档解析。

Q：如何提高文档解析和数据提取的准确率？
A：一是使用text-embedding-v3模型做向量化；二是根据文档类型优化分块策略；三是适当增加检索的文档数量k值。

Q：如何将TextIn xParse+LangChain 财务审计Agent集成到现有审计系统？
A：可将Agent封装为REST API，通过HTTP接口实现调用，也可直接将Agent代码集成到企业现有审计工作流中。

​

本文所有产品性能效果和案例数据仅供参考，均不作为履约依据，具体效果请以本司实测效果为准。