在企业财务审计和合规审核场景中，审计人员面临以下挑战：

通过构建财务审计 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 -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 # 本教程使用通义千问大模型，也可以替换成其他大模型

针对财务审计场景，我们使用以下配置：

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()

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)

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 not in text]
        if missing_terms:
            checks["issues"].append(f"缺少关键条款: {', '.join(missing_terms)}")
        
        if checks["issues"]:
            compliance_checks.append(checks)
    
    if not compliance_checks:
        return "✅ 未发现合规性问题"
    
    return json.dumps(compliance_checks, ensure_ascii=False, indent=2)

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
        })
    
    if not anomalies:
        return "✅ 未发现异常"
    
    return json.dumps(anomalies, ensure_ascii=False, indent=2)

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 <= 2 else "中"
            }
        })
    
    return json.dumps(results, ensure_ascii=False, indent=2)

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))
        ])
    )
]

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. 检索历史审计案例

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

#!/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()

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

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

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

 参数，或在 Pipeline 运行前先解密文档。

 模型；2) 优化分块策略；3) 增加检索的文档数量（k 值）。

A: 可以将 Agent 封装为 REST API，通过 HTTP 接口调用，或集成到现有的审计工作流中。

A: 在 xParse 配置中添加 pdf_pwd 参数，或在 Pipeline 运行前先解密文档。

财务审计Agent：自动化合规审核与异常检测

本教程介绍如何利用xParse构建财务审计智能Agent，解决财务文档处理中的五大痛点：文档量大、信息提取繁琐、合规检查复杂、异常检测困难和追溯验证不便。通过xParse Pipeline实现自动化文档解析，按页面分块保持完整性，并保留原始元素便于追溯。

人工智能

大数据

职业发展

本教程介绍如何利用xParse构建智能财务审计Agent，实现财务文档自动解析、关键信息提取、合规性检查和异常检测功能。通过AI技术解决审计工作中文档量大、信息提取繁琐、合规检查复杂等痛点，提升审计效率和准确性。

向量数据库

数据合规

数据库

Agent

2026采购季 | AI焕新·智启新局

tencentdb-catalog

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云AI代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

功能1上新10个字符

功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符。

功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符。

功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符

功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符

功能4上新

文章&问答评论现已支持表情

全新交互，全新视觉，新增快捷键、悬浮工具栏、高亮块等功能并同时优化现有功能，全面提升创作效率和体验

社区富文本编辑器全新改版！诚邀体验～ 

精选全网热门MCP server，让你的AI更好用 🚀

💥开发者 MCP广场重磅上线！

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

财务审计Agent：自动化合规审核与异常检测

财务审计Agent：自动化合规审核与异常检测

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐