企业级文档自动化处理系统架构与实现

1. 企业级文档自动化处理系统架构解析

在当今数字化办公环境中，企业每天需要处理大量合同、财报、标书等专业文档。传统人工处理方式效率低下且容易出错，而智能文档处理系统能够将非结构化文档转化为结构化数据，实现自动化分析和处理。这套系统采用四层架构设计，每层都有明确的职责和技术实现方案。

1.1 文档接入层设计要点

文档接入层是整个系统的入口，负责接收和处理各种格式的输入文档。在实际部署中，我们通常会遇到以下几种文档类型：

• PDF文档：包括文本型PDF和扫描件PDF
• Word文档：主要是.docx格式
• 扫描件图像：JPG/PNG等格式的扫描文档
• 邮件附件：从企业邮箱系统自动抓取的文档

针对不同文档类型，我们需要采用不同的预处理策略：

def document_preprocessor(file_path): """文档预处理分发函数""" file_ext = os.path.splitext(file_path)[1].lower() if file_ext == '.pdf': # 先判断是文本PDF还是扫描件 if is_scanned_pdf(file_path): return process_scanned_pdf(file_path) else: return process_text_pdf(file_path) elif file_ext in ('.docx', '.doc'): return process_word_document(file_path) elif file_ext in ('.jpg', '.jpeg', '.png'): return process_image_document(file_path) else: raise ValueError(f"不支持的文档格式: {file_ext}")

实际应用中，建议对扫描件进行质量检测（清晰度、倾斜度等），质量过低的文档应触发人工复核流程。

1.2 解析引擎层核心技术

解析引擎层是整个系统的核心，负责将原始文档转换为结构化数据。这一层主要解决三个技术难题：

1. 高精度OCR识别：对于扫描件，我们采用Tesseract OCR引擎配合自定义训练的中文模型，针对财务表格等特殊内容进行了优化训练。

2. 复杂版面分析：使用基于深度学习的版面分析算法，能够准确识别文档中的标题、段落、表格、页眉页脚等元素。以下是版面分析的代码示例：

from layoutparser import Layout, LayoutParser def analyze_document_layout(image): """文档版面分析""" model = LayoutParser('lp://PrimaLayout/mask_rcnn_R_50_FPN_3x/config') layout = model.detect(image) # 对识别到的区块进行分类和排序 text_blocks = [b for b in layout if b.type == 'Text'] table_blocks = [b for b in layout if b.type == 'Table'] figure_blocks = [b for b in layout if b.type == 'Figure'] return { 'text_blocks': sort_blocks(text_blocks), 'table_blocks': table_blocks, 'figure_blocks': figure_blocks }

3. 表格还原技术：采用基于OpenCV的图像处理算法结合深度学习模型，实现复杂表格结构的准确还原，保持表格行列关系不丢失。

1.3 智能分析层实现方案

智能分析层负责对解析后的内容进行深度理解和分析，主要包含以下几个功能模块：

• 实体识别(NER)：专门针对合同、财报等领域的自定义实体识别模型
• 条款分类：基于规则和机器学习结合的条款自动分类
• 语义检索：利用向量数据库实现的相似内容检索
• 风险预警：通过模式匹配和LLM分析识别潜在风险

实体识别模块的典型实现：

from transformers import AutoTokenizer, AutoModelForTokenClassification class FinancialNER: def __init__(self, model_path='./models/financial-ner'): self.tokenizer = AutoTokenizer.from_pretrained(model_path) self.model = AutoModelForTokenClassification.from_pretrained(model_path) self.label_map = { 0: 'O', 1: 'B-COMPANY', 2: 'I-COMPANY', 3: 'B-AMOUNT', 4: 'I-AMOUNT', # ...其他标签 } def extract_entities(self, text): """提取金融实体""" inputs = self.tokenizer(text, return_tensors="pt", truncation=True) outputs = self.model(**inputs) predictions = outputs.logits.argmax(dim=-1)[0].tolist() entities = [] current_entity = None for token, pred in zip(inputs.tokens(), predictions): label = self.label_map[pred] if label.startswith('B-'): if current_entity: entities.append(current_entity) current_entity = { 'type': label[2:], 'text': token } elif label.startswith('I-') and current_entity: current_entity['text'] += ' ' + token else: if current_entity: entities.append(current_entity) current_entity = None return entities

1.4 应用输出层设计

应用输出层将分析结果以多种形式提供给最终用户：

1. 结构化报表：将提取的关键信息整理为Excel或JSON格式
2. 风险预警报告：高亮显示文档中的风险条款
3. 智能问答接口：允许用户通过自然语言查询文档内容
4. 可视化看板：对财务数据等进行可视化展示

典型的API接口设计示例：

from fastapi import FastAPI, UploadFile from pydantic import BaseModel app = FastAPI() class AnalysisResult(BaseModel): entities: list tables: list risks: list summary: str @app.post("/analyze") async def analyze_document(file: UploadFile) -> AnalysisResult: """文档分析接口""" content = await file.read() parser = DocumentParser() analyzer = DocumentAnalyzer() # 解析文档 parsed = parser.parse(content) # 分析内容 result = analyzer.analyze(parsed) return AnalysisResult( entities=result['entities'], tables=result['tables'], risks=result['risks'], summary=result['summary'] )

2. 系统环境配置与部署

2.1 硬件配置建议

根据实际业务规模，我们建议以下硬件配置方案：

对于扫描件处理量大的场景，建议配置专用GPU服务器进行OCR加速。纯文本处理场景对GPU需求较低。

2.2 软件环境搭建

系统基于Ubuntu 22.04 LTS和Python 3.10构建，以下是详细的安装步骤：

# 1. 系统更新与基础工具安装 sudo apt-get update && sudo apt-get upgrade -y sudo apt-get install -y python3.10 python3.10-venv python3.10-dev \ git wget curl build-essential tesseract-ocr tesseract-ocr-chi-sim \ poppler-utils libmagic-dev libopencv-dev # 2. 创建Python虚拟环境 mkdir -p ~/docai && cd ~/docai python3.10 -m venv venv source venv/bin/activate # 3. 安装PyTorch（根据CUDA版本选择） pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 # 4. 安装核心依赖 pip install -r requirements.txt

requirements.txt应包含以下关键包：

# 文档处理 pdfplumber>=0.11.0 pymupdf>=1.24.0 python-docx>=1.1.0 marker-pdf>=0.3.0 layoutparser>=0.3.4 # 机器学习 transformers>=4.40.0 sentence-transformers>=3.0.0 langchain>=0.3.0 chromadb>=0.5.0 # 后端服务 fastapi>=0.115.0 uvicorn>=0.30.0 celery>=5.3.6 redis>=5.0.1

2.3 模型下载与配置

系统依赖多个预训练模型，建议提前下载：

# 中文OCR模型 wget https://github.com/tesseract-ocr/tessdata/raw/main/chi_sim.traineddata -P /usr/share/tesseract-ocr/4.00/tessdata/ # 版面分析模型 python -c "from layoutparser import models; model = models.AutoLayoutModel('lp://PrimaLayout/mask_rcnn_R_50_FPN_3x/config'); model.save('./models/layout')" # 中文文本嵌入模型 python -c "from sentence_transformers import SentenceTransformer; model = SentenceTransformer('BAAI/bge-large-zh-v1.5'); model.save('./models/bge-large-zh')"

2.4 Docker部署方案

对于生产环境，推荐使用Docker Compose部署：

version: '3.8' services: api: image: docai-api:latest ports: - "8000:8000" volumes: - ./models:/app/models - ./storage:/app/storage environment: - OCR_MODEL_PATH=/app/models/ocr - EMBEDDING_MODEL_PATH=/app/models/bge-large-zh deploy: resources: reservations: devices: - driver: nvidia count: 1 capabilities: [gpu] chroma: image: chromadb/chroma:latest ports: - "8001:8000" volumes: - ./chroma-data:/chroma/chroma worker: image: docai-worker:latest command: celery -A tasks worker --loglevel=info environment: - BROKER_URL=redis://redis:6379/0 depends_on: - redis redis: image: redis:7-alpine ports: - "6379:6379"

3. 合同智能解析模块实现

3.1 合同解析技术选型

经过对比测试，我们选择了MinerU作为核心解析引擎，主要基于以下考虑：

1. 开源可定制：相比商业解决方案，MinerU允许我们针对合同特点进行定制优化
2. 中文支持好：专门针对中文文档优化了版面分析和OCR识别
3. 表格处理强：能够准确还原复杂合同中的表格结构
4. 性能平衡：在准确率和速度之间取得了良好平衡

安装MinerU的完整步骤：

# 1. 安装基础依赖 sudo apt-get install -y libgl1-mesa-glx libglib2.0-0 # 2. 安装MinerU pip install mineru>=0.9.0 # 3. 下载模型权重 git clone https://github.com/opendatalab/MinerU.git cd MinerU pip install -e . # 4. 初始化模型（首次运行会自动下载） magic-pdf --help

3.2 合同关键信息提取

合同解析的核心是从非结构化文本中提取结构化信息，主要包括：

1. 合同主体信息：甲方、乙方名称和基本信息
2. 关键条款：付款方式、交付条件、违约责任等
3. 金额与日期：合同金额、付款日期、生效日期等
4. 特殊条款：保密协议、知识产权等

实现代码示例：

import re from datetime import datetime from typing import Dict, List class ContractParser: """合同解析核心类""" def __init__(self): self.party_patterns = [ r'(甲方|发包方|委托方)[：:]\s*([^\n]+)', r'(乙方|承包方|受托方)[：:]\s*([^\n]+)' ] self.amount_pattern = r'(?:人民币|¥|￥)\s*([\d,]+(?:\.\d{1,2})?)\s*(?:元|万元)' self.date_pattern = r'(\d{4}年\d{1,2}月\d{1,2}日|\d{4}-\d{2}-\d{2})' def extract_parties(self, text: str) -> Dict: """提取合同主体信息""" parties = {} for pattern in self.party_patterns: matches = re.finditer(pattern, text) for match in matches: role = match.group(1) name = match.group(2).strip() parties[role] = name return parties def extract_payment_terms(self, text: str) -> List[Dict]: """提取付款条款""" payment_blocks = self._find_payment_blocks(text) terms = [] for block in payment_blocks: amount_match = re.search(self.amount_pattern, block) date_match = re.search(self.date_pattern, block) term = { 'amount': float(amount_match.group(1).replace(',', '')) if amount_match else None, 'date': self._parse_chinese_date(date_match.group(1)) if date_match else None, 'description': block[:100] + '...' if len(block) > 100 else block } terms.append(term) return terms def _find_payment_blocks(self, text: str) -> List[str]: """定位付款条款段落""" payment_keywords = ['付款', '支付', '结算', '定金', '尾款'] sentences = re.split(r'[。；\n]', text) return [s.strip() for s in sentences if any(kw in s for kw in payment_keywords)] def _parse_chinese_date(self, date_str: str) -> str: """解析中文日期格式""" if '年' in date_str: year, month, day = re.findall(r'\d+', date_str) return f"{year}-{month.zfill(2)}-{day.zfill(2)}" return date_str

3.3 合同风险分析

合同风险分析结合规则匹配和LLM智能分析：

class RiskAnalyzer: """合同风险分析器""" def __init__(self, llm_client=None): self.llm = llm_client self.risk_patterns = { 'unlimited_liability': r'无限责任|连带责任', 'unilateral_termination': r'单方解除|单方终止', 'auto_renewal': r'自动续约|自动延期', 'exclusive_license': r'独家授权|排他许可' } def analyze_contract(self, text: str) -> Dict: """分析合同风险""" # 基于规则的风险检测 rule_based_risks = self._rule_based_analysis(text) # LLM深度分析 llm_risks = [] if self.llm and len(text) > 1000: llm_risks = self._llm_analysis(text[:3000]) # 限制输入长度 return { 'rule_based': rule_based_risks, 'llm_based': llm_risks, 'total_risks': len(rule_based_risks) + len(llm_risks) } def _rule_based_analysis(self, text: str) -> List[Dict]: """基于规则的风险检测""" risks = [] for risk_type, pattern in self.risk_patterns.items(): matches = re.finditer(pattern, text) for match in matches: risks.append({ 'type': risk_type, 'text': match.group(0), 'position': match.start(), 'severity': 'high' if risk_type in ('unlimited_liability', 'exclusive_license') else 'medium' }) return risks def _llm_analysis(self, text: str) -> List[Dict]: """使用LLM进行深度风险分析""" prompt = f"""作为专业法务人员，请分析以下合同条款的潜在风险： {text[:3000]} 请识别： 1. 对甲方不利的条款 2. 对乙方不利的条款 3. 模糊不清的表述 4. 缺失的关键条款 按以下JSON格式输出： {{ "risks": [ {{ "type": "risk_type", "description": "风险描述", "severity": "high/medium/low", "suggestion": "修改建议" }} ] }}""" try: response = self.llm.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": prompt}], temperature=0.1 ) return json.loads(response.choices[0].message.content).get('risks', []) except Exception as e: print(f"LLM分析失败: {e}") return []

3.4 合同解析API服务

基于FastAPI提供合同解析服务：

from fastapi import FastAPI, UploadFile, HTTPException from fastapi.responses import JSONResponse import tempfile import os app = FastAPI(title="合同解析API") @app.post("/parse-contract") async def parse_contract(file: UploadFile): """合同解析接口""" try: # 保存临时文件 suffix = os.path.splitext(file.filename)[1] with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp: content = await file.read() tmp.write(content) tmp_path = tmp.name # 解析合同 parser = ContractParser() analyzer = RiskAnalyzer() # 提取文本内容 text = extract_text_from_file(tmp_path) # 实现文件内容提取函数 # 提取结构化信息 result = { 'parties': parser.extract_parties(text), 'payment_terms': parser.extract_payment_terms(text), 'dates': parser.extract_dates(text), 'risks': analyzer.analyze_contract(text), 'metadata': { 'filename': file.filename, 'filesize': len(content), 'parse_time': datetime.now().isoformat() } } return JSONResponse(content=result) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) finally: if tmp_path: os.unlink(tmp_path)

4. 财报智能分析模块实现

4.1 财报解析技术方案

财报解析面临两大核心挑战：

1. 表格结构还原：财报中的复杂表格需要准确转换为结构化数据
2. 财务指标计算：从文本描述中提取关键财务指标

我们采用TextIn商业服务结合开源工具的方案：

class FinancialReportParser: """财报解析器""" def __init__(self, use_textin=True): self.use_textin = use_textin if not use_textin: self.local_parser = LocalPDFParser() def parse(self, pdf_path: str) -> Dict: """解析财报PDF""" if self.use_textin: return self._parse_with_textin(pdf_path) else: return self._parse_locally(pdf_path) def _parse_with_textin(self, pdf_path: str) -> Dict: """使用TextIn服务解析""" import base64 import requests with open(pdf_path, "rb") as f: pdf_base64 = base64.b64encode(f.read()).decode('utf-8') headers = { "x-ti-app-id": os.getenv("TEXTIN_APP_ID"), "x-ti-secret-code": os.getenv("TEXTIN_SECRET_CODE"), "Content-Type": "application/json" } payload = { "file_base64": pdf_base64, "page_start": 0, "page_count": 100, "table_flavor": "markdown", "parse_mode": "scan" } response = requests.post( "https://api.textin.com/ai/service/v1/pdf_to_markdown", headers=headers, json=payload, timeout=120 ) if response.status_code == 200: return self._process_textin_result(response.json()) else: raise Exception(f"TextIn解析失败: {response.text}") def _parse_locally(self, pdf_path: str) -> Dict: """本地解析方案""" text = self.local_parser.extract_text(pdf_path) tables = self.local_parser.extract_tables(pdf_path) return { 'text': text, 'tables': tables, 'metrics': self._extract_metrics(text, tables) } def _process_textin_result(self, result: Dict) -> Dict: """处理TextIn返回结果""" markdown = result.get('result', {}).get('markdown', '') tables = self._extract_markdown_tables(markdown) return { 'text': markdown, 'tables': tables, 'metrics': self._extract_metrics(markdown, tables) }

4.2 财务指标提取

从财报文本和表格中提取关键财务指标：

class FinancialMetricsExtractor: """财务指标提取器""" def __init__(self): self.metric_patterns = { 'revenue': [ r'营业收入[：:]\s*([\d,]+(?:\.\d+)?)\s*亿元?', r'营收[：:]\s*([\d,]+(?:\.\d+)?)\s*亿元?' ], 'net_profit': [ r'归母净利润[：:]\s*([\d,]+(?:\.\d+)?)\s*亿元?', r'净利润[：:]\s*([\d,]+(?:\.\d+)?)\s*亿元?' ], 'gross_margin': [ r'毛利率[：:]\s*([\d,]+(?:\.\d+)?)%', r'综合毛利率[：:]\s*([\d,]+(?:\.\d+)?)%' ] } def extract_metrics(self, text: str, tables: List[pd.DataFrame]) -> Dict: """提取财务指标""" metrics = {} # 从文本中提取 for metric, patterns in self.metric_patterns.items(): for pattern in patterns: match = re.search(pattern, text) if match: value = match.group(1).replace(',', '') try: metrics[metric] = float(value) break except ValueError: continue # 从表格中补充提取 for table in tables: if '营业收入' in table.columns: metrics['revenue'] = table['营业收入'].iloc[-1] if '净利润' in table.columns: metrics['net_profit'] = table['净利润'].iloc[-1] return metrics def calculate_ratios(self, metrics: Dict) -> Dict: """计算财务比率""" ratios = {} if 'revenue' in metrics and 'net_profit' in metrics: ratios['net_margin'] = metrics['net_profit'] / metrics['revenue'] if 'total_assets' in metrics and 'total_liabilities' in metrics: ratios['debt_to_asset'] = metrics['total_liabilities'] / metrics['total_assets'] return ratios

4.3 财报情感分析

使用FinBERT模型分析管理层讨论与分析(MD&A)部分的情感倾向：

from transformers import BertTokenizer, BertForSequenceClassification import torch class FinancialSentimentAnalyzer: """财报情感分析""" def __init__(self, model_path='ProsusAI/finbert'): self.tokenizer = BertTokenizer.from_pretrained(model_path) self.model = BertForSequenceClassification.from_pretrained(model_path) self.model.eval() def analyze(self, text: str) -> Dict: """分析文本情感""" inputs = self.tokenizer( text[:512], # 截取前512个token return_tensors="pt", truncation=True, max_length=512 ) with torch.no_grad(): outputs = self.model(**inputs) probs = torch.softmax(outputs.logits, dim=1) sentiment = torch.argmax(probs).item() confidence = probs[0][sentiment].item() return { 'sentiment': ['negative', 'neutral', 'positive'][sentiment], 'confidence': round(confidence, 3), 'text_snippet': text[:200] + '...' }

4.4 财报对比分析

实现多期财报的对比分析功能：

class FinancialReportComparator: """财报对比分析器""" def compare(self, reports: List[Dict]) -> Dict: """对比多期财报""" comparison = { 'metrics_trend': {}, 'growth_rates': {}, 'anomalies': [] } # 提取各期指标 metrics_list = [r['metrics'] for r in reports] # 计算指标趋势 for metric in metrics_list[0].keys(): values = [m.get(metric) for m in metrics_list] if all(v is not None for v in values): comparison['metrics_trend'][metric] = values # 计算增长率 for i in range(1, len(metrics_list)): growth = {} for metric, values in comparison['metrics_trend'].items(): if i < len(values): try: growth_rate = (values[i] - values[i-1]) / abs(values[i-1]) * 100 growth[metric] = round(growth_rate, 2) # 异常检测 if abs(growth_rate) > 50: comparison['anomalies'].append({ 'metric': metric, 'period': i, 'growth_rate': growth_rate, 'description': f"{metric}在期{i}增长异常: {growth_rate}%" }) except (ZeroDivisionError, TypeError): continue comparison['growth_rates'][f"period_{i}"] = growth return comparison

5. 标书智能处理模块实现

5.1 标书解析与RAG系统

标书处理的核心是构建高效的检索增强生成(RAG)系统：

from langchain_community.vectorstores import Chroma from langchain_community.embeddings import HuggingFaceEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.document_loaders import PyPDFLoader, Docx2txtLoader class BidDocumentRAG: """标书RAG系统""" def __init__(self, persist_dir="./chroma_db"): self.embedding = HuggingFaceEmbeddings( model_name="./models/bge-large-zh" ) self.vectorstore = Chroma( persist_directory=persist_dir, embedding_function=self.embedding ) self.text_splitter = RecursiveCharacterTextSplitter( chunk_size=512, chunk_overlap=50, separators=["\n\n", "\n", "。", "；"] ) def ingest_documents(self, file_paths: List[str]): """摄入标书文档""" documents = [] for path in file_paths: if path.endswith('.pdf'): loader = PyPDFLoader(path) elif path.endswith('.docx'): loader = Docx2txtLoader(path) else: continue docs = loader.load() docs = [doc for doc in docs if len(doc.page_content) > 20] # 过滤空文档 documents.extend(docs) # 分块处理 chunks = self.text_splitter.split_documents(documents) # 存入向量数据库 self.vectorstore.add_documents(chunks) self.vectorstore.persist() print(f"已摄入 {len(documents)} 个文档，切分为 {len(chunks)} 个片段") def query(self, question: str, k: int = 3) -> Dict: """查询标书内容""" docs = self.vectorstore.similarity_search(question, k=k) return { 'answer': self._generate_answer(question, docs), 'sources': [{ 'content': doc.page_content[:200] + '...', 'source': doc.metadata.get('source', 'unknown'), 'page': doc.metadata.get('page', 0) } for doc in docs] } def _generate_answer(self, question: str, docs: List) -> str: """生成回答""" context = "\n\n".join([doc.page_content for doc in docs]) prompt = f"""基于以下标书内容回答问题： {context} 问题：{question} 要求： 1. 回答需明确具体 2. 引用标书中的具体章节或页码 3. 如不确定，请说明"根据现有文档无法确定" 回答：""" # 这里可以接入LLM生成更专业的回答 return self._simplify_answer(question, context) def _simplify_answer(self, question: str, context: str) -> str: """简化版回答生成（无LLM时使用）""" if "资质要求" in question: return "满足招标文件第三章第2.1条资质要求" elif "交付时间" in question: return "根据标书第五章约定，项目交付时间为合同签订后90天内" else: return "相关信息见标书" + context[:50] + "..."

5.2 标书自动生成

基于模板和RAG系统实现标书自动生成：

from docx import Document from docx.shared import Pt, Inches class BidGenerator: """标书生成器""" def __init__(self, template_path: str): self.template = Document(template_path) self.rag = None def set_rag_system(self, rag: BidDocumentRAG): """设置RAG系统""" self.rag = rag def generate(self, requirements: List[Dict], output_path: str): """生成标书""" doc = Document() # 封面页 self._add_cover_page(doc) # 点对点应答表 self._add_response_table(doc, requirements) # 技术方案章节 self._add_technical_solution(doc) # 保存文档 doc.save(output_path) def _add_cover_page(self, doc: Document): """添加封面页""" doc.add_heading('技术投标文件', 0).alignment = 1 # 居中 doc.add_paragraph('\n\n\n') # 空行 doc.add_heading('项目名称: XXX项目', 1).alignment = 1 doc.add_heading('投标单位: XXX公司', 1).alignment = 1 doc.add_paragraph('\n\n\n\n') doc.add_paragraph('日期: ' + datetime.now().strftime('%Y年%m月%d日')).alignment = 2 # 右对齐 # 添加分页符 doc.add_page_break() def _add_response_table(self, doc: Document, requirements: List[Dict]): """添加点对点应答表""" doc.add_heading('一、点对点应答', level=1) table = doc.add_table(rows=1, cols=4) table.style = 'Light Grid Accent 1' # 表头 hdr_cells = table.rows[0].cells hdr_cells[0].text = '序号' hdr_cells[1].text = '招标要求' hdr_cells[2].text = '应答' hdr_cells[3].text = '说明' # 填充内容 for i, req in enumerate(requirements, 1): row_cells = table.add_row().cells row_cells[0].text = str(i) row_cells[1].text = req['description'] if self.rag: response = self.rag.query(req['description']) row_cells[2].text = "完全满足" if "满足" in response['answer'] else "部分满足" row_cells[3].text = response['answer'] else: row_cells[2].text = "完全满足" row_cells[3].text = "详见技术方案部分" def _add_technical_solution(self, doc: Document): """添加技术方案章节""" doc.add_heading('二、技术方案', level=1) sections = [ ('1. 系统架构设计', 'architecture'), ('2. 功能实现方案', 'implementation'), ('3. 项目实施计划', 'timeline'), ('4. 售后服务方案', 'service') ] for title, section_key in sections: doc.add_heading(title, level=2) if self.rag: response = self.rag.query(title) doc.add_paragraph(response['answer']) else: doc.add_paragraph(f"此处详细描述{title}...") # 添加示意图 # self._add_placeholder_image(doc)

6. 系统集成与运维

6.1 系统监控实现

完善的监控系统对生产环境至关重要：

import psutil import requests from datetime import datetime import logging class SystemMonitor: """系统监控""" def __init__(self, api_endpoint=None): self.api_endpoint = api_endpoint self.logger = logging.getLogger('monitor') def check_system_health(self) -> Dict: """检查系统健康状态""" status = { 'timestamp': datetime.now().isoformat(), 'cpu': psutil.cpu_percent(interval=1), 'memory': psutil.virtual_memory().percent, 'disk': psutil.disk_usage('/').percent, 'processes': len(psutil.pids()), 'gpu': self._get_gpu_status() } # 记录日志 self.logger.info(f"系统状态: CPU {status['cpu']}%, 内存 {status['memory']}%") # 发送告警 if status['cpu'] > 90 or status['memory'] > 90: self._send_alert(status) return status def _get_gpu_status(self) -> Dict: """获取GPU状态""" try: import pynvml pynvml.nvmlInit() device_count = pynvml.nv