QAnything插件开发:扩展自定义文件解析器
QAnything插件开发:扩展自定义文件解析器
如果你用过QAnything,可能会发现它支持的文件格式已经挺多的了,PDF、Word、图片这些常见格式都能处理。但有时候,我们手头会有一些特殊格式的文件,比如CAD图纸、工业设计文件,或者公司内部特有的数据格式。这时候,原生的解析器就无能为力了。
好消息是,QAnything的架构设计得很灵活,它允许我们通过插件的方式,为这些特殊格式的文件开发自定义解析器。今天,我就来带你走一遍完整的开发流程,从理解接口设计,到实现数据处理,再到性能优化,最后用一个CAD文件解析的完整例子,让你彻底掌握这门手艺。
1. 为什么需要自定义解析器?
在深入代码之前,我们先搞清楚一个问题:为什么非得自己动手写解析器?
想象一下,你在一家建筑设计公司工作,每天要处理大量的CAD图纸。这些图纸里包含了建筑的结构信息、尺寸标注、材料说明等关键数据。如果能把它们导入QAnything的知识库,设计师就能直接问:“帮我找一下三楼平面图中所有承重墙的厚度”,系统就能从图纸里提取相关信息给出答案。
但CAD文件(比如.dwg格式)是二进制文件,结构复杂,QAnything默认的文本解析器根本看不懂。这就是自定义解析器的用武之地——它就像一个翻译官,把QAnything看不懂的特殊文件格式,转换成它能理解的文本或结构化数据。
除了CAD,还有很多场景需要自定义解析器:
- 工业领域:PLC程序文件、数控机床的G代码
- 科研领域:特定仪器的数据输出文件(如光谱仪、色谱仪)
- 金融领域:特定交易系统的日志格式
- 游戏开发:游戏资源文件、关卡配置文件
说白了,只要你的文件格式不是QAnything原生支持的,而你又想让它进入知识库被检索和问答,自定义解析器就是唯一的出路。
2. 理解QAnything的解析器架构
要开发插件,首先得知道系统是怎么工作的。QAnything的文件解析流程可以概括为三个步骤:
文件上传 → 格式识别 → 解析转换
当用户上传一个文件时,QAnything会先根据文件后缀判断类型,然后调用对应的解析器。解析器的核心任务是把原始文件内容转换成Document对象列表。每个Document包含两部分:
page_content:文件的文本内容metadata:文件的元信息,比如来源、页码等
QAnything内置的解析器都继承自LangChain的BaseLoader类,遵循统一的接口规范。我们开发自定义解析器,实际上就是实现一个符合这个规范的类。
让我用一个简单的类图帮你理解这个关系:
class BaseLoader:
"""所有解析器的基类"""
def load(self) -> List[Document]:
"""加载文件并返回Document列表"""
pass
def load_and_split(self, text_splitter) -> List[Document]:
"""加载文件并分割成chunk"""
pass
class UnstructuredFileLoader(BaseLoader):
"""QAnything内置的基础解析器"""
def __init__(self, file_path: str):
self.file_path = file_path
def _get_elements(self) -> List:
"""获取文件元素的具体实现"""
pass
class CustomCADLoader(UnstructuredFileLoader):
"""我们要开发的自定义CAD解析器"""
def _get_elements(self) -> List:
"""实现CAD文件的具体解析逻辑"""
# 这里就是我们要写代码的地方
pass
看到没?我们不需要从头造轮子,只需要继承现有的解析器基类,然后实现核心的解析方法就行了。这种设计让插件开发变得简单而规范。
3. 开发环境准备
在动手写代码之前,我们需要准备好开发环境。QAnything推荐使用Docker部署,但对于插件开发,我们更关心的是代码层面的依赖。
首先,确保你的Python环境是3.8或更高版本。然后安装必要的依赖:
# 创建虚拟环境(推荐)
python -m venv qanything_plugin_env
source qanything_plugin_env/bin/activate # Linux/Mac
# 或者 qanything_plugin_env\Scripts\activate # Windows
# 安装核心依赖
pip install langchain==0.0.340
pip install unstructured==0.10.30
# 安装开发工具
pip install black isort flake8 # 代码格式化工具
对于CAD文件解析,我们还需要专门的CAD处理库。这里我推荐使用ezdxf,它是一个纯Python的DXF文件处理库,轻量且功能强大:
pip install ezdxf
如果你需要处理其他CAD格式(比如.dwg),可能需要额外的库。.dwg是AutoCAD的专有格式,处理起来比较复杂,通常需要依赖pyautocad或python-dwg,但这些库的安装和配置比较麻烦。为了教程的简洁性,我们以.dxf格式为例,这是CAD领域通用的交换格式,大多数CAD软件都支持导出为DXF。
4. 实现CAD文件解析器
现在进入最核心的部分——编写CAD解析器。我们要创建一个CADFileLoader类,它能够读取DXF文件,提取其中的文本信息(如图层名、标注文字、块属性等),并转换成QAnything能处理的Document格式。
4.1 创建解析器类
首先,在QAnything的代码结构中找到合适的位置。我建议在qanything_kernel/core/目录下创建一个新的文件cad_loader.py:
"""
CAD文件解析器插件
支持DXF格式的CAD文件解析
"""
import os
import tempfile
from typing import List, Dict, Any
import logging
from langchain.schema import Document
from qanything_kernel.core.local_file import UnstructuredFileLoader
# 尝试导入ezdxf,如果失败则给出友好提示
try:
import ezdxf
EZDXF_AVAILABLE = True
except ImportError:
EZDXF_AVAILABLE = False
logging.warning("ezdxf库未安装,CAD解析功能将不可用。请运行: pip install ezdxf")
logger = logging.getLogger(__name__)
class CADFileLoader(UnstructuredFileLoader):
"""CAD文件解析器,支持DXF格式"""
def __init__(self, file_path: str, **kwargs):
"""
初始化CAD文件解析器
Args:
file_path: CAD文件路径
**kwargs: 其他参数,传递给父类
"""
super().__init__(file_path, **kwargs)
# 检查文件格式
if not file_path.lower().endswith('.dxf'):
raise ValueError(f"不支持的文件格式: {file_path}。目前仅支持.dxf格式")
# 检查依赖库
if not EZDXF_AVAILABLE:
raise ImportError("ezdxf库未安装,无法解析CAD文件。请安装: pip install ezdxf")
def _get_elements(self) -> List[Dict[str, Any]]:
"""
解析CAD文件的核心方法
返回:
包含解析结果的元素列表
"""
logger.info(f"开始解析CAD文件: {self.file_path}")
try:
# 使用ezdxf加载DXF文件
doc = ezdxf.readfile(self.file_path)
# 提取CAD文件中的文本信息
extracted_text = self._extract_cad_text(doc)
# 将提取的文本转换为unstructured库需要的格式
elements = self._text_to_elements(extracted_text)
logger.info(f"CAD文件解析完成,提取到{len(elements)}个文本元素")
return elements
except ezdxf.DXFStructureError as e:
logger.error(f"DXF文件结构错误: {e}")
raise ValueError(f"无效的DXF文件: {e}")
except Exception as e:
logger.error(f"解析CAD文件时出错: {e}")
raise RuntimeError(f"解析CAD文件失败: {e}")
def _extract_cad_text(self, doc) -> str:
"""
从DXF文档中提取所有文本内容
Args:
doc: ezdxf文档对象
Returns:
提取的文本内容
"""
text_parts = []
# 1. 提取模型空间中的文本实体
modelspace = doc.modelspace()
# 提取TEXT实体(单行文本)
for text in modelspace.query('TEXT'):
text_parts.append(f"文本: {text.dxf.text}")
# 提取MTEXT实体(多行文本)
for mtext in modelspace.query('MTEXT'):
# MTEXT可能包含格式代码,我们只提取纯文本
plain_text = mtext.plain_text()
if plain_text:
text_parts.append(f"多行文本: {plain_text}")
# 2. 提取标注和尺寸
for dim in modelspace.query('*[dimension]'):
# 这里简化处理,实际中可能需要提取更多标注信息
if hasattr(dim, 'dxf') and hasattr(dim.dxf, 'text'):
text_parts.append(f"标注: {dim.dxf.text}")
# 3. 提取图层信息
text_parts.append("\n=== 图层信息 ===")
for layer in doc.layers:
text_parts.append(f"图层: {layer.dxf.name}")
# 4. 提取块定义信息
text_parts.append("\n=== 块定义 ===")
for block in doc.blocks:
# 只记录非标准块
if not block.name.startswith('*'):
text_parts.append(f"块: {block.name}")
# 提取块中的属性定义
for entity in block:
if entity.dxftype() == 'ATTDEF':
text_parts.append(f" 属性: {entity.dxf.tag} = {entity.dxf.default}")
# 5. 提取文件元信息
text_parts.append("\n=== 文件信息 ===")
header = doc.header
text_parts.append(f"创建时间: {header.get('$TDCREATE', '未知')}")
text_parts.append(f"最后修改: {header.get('$TDUPDATE', '未知')}")
text_parts.append(f"作者: {header.get('$AUTHOR', '未知')}")
# 将所有文本部分合并
full_text = "\n".join(text_parts)
logger.debug(f"从CAD文件中提取的文本长度: {len(full_text)}字符")
return full_text
def _text_to_elements(self, text: str) -> List[Dict[str, Any]]:
"""
将提取的文本转换为unstructured库需要的元素格式
Args:
text: 提取的文本内容
Returns:
格式化后的元素列表
"""
if not text.strip():
return []
# 这里我们创建一个简单的元素结构
# 在实际应用中,你可能需要根据文本内容创建更复杂的结构
element = {
"type": "NarrativeText", # 元素类型
"text": text, # 文本内容
"metadata": {
"source": self.file_path,
"file_type": "CAD/DXF",
"category_depth": 0,
}
}
return [element]
这个解析器做了几件关键的事情:
- 检查依赖:确保
ezdxf库已安装 - 验证格式:只处理
.dxf文件 - 提取文本:从DXF文件中提取各种文本信息
- 结构化输出:将提取的文本转换成标准格式
4.2 集成到QAnything系统
有了解析器类,下一步是让它能被QAnything系统识别和调用。我们需要修改文件加载的逻辑。
找到qanything_kernel/core/local_file.py文件中的split_file_to_docs方法,添加对CAD文件的支持:
# 在split_file_to_docs方法中添加以下代码
elif self.file_path.lower().endswith(".dxf"):
# 导入我们的CAD解析器
from .cad_loader import CADFileLoader
# 创建CAD解析器实例
loader = CADFileLoader(self.file_path)
# 使用中文文本分割器
texts_splitter = ChineseTextSplitter(pdf=False, sentence_size=sentence_size)
# 加载并分割文档
docs = loader.load_and_split(texts_splitter)
这样,当用户上传.dxf文件时,QAnything就会自动使用我们的CAD解析器来处理。
5. 处理复杂CAD结构
上面的基础版本能处理大多数文本信息,但CAD文件中的图形信息(如线条、圆弧、多边形)也很重要。有时候用户可能会问:“这张图里有多少个圆形?”或者“最长的直线有多长?”
为了支持这类查询,我们需要增强解析器,让它能提取几何信息。下面是一个增强版的几何信息提取方法:
def _extract_geometry_info(self, doc) -> str:
"""
提取CAD文件中的几何信息
Args:
doc: ezdxf文档对象
Returns:
几何信息描述文本
"""
geometry_parts = []
modelspace = doc.modelspace()
# 统计各种图形元素
line_count = 0
circle_count = 0
arc_count = 0
polyline_count = 0
# 收集图形详细信息
lines_info = []
circles_info = []
for entity in modelspace:
entity_type = entity.dxftype()
if entity_type == 'LINE':
line_count += 1
# 计算线段长度
start = entity.dxf.start
end = entity.dxf.end
length = ((end.x - start.x)**2 + (end.y - start.y)**2)**0.5
lines_info.append(f"线段{line_count}: 长度={length:.2f}, 起点({start.x:.1f},{start.y:.1f})")
elif entity_type == 'CIRCLE':
circle_count += 1
center = entity.dxf.center
radius = entity.dxf.radius
circles_info.append(f"圆形{circle_count}: 半径={radius:.2f}, 中心({center.x:.1f},{center.y:.1f})")
elif entity_type == 'ARC':
arc_count += 1
elif entity_type in ('LWPOLYLINE', 'POLYLINE'):
polyline_count += 1
# 组织几何信息
geometry_parts.append("=== 几何元素统计 ===")
geometry_parts.append(f"线段: {line_count}个")
geometry_parts.append(f"圆形: {circle_count}个")
geometry_parts.append(f"圆弧: {arc_count}个")
geometry_parts.append(f"多段线: {polyline_count}个")
# 添加详细信息(限制数量,避免文本过长)
if lines_info:
geometry_parts.append("\n--- 线段详情(前5个)---")
for info in lines_info[:5]:
geometry_parts.append(info)
if len(lines_info) > 5:
geometry_parts.append(f"... 还有{len(lines_info)-5}个线段")
if circles_info:
geometry_parts.append("\n--- 圆形详情(前5个)---")
for info in circles_info[:5]:
geometry_parts.append(info)
if len(circles_info) > 5:
geometry_parts.append(f"... 还有{len(circles_info)-5}个圆形")
return "\n".join(geometry_parts)
然后在_extract_cad_text方法中调用这个函数:
def _extract_cad_text(self, doc) -> str:
text_parts = []
# ... 之前的文本提取代码 ...
# 6. 提取几何信息
geometry_text = self._extract_geometry_info(doc)
if geometry_text:
text_parts.append("\n=== 几何信息 ===")
text_parts.append(geometry_text)
# ... 合并文本的代码 ...
现在,当用户上传CAD文件时,解析器不仅会提取文本信息,还会统计图形元素,这样用户就能问出更具体的问题,比如“这个图纸里有多少条线段?”或者“最大的圆形半径是多少?”
6. 性能优化技巧
CAD文件可能很大,包含成千上万个图形元素。如果全部解析,可能会消耗大量内存和时间。下面是一些优化建议:
6.1 增量解析
对于大型CAD文件,我们可以采用增量解析的方式,只提取最关键的信息:
def _extract_cad_text_optimized(self, doc) -> str:
"""优化版的CAD文本提取,处理大型文件"""
text_parts = []
# 策略1:优先提取文本实体(通常最重要)
modelspace = doc.modelspace()
# 限制提取的数量
max_text_entities = 1000
text_count = 0
for text in modelspace.query('TEXT'):
if text_count >= max_text_entities:
text_parts.append(f"... 还有更多文本实体未显示")
break
text_parts.append(f"文本: {text.dxf.text}")
text_count += 1
# 策略2:抽样提取几何信息
# 对于大型文件,只统计数量,不记录每个实体的详细信息
total_entities = len(modelspace)
if total_entities > 5000:
text_parts.append(f"\n注:文件包含{total_entities}个图形实体,为提升性能仅显示统计信息")
# 快速统计
from collections import Counter
entity_types = Counter(entity.dxftype() for entity in modelspace)
text_parts.append("\n=== 图形实体统计 ===")
for etype, count in entity_types.most_common(10): # 只显示最多的10种类型
text_parts.append(f"{etype}: {count}个")
return "\n".join(text_parts)
6.2 缓存机制
如果同一个CAD文件可能被多次处理,我们可以添加缓存:
import hashlib
import pickle
from pathlib import Path
class CADFileLoaderWithCache(CADFileLoader):
"""带缓存的CAD解析器"""
def __init__(self, file_path: str, cache_dir: str = "./cache", **kwargs):
super().__init__(file_path, **kwargs)
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(exist_ok=True)
def _get_elements(self) -> List[Dict[str, Any]]:
# 生成缓存键(基于文件内容和修改时间)
cache_key = self._generate_cache_key()
cache_file = self.cache_dir / f"{cache_key}.pkl"
# 检查缓存
if cache_file.exists():
logger.info(f"从缓存加载解析结果: {cache_file}")
with open(cache_file, 'rb') as f:
return pickle.load(f)
# 执行解析
elements = super()._get_elements()
# 保存到缓存
with open(cache_file, 'wb') as f:
pickle.dump(elements, f)
logger.info(f"解析结果已缓存: {cache_file}")
return elements
def _generate_cache_key(self) -> str:
"""生成缓存键"""
file_stat = os.stat(self.file_path)
# 使用文件大小、修改时间和前1KB内容生成哈希
with open(self.file_path, 'rb') as f:
file_head = f.read(1024)
hash_input = f"{file_stat.st_size}_{file_stat.st_mtime}_{file_head}"
return hashlib.md5(hash_input.encode()).hexdigest()
6.3 并行处理
对于特别大的CAD文件,我们可以考虑并行处理不同的部分:
import concurrent.futures
def _extract_cad_text_parallel(self, doc) -> str:
"""并行提取CAD文本"""
text_parts = []
# 定义提取任务
extraction_tasks = [
("图层信息", self._extract_layers, doc),
("文本实体", self._extract_text_entities, doc),
("块定义", self._extract_blocks, doc),
("几何统计", self._extract_geometry_stats, doc),
]
# 使用线程池并行执行
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
# 提交任务
future_to_task = {
executor.submit(func, doc): name
for name, func, _ in extraction_tasks
}
# 收集结果
for future in concurrent.futures.as_completed(future_to_task):
task_name = future_to_task[future]
try:
result = future.result()
if result:
text_parts.append(f"\n=== {task_name} ===")
text_parts.append(result)
except Exception as e:
logger.warning(f"提取{task_name}时出错: {e}")
text_parts.append(f"\n=== {task_name} ===")
text_parts.append(f"(提取失败: {e})")
return "\n".join(text_parts)
7. 测试你的解析器
开发完成后,一定要进行测试。创建一个简单的测试脚本:
#!/usr/bin/env python3
"""
测试CAD文件解析器
"""
import sys
import os
# 添加QAnything模块路径
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from qanything_kernel.core.cad_loader import CADFileLoader
def test_cad_loader():
"""测试CAD解析器"""
# 使用示例DXF文件(你需要准备一个测试文件)
test_file = "sample.dxf"
if not os.path.exists(test_file):
print(f"测试文件不存在: {test_file}")
print("请创建一个简单的DXF文件用于测试")
return
print(f"测试文件: {test_file}")
print("文件大小:", os.path.getsize(test_file), "字节")
try:
# 创建解析器
loader = CADFileLoader(test_file)
# 测试加载
print("\n正在解析CAD文件...")
documents = loader.load()
print(f"\n解析成功!生成{len(documents)}个文档片段")
# 显示前3个片段的内容预览
for i, doc in enumerate(documents[:3]):
print(f"\n--- 片段 {i+1} ---")
content_preview = doc.page_content[:200] + "..." if len(doc.page_content) > 200 else doc.page_content
print(f"内容预览: {content_preview}")
print(f"元数据: {doc.metadata}")
if len(documents) > 3:
print(f"\n... 还有{len(documents)-3}个片段未显示")
except Exception as e:
print(f"\n解析失败: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
test_cad_loader()
运行这个测试脚本,确保你的解析器能正常工作。如果遇到问题,检查以下几点:
- 依赖库是否正确安装
- DXF文件格式是否有效
- 文件路径是否正确
- 是否有足够的读取权限
8. 打包和分发插件
如果你开发的解析器很有用,可能会想分享给其他人。这里有一些打包和分发的建议:
8.1 创建独立的插件包
# setup.py
from setuptools import setup, find_packages
setup(
name="qanything-cad-plugin",
version="1.0.0",
description="QAnything CAD文件解析插件",
author="你的名字",
packages=find_packages(),
install_requires=[
"ezdxf>=1.1.0",
"langchain>=0.0.340",
],
entry_points={
"qanything.plugins": [
"cad_loader = qanything_cad_plugin.cad_loader:CADFileLoader",
],
},
classifiers=[
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
],
)
8.2 提供安装和使用文档
创建一个README.md文件:
# QAnything CAD解析插件
这个插件为QAnything添加了对CAD文件(DXF格式)的支持。
## 安装
```bash
pip install qanything-cad-plugin
使用
安装后,QAnything会自动识别.dxf文件并使用本插件进行解析。
支持的功能
- 提取文本实体(TEXT, MTEXT)
- 提取图层信息
- 提取块定义和属性
- 统计几何元素(线段、圆形、圆弧等)
- 提取文件元信息
开发
要修改或扩展插件,请克隆仓库并安装开发依赖:
git clone https://github.com/yourname/qanything-cad-plugin.git
cd qanything-cad-plugin
pip install -e ".[dev]"
许可证
MIT
## 9. 扩展思路:支持更多格式
掌握了CAD文件解析后,你可以用同样的方法为其他格式开发解析器。比如:
### 9.1 3D模型文件(STL, OBJ)
```python
class STLLoader(UnstructuredFileLoader):
"""STL 3D模型文件解析器"""
def _get_elements(self) -> List:
import numpy as np
with open(self.file_path, 'rb') as f:
# 读取STL文件头
header = f.read(80)
# 解析三角形面片数量
num_triangles = np.fromfile(f, dtype=np.uint32, count=1)[0]
text_parts = [
f"STL 3D模型文件",
f"面片数量: {num_triangles}",
f"文件大小: {os.path.getsize(self.file_path)} 字节",
]
# 可以进一步解析法向量、顶点坐标等
return [{"type": "NarrativeText", "text": "\n".join(text_parts)}]
9.2 地理空间数据(Shapefile, KML)
class ShapefileLoader(UnstructuredFileLoader):
"""Shapefile地理数据解析器"""
def _get_elements(self) -> List:
try:
import geopandas as gpd
# 读取shapefile
gdf = gpd.read_file(self.file_path)
text_parts = [
f"Shapefile地理数据",
f"要素数量: {len(gdf)}",
f"几何类型: {gdf.geometry.type.iloc[0] if len(gdf) > 0 else '未知'}",
f"坐标系统: {gdf.crs}",
]
# 提取属性表信息
if len(gdf) > 0:
text_parts.append("\n属性字段:")
for col in gdf.columns:
if col != 'geometry':
sample = str(gdf[col].iloc[0])[:50]
text_parts.append(f" {col}: {sample}...")
return [{"type": "NarrativeText", "text": "\n".join(text_parts)}]
except ImportError:
return [{"type": "NarrativeText", "text": "需要geopandas库来解析shapefile"}]
总结
开发QAnything的自定义文件解析器,本质上就是理解系统的扩展接口,然后针对特定文件格式实现解析逻辑。整个过程可以概括为:理解需求 → 设计接口 → 实现解析 → 优化性能 → 测试验证 → 打包分发。
CAD文件解析的例子展示了如何处理二进制、结构复杂的专业格式。同样的思路可以应用到任何其他文件格式上。关键是要深入理解目标格式的结构,提取出对问答最有价值的信息,同时注意性能优化,确保在大文件场景下也能稳定工作。
实际开发中,你可能会遇到各种挑战,比如文件格式的版本差异、损坏文件的容错处理、特殊字符的编码问题等。这时候,良好的日志记录和错误处理就特别重要。记得在代码中添加详细的日志,方便调试和问题排查。
最后,不要忘记文档。好的插件不仅要有好的代码,还要有清晰的文档,告诉用户如何安装、如何使用、有哪些限制。这样你的插件才能真正帮助到更多人。
获取更多AI镜像
想探索更多AI镜像和应用场景?访问 CSDN星图镜像广场,提供丰富的预置镜像,覆盖大模型推理、图像生成、视频生成、模型微调等多个领域,支持一键部署。
更多推荐

所有评论(0)