财务分析Agent：基于DuckDB+LLM的智能财务数据分析工具
·
@TOC
你好！这是你第一次接触 CW 财务分析 Agent。本文将手把手教你搭建一个能够处理 500MB 以上大型 CSV/Excel 文件 的智能财务分析工具，支持自然语言查询、多文件关联分析、Plan模式交互，并提供桌面 GUI 应用。
项目概述
项目背景
面对 500MB 以上的 CSV 文件（通常包含数百万行数据），直接将整个文件塞给大语言模型（LLM）是不现实的——这会瞬间撑爆上下文窗口，且成本高昂、效率低下。
CW 财务分析 Agent 就是为了解决这一问题而诞生的。它采用 "代码解释器 (Code Interpreter) 模式" + "结构化查询" 的策略，既能保证数据安全，又能进行高效的财务报表分析。
项目目标
| 功能 | 说明 |
|---|---|
| 大文件处理 | 支持 500MB 以上 CSV/Excel 文件 |
| 数据安全 | 所有计算在本地执行，数据不上传云端 |
| 多文件关联 | 支持 Excel + CSV 混合文件，自动检测关联字段 |
| 自然语言查询 | 输入中文即可生成分析结果 |
| Plan 模式 | 输入 → 解析 → 确认 → 执行的交互流程 |
| 本地 LLM | 支持 Ollama 本地部署（deepseek-r1:8b） |
| 桌面应用 | PySimpleGUI 跨平台桌面应用（Windows/Mac） |
项目结构
D:/CW_analyzer/
├── main.py # 命令行入口
├── gui_app.py # GUI桌面应用
├── app.spec # PyInstaller打包配置
├── requirements.txt # 依赖列表
├── .env # 环境配置
├── MAC_DEPLOY.md # Mac部署文档
├── 需求文档.md # 完整需求文档
├── src/
│ ├── agent/
│ │ ├── financial_agent.py # Agent主类
│ │ └── plan_mode.py # Plan模式交互
│ ├── core/
│ │ ├── metadata.py # Schema探测
│ │ ├── code_executor.py # 代码生成+执行
│ │ ├── llm_tools.py # LLM管理
│ │ ├── security.py # 数据安全+审计
│ │ ├── task_planner.py # 任务规划
│ │ └── visualization.py # 可视化
│ └── utils/ # 工具函数
├── data/
│ ├── sample/ # 示例数据
│ └── 25year_4Q/ # 实际数据
├── output/ # 导出文件
├── logs/ # 运行日志
├── audit/ # 审计日志
└── tests/ # 测试文件环境准备
系统要求
-
操作系统: Windows 10/11 或 macOS（Linux 也可）
-
Python: 3.8+（推荐 3.10+）
-
Ollama（可选，用于本地 LLM）
推荐硬件
| 配置 | 最低要求 | 推荐配置 |
|---|---|---|
| 内存 | 8GB | 16GB+ |
| 磁盘 | 10GB 空闲 | 50GB+（处理大文件） |
| GPU | 可选 | 6GB+ 显存（加速 Ollama） |
安装步骤
1. 克隆/复制项目
确保你已经将项目放在 D:/CW_analyzer/ 目录下（也可放在其他位置）：
cd D:/CW_analyzer
2. 创建虚拟环境（推荐）
# Windows PowerShell
python -m venv venv
.\venv\Scripts\Activate.ps1
# 或使用 Conda
conda create -n cw_agent python=3.10 -y
conda activate cw_agent3. 安装依赖
pip install -r requirements.txtrequirements.txt 内容如下：
# 核心依赖
pandas
duckdb
langchain
langchain-openai
langchain-anthropic
openai
anthropic
python-dotenv
openpyxl
# GUI依赖
PySimpleGUI
pyinstaller
# 开发依赖
pytest
flake8
black⚠️ 注意：如果安装
langchain相关包速度慢，可以使用国内镜像源：pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple
4. 配置 Ollama（可选）
如果你有本地 GPU，推荐使用 Ollama 部署本地 LLM：
# 1. 安装 Ollama（从 https://ollama.com 下载）
# 2. 拉取模型
ollama pull deepseek-r1:8b
# 3. 启动 Ollama 服务
ollama serve
# 4. 验证服务
curl http://localhost:11434/api/tags5. 配置环境变量
编辑项目根目录下的 .env 文件：
# Ollama 配置（兼容 OpenAI 格式）
OPENAI_API_KEY=ollama # 使用 Ollama 时随便填
OPENAI_API_BASE=http://localhost:11434/v1 # Ollama 地址
DEFAULT_MODEL=deepseek-r1:8b # 模型名称
# 如果使用 OpenAI 官方 API
# OPENAI_API_KEY=sk-your-key-here
# OPENAI_API_BASE=https://api.openai.com/v1
# DEFAULT_MODEL=gpt-4o运行方式
方式一：命令行模式
python main.py
交互示例：
=== 财务分析Agent (多文件支持) ===
请输入文件路径 (多个文件用逗号分隔):
D:/CW_analyzer/data/25year_4Q/yue12.csv, D:/CW_analyzer/data/25year_4Q/quest.xlsx
正在初始化...
正在加载 yue12.csv (csv)...
已加载为表: yue12, 列数: 61
正在加载 quest.xlsx (excel)...
已加载为表: quest, 列数: 5
已加载 2 个文件
检测到可关联列: ['使用者UIN', '用户ID']
=== 开始分析 ===
输入查询内容 (输入 'exit' 退出):
查询: 按产品统计金额
处理查询中...
--------------------------------------------------
数据 (10 条):
{'产品': '云服务器CVM', '原价': 125000.0, '优惠后价': 89500.0}
...
📋 表格数据: 15 行
==================================================
📊 分析报告
==================================================
📈 盈利统计:
盈利项目数: 12 个
亏损项目数: 3 个
盈利总额: ¥45,678.90
亏损总额: ¥2,345.67
💰 净盈亏: ¥43,333.23
是否需要导出表格文件? (y/n):方式二：GUI 桌面应用
python gui_app.py
界面布局：
┌──────────────────────────────────────────────────────────────┐
│ 💰 财务分析Agent 系统: Windows | Ollama │
├──────────────────────────────────────────────────────────────┤
│ 📁 数据文件 │ 📊 查询结果 │
│ [文件路径输入...] │ ┌────┬────┬────┬────┬────┐ │
│ [🚀 加载数据] │ │账号│产品│月份│折扣│盈利│ │
│ 已加载 2 个文件 │ └────┴────┴────┴────┴────┘ │
│ ──────────────────── │ 显示 100 条记录 │
│ 🔗 Ollama服务器 │ │
│ [http://localhost:...] │ │
│ [🔄 测试连接] │ │
│ ──────────────────── │ │
│ 🔍 查询 │ │
│ [输入查询...] [普通] [Plan] │
│ [▶ 执行] │
│ ──────────────────── │
│ 💾 导出: [CSV▼] [📥 导出结果] │
│ ──────────────────── │
│ 📜 Plan历史 │
│ - 2025-07-15 - 分析盈利 │
└──────────────────────────────────────────────────────────────┘GUI 操作流程：
-
加载数据：在文件路径框输入 CSV/Excel 文件路径，多个用逗号分隔，点击 "加载数据"
-
测试 Ollama：点击 "测试连接" 确保 LLM 可用
-
执行查询：输入查询内容，选择普通或 Plan 模式，点击 "执行"
-
导出结果：选择 CSV/Excel/JSON 格式，点击 "导出结果"
支持的查询场景
普通查询
| 查询示例 | 说明 |
|---|---|
按产品统计金额 |
按产品分组统计原价和优惠后价 |
按使用者排名，查看前10大用户 |
TOP N 分析 |
月度趋势分析 |
按账单月分组统计趋势 |
统计总金额 |
全局汇总统计 |
优惠分析 |
原价 vs 优惠后价对比 |
分析盈利情况 |
多表关联盈利分析（核心功能） |
导出所有用户ID |
导出去重后的用户列表 |
Plan 模式
Plan 模式是 CW Agent 的一大亮点，它将查询拆解为"输入 → 解析 → 确认 → 执行"的交互流程：
| 命令 | 说明 |
|---|---|
/plan 分析盈利 |
进入 Plan 模式，等待用户确认 |
/plan! 分析盈利 |
Plan 模式 + 自动确认执行 |
Plan 模式执行示例：
查询: /plan 分析盈利
📋 **Plan Mode - 执行计划**
---
### 1. 数据源
| 类型 | 表名 | 说明 |
|------|------|------|
| 主表 | yue12 | 账单明细 |
| 配置表 | quest | 客户配置 |
| 关联 | 使用者UIN = 用户ID | 关联字段 |
### 2. 计算逻辑
- **原价**: SUM(原价)
- **应付**: 原价 × 客户折扣
- **实付**: SUM(优惠后总价)
- **盈利**: 实付 - 应付
### 3. 输出维度
| 层级 | 分组 | 字段 |
|------|------|------|
| 层级1 | 账号 | 账号, 原价总计, 实付总计, 应付总计, 盈利金额, 盈亏状态 |
| 层级2 | 按产品 | 账号, 产品, 原价, 实付, 应付, 盈利金额, 盈亏状态 |
| 层级3 | 按月份 | 账号, 月份, 原价, 实付, 盈利金额 |
### 4. 排序规则
盈利金额升序（亏损在前）
---
**输入 'y' 执行** | **输入 'n' 重新生成** | **输入补充内容完善Plan**核心代码解析
下面按文件划分，重点说明关键函数与实现思路。
src/agent/financial_agent.py — Agent 主类
Agent 是系统的核心调度器，协调各模块工作：
class FinancialAgent:
def __init__(self, file_paths, ollama_url=None):
self.file_paths = file_paths
self.code_executor = CodeExecutor(file_paths)
self.security = DataSecurity()
self.task_planner = TaskPlanner()
self.llm_manager = LLMManager(ollama_url=ollama_url)
self.llm = self.llm_manager.get_model()
if self.llm:
self.code_generator = CodeGenerator(self.llm)
self.query_analyzer = QueryAnalyzer(self.llm)
self.result_interpreter = ResultInterpreter(self.llm)
self.plan_mode = PlanMode(self.llm, file_paths, None, None)-
initialize(): 加载所有数据文件，探测 Schema，检测可关联列 -
process_query(): 处理用户查询的主流程 -
支持两种路径：LLM 代码生成 → 自动修复 → 执行，或传统规则引擎
src/core/code_executor.py — 代码生成与执行
这是系统的发动机，包含两大能力：
1. 多文件格式自动检测
def _detect_format(self, path):
"""通过文件头判断真实格式"""
ext = os.path.splitext(path)[1].lower()
if ext in ['.xlsx', '.xls']:
return 'excel'
# 检查文件头
with open(path, 'rb') as f:
header = f.read(4)
if header[:2] == b'PK': # ZIP = xlsx
return 'excel'
if header[:4] == b'\xd0\xcf\x11\xe0': # OLE = xls
return 'excel'
return 'csv'亮点：不依赖文件扩展名，通过文件头二进制魔数判断真实格式。即使
.csv后缀的文件实际是 Excel，也能正确识别。
2. 双模式代码生成
| 模式 | 原理 | 优点 | 缺点 |
|---|---|---|---|
| LLM 生成 | 把表结构和查询意图传给 LLM，让其生成 Python 代码 | 灵活，能处理复杂查询 | 依赖 LLM 可用性，可能输出无效代码 |
| 传统规则 | 基于关键词映射的规则引擎，生成对应的 SQL | 稳定可靠，无需 LLM | 只能处理预定义的查询类型 |
双模式协作流程：
用户查询 ↓
┌─────────────────────┐
│ LLM 代码生成 │ ← 如果 Ollama 可用
│ → 输出验证 │ ← 检查是否为有效代码
│ → 自动修复 │ ← 修复常见错误
│ → 编译检查 │ ← SyntaxError 检查
└─────────┬───────────┘
↓ 失败
┌─────────────────────┐
│ 传统规则引擎 │ ← Fallback
│ → 关键词匹配 │
│ → 生成 SQL │
└─────────┬───────────┘
↓
┌─────────────────────┐
│ 代码安全验证 │ ← 拦截危险操作
├─────────────────────┤
│ 执行代码 │ ← exec() 执行
└─────────────────────┘3. 盈利分析 SQL（核心功能）
核心 SQL
def _generate_profit_sql(self, query, original_query, is_export=False):
# 自动识别哪个是账单表（列数多）、哪个是配置表（列数少）
if len(cols1) < len(cols2):
table1, table2 = table2, table1 # 交换
# 使用 COLUMN_MAP 做列名映射，适应不同命名习惯
bill_user = self._get_column(cols1, '使用者')
bill_amount = self._get_column(cols1, '原价')
bill_discount = self._get_column(cols1, '优惠后价')
...：
SELECT
a.使用者UIN as 账号,
a.产品名称 as 产品,
a.账单月 as 月份,
b.客户简称 as 客户,
MAX(b.用户折扣) as 折扣率,
SUM(TRY_CAST(a.原价 AS DOUBLE)) as 原价总计,
SUM(TRY_CAST(a."优惠后总价(不含税)" AS DOUBLE)) as 实付总计,
SUM(TRY_CAST(a.原价 AS DOUBLE) * COALESCE(TRY_CAST(b.用户折扣 AS DOUBLE), 1.0)) as 应付总计,
(SUM(TRY_CAST(a."优惠后总价(不含税)" AS DOUBLE)) -
SUM(TRY_CAST(a.原价 AS DOUBLE) * COALESCE(TRY_CAST(b.用户折扣 AS DOUBLE), 1.0))) as 盈利金额
FROM yue12 a
JOIN quest b ON a.使用者UIN = b.用户ID
GROUP BY a.使用者UIN, a.产品名称, a.账单月, b.客户简称
ORDER BY 盈利金额盈利公式：盈利 = 实付总额 - 应付总额 = SUM(优惠后总价) - SUM(原价 × 折扣率)
src/core/llm_tools.py — LLM 管理
LLM 管理器
class LLMManager:
def __init__(self, ollama_url=None):
if ollama_url:
self.openai_api_base = f"{ollama_url.rstrip('/')}/v1"
else:
self.openai_api_base = os.getenv('OPENAI_API_BASE', 'https://api.openai.com/v1')
self.models = {
'openai': self._init_openai(), # 支持 Ollama / OpenAI
'anthropic': self._init_anthropic() # 支持 Claude
}代码生成器（最关键的部分）
class CodeGenerator:
def __init__(self, llm):
self.prompt_template = (
"你是一个Python代码生成器，只输出可执行的Python代码...\n"
"{file_info}\n"
"用户查询: {user_query}\n"
"【强制要求】...\n"
"【SQL规则】...\n"
"【盈利分析SQL示例】...\n"
)
def generate_code(self, file_path, metadata, user_query):
response = self.llm.invoke(prompt)
code = extract_code(response) # 提取 ```python ... ``` 中的代码
if code and self._is_valid_code(code):
code = self._fix_generated_code(code, file_path, metadata)
return code
else:
return None # 回退到传统引擎LLM 输出验证机制：
def _is_valid_code(self, code):
"""验证LLM输出是否为有效代码"""
invalid_patterns = ['{', '":', '增长率', '利润率', '盈亏']
for pattern in invalid_patterns:
if pattern in code:
return False # 检测到JSON或分析结论
if 'con.execute' not in code:
return False # 缺少数据库操作
return Truesrc/core/security.py — 数据安全
安全是财务分析工具的重中之重，CW Agent 构建了多层防护：
class DataSecurity:
# 数据脱敏
def scrub_pii(self, data):
"""抹除姓名、手机号、身份证、银行卡等敏感信息"""
# 危险操作拦截
def validate_data_access(self, code):
dangerous_operations = [
r'rmtree\(', r'\.remove\(', r'unlink\(',
r'DROP\s+TABLE', r'DROP\s+DATABASE',
]
# 审计日志
def log_audit(self, action, details):
# 记录到 logs/security_20250715.log
# 同时记录到 audit/audit_20250715.jsonl
# 结果限制
def sanitize_result(self, result):
return result[:100] # 最多返回100条| 防护层 | 措施 |
|---|---|
| 数据脱敏 | 抹除姓名、手机号、身份证、银行卡等敏感信息 |
| 计算本地化 | 所有计算在本地执行，数据不上传 |
| 审计日志 | 记录所有查询和代码执行 |
| 代码安全 | 拦截危险操作（文件删除、网络请求等） |
| 结果限制 | 最多返回100条记录 |
src/core/metadata.py — Schema 探测
class MetadataAnalyzer:
def analyze(self):
con = duckdb.connect(':memory:')
desc = con.execute(
f"DESCRIBE SELECT * FROM read_csv_auto('{path}', nullstr='-')"
).df()
self.metadata['columns'] = [
{'name': row['column_name'], 'dtype': row['column_type']}
for _, row in desc.iterrows()
]DuckDB 的 DESCRIBE 语句 能在不加载全部数据的情况下获取表结构，对 500MB 文件也能秒级响应。
src/agent/plan_mode.py — Plan 模式交互
class PlanMode:
MAX_RETRIES = 3
def run(self, user_query, auto_confirm=False):
plan = self.parse_query(user_query) # LLM 解析为结构化 Plan
self.display_plan(plan) # Markdown 格式展示
choice = self.ask_confirmation() # y/n/补充
if choice == 'execute':
return self.execute_plan() # 执行 SQL
elif choice == 'retry':
return self.parse_query(user_query) # 重新解析交互流程：
用户输入: /plan 分析盈利
↓
┌────────────────┐
│ ① 解析 │ → LLM将短句解析为结构化Plan (JSON)
├────────────────┤
│ ② 展示 │ → Markdown表格展示数据源、计算逻辑、输出维度
├────────────────┤
│ ③ 确认 │ → y: 执行 | n: 重新生成 | 补充内容: 完善Plan
├────────────────┤
│ ④ 执行 │ → 生成SQL → DuckDB执行 → 返回结果
├────────────────┤
│ ⑤ 保存 │ → 保存到 output/plan_history.json
└────────────────┘Parquet 缓存机制（性能优化）
CW Agent 通过 Parquet 格式将 CSV 转换为列式存储，大幅提升性能：
| 指标 | CSV | Parquet | 改进 |
|---|---|---|---|
| 文件大小 | 578 MB | 30 MB | 95% 压缩 |
| 查询速度 | ~10 秒 | ~1 秒 | 10 倍提升 |
| 类型安全 | 需手动处理 | 自带类型信息 | 免维护 |
常见问题 & FAQ
| 问题 | 解决方案 |
|---|---|
| Ollama 连接失败 | 检查 ollama serve 是否运行；测试 curl http://localhost:11434/api/tags；确认 .env 中的 OPENAI_API_BASE 端口正确 |
| 加载 CSV 报错 | 使用 --encoding utf-8；检查文件是否被 Excel 锁定；确认路径中的反斜杠已改为正斜杠 |
| 列名包含括号报错 | DuckDB 中括号会被解析为函数调用，使用双引号包裹列名：SELECT "优惠后总价(不含税)" FROM data |
| LLM 返回无效代码 | 系统会自动回退到传统规则引擎，无需手动干预。可在 .env 中降低 TEMPERATURE=0 提高稳定性 |
| Excel + CSV 混合加载 | 项目自动检测文件格式（通过文件头魔数），无需手动指定。确保安装了 openpyxl |
| 内存不足 | 处理超大文件时，DuckDB 会自动使用磁盘交换。也可先将 CSV 转换为 Parquet |
| GUI 不显示中文 | 安装中文字体，或在环境变量中设置 LANG=zh_CN.UTF-8 |
| 打包后运行提示缺少依赖 | 在 app.spec 的 hiddenimports 中添加缺失的模块名 |
DuckDB 列名引号注意事项
-- ❌ 错误：括号被解析为函数调用
SELECT 优惠后总价(不含税) FROM data
-- ✅ 正确：使用双引号包裹
SELECT "优惠后总价(不含税)" FROM data
-- ✅ 推荐：使用 TRY_CAST 处理类型转换
SELECT TRY_CAST("优惠后总价(不含税)" AS DOUBLE) FROM data
技术架构总览
┌─────────────────────────────────────────────────────────────┐
│ Agent 工作流 │
├─────────────────────────────────────────────────────────────┤
│ 1. 初始化 │
│ └─ Schema探测 + 多文件加载 + 格式检测 │
│ ↓ │
│ 2. 查询处理 │
│ ├─ 普通模式: LLM代码生成 → 自动修复 → 执行 │
│ └─ Plan模式: 解析Plan → 确认 → 执行 │
│ ↓ │
│ 3. 错误自愈（Loop） │
│ └─ 失败 → 修复 → 重试 (最多3次) │
│ ↓ │
│ 4. Fallback机制 │
│ └─ LLM失败 → LLM输出无效 → 传统规则引擎 │
└─────────────────────────────────────────────────────────────┘技术栈
| 技术 | 用途 |
|---|---|
| Python | 主要开发语言 |
| DuckDB | 高效处理大型 CSV/Parquet 文件（内存中 OLAP 引擎） |
| Pandas | 数据处理和 Excel 读取 |
| LangChain | 语言模型集成框架（支持 OpenAI / Ollama） |
| Ollama | 本地 LLM 部署（deepseek-r1:8b） |
| PySimpleGUI | 跨平台桌面应用 UI |
| PyInstaller | Mac/Windows 应用打包 |
| openpyxl | Excel 文件读写 |
进阶：打包为桌面应用
Windows 打包
# 安装 PyInstaller
pip install pyinstaller
# 打包为 exe
pyinstaller --onefile --windowed --name "财务分析Agent" gui_app.pyMac 打包
# 使用已有的 app.spec 配置
pyinstaller app.spec --clean
# 打包产物在 dist/财务分析Agent.app
open dist/财务分析Agent.app参考资源
-
DuckDB 官方文档：https://duckdb.org/docs/
-
LangChain 官方文档：https://python.langchain.com/docs/
-
Ollama 官网：https://ollama.com/
-
PySimpleGUI 文档：https://docs.pysimplegui.com/
-
Scapy 官方文档：https://scapy.readthedocs.io/（用于姊妹项目 Cyber-Analyzer）
🎉 完成
按照本指南，你可以快速完成 财务分析 Agent 的部署与使用：
-
环境搭建 ✅
-
安装依赖 ✅
-
配置 Ollama ✅
-
运行命令行/GUI ✅
-
执行自然语言查询 ✅
-
Plan 模式交互 ✅
-
导出分析结果 ✅
该工具的核心价值在于：
-
大文件处理能力：500MB+ CSV 秒级加载
-
双模式保障：LLM 智能 + 传统规则双保险
-
数据安全：本地计算 + 脱敏 + 审计
-
用户友好：GUI 界面 + Plan 交互式分析
如果还有细节需要补充（如数据集的具体说明、自定义可视化图表、多数据库支持），欢迎在评论区交流。祝你玩得开心！ 🎓
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
14
0- 0
已为社区贡献3条内容
所有评论(0)