基于 Qwen3.8-Max 构建 AI 合同精审系统：文档解析、多轮条款分析 Pipeline 工程实战

python --version# Python 3.12.x

cd contract-reviewerpip install -r requirements.txt

flask>=3.0.0          # Web 框架flask-cors>=4.0.0     # 跨域支持python-dotenv>=1.0.0  # 环境变量管理openai>=1.12.0        # Qwen API 调用（OpenAI 兼容格式）pdfplumber>=0.10.3    # PDF 文档解析python-docx>=1.1.0    # Word 文档解析fpdf2>=2.7.7          # PDF 报告生成

# .env 配置文件DASHSCOPE_API_KEY=sk-your-api-key-hereDASHSCOPE_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1MODEL_NAME=qwen3.8-maxPORT=5000

python app.py

contract-reviewer/├── app.py              # Flask 主应用（API 路由 + 演示数据 + 对比引擎）├── analyzer.py         # AI 分析引擎（文档解析 + 条款提取 + 批量风险审查）├── config.py           # 配置管理（环境变量 + 路径常量）├── reports.py          # PDF 审查报告生成（封面 + 摘要 + 逐条分析 + 建议）├── requirements.txt    # Python 依赖清单│├── templates/│   └── index.html      # 前端主页面（上传 / 预览 / 加载 / 结果四区域）│├── static/│   ├── css/style.css   # 完整样式表（CSS 变量主题 + 响应式布局）│   └── js/app.js       # 前端交互逻辑（上传 / 分析 / 渲染 / 图表）│└── samples/    └── sample_contract.txt  # 示例合同（技术咨询服务合同，含已知风险条款）

def parse_document(self, file_path: str) -> dict:    ext = os.path.splitext(file_path)[1].lower()    parsers = {        ".pdf": self._parse_pdf,     # pdfplumber 逐页提取        ".docx": self._parse_docx,   # python-docx 段落合并        ".doc":  self._parse_docx,        ".txt":  self._parse_txt,    # 多编码尝试 utf-8/gbk/gb2312        ".md":   self._parse_txt,    }    parser = parsers.get(ext)    if not parser:        raise ValueError(f"不支持的文件格式: {ext}")    text = parser(file_path)    if not text.strip():        raise ValueError("文档内容为空，请检查文件")    return {"text": text, "char_count": len(text), "filename": os.path.basename(file_path)}

uploadArea.addEventListener('drop', (e) => {    e.preventDefault();    uploadArea.classList.remove('dragover');    if (e.dataTransfer.files.length) handleFileUpload(e.dataTransfer.files[0]);});

上传文档 → 解析文本 → AI 提取条款 → AI 批量风险分析 → AI 整体评估 → 加权评分 → 渲染结果                                    ↓ (批量失败时)                              回退到逐条分析（每条单独调 API）

prompt = """你是一位专业的中国合同审查律师。请仔细阅读以下合同文本，提取所有独立条款。
要求：1. 保留原始条款编号2. 完整保留条款原文内容3. 每个条款独立编号4. 包括合同的基本信息条款（如当事方、签署日期等）5. 包括所有附件和补充条款
请以JSON数组格式返回：[{"id": "条款编号", "text": "条款完整原文", "title": "条款标题/主题"}]"""

prompt = f"""c。请对以下合同的所有条款逐条进行风险分析。
分析维度：1. 霸王条款：单方面加重对方义务或限制对方权利2. 模糊表述：可能导致争议的模糊用语3. 缺失条款：缺少必要的限定条件或例外情形4. 条款矛盾：与其他条款存在逻辑矛盾5. 违法风险：可能违反《民法典》《消费者权益保护法》《劳动法》等6. 违约金异常：违约金比例过高或过低
风险等级标准：- critical：严重风险，可能导致重大经济损失或法律纠纷- high：高风险，需要立即修改- medium：中风险，建议优化- low：低风险，基本可接受
请以JSON数组格式返回每个条款的分析结果：[{{    "clause_index": 0,    "risk_level": "critical/high/medium/low",    "risk_types": ["具体风险类型"],    "analysis": "详细风险分析说明（100-200字）",    "suggestion": "具体修改建议，给出修改后的条款文本",    "risk_score": 1到10的整数,    "legal_basis": "相关法律条文引用"}}]"""

prompt = f"""你是一位资深的合同审查律师。请基于以下信息对合同进行整体评估。
合同原文摘要（前1000字）：{{text[:1000]}}
高风险条款摘要：{{risk_summary}}
共审查 {{len(clauses)}} 条条款。
请以JSON格式返回整体评估：{{    "summary": "合同整体评估概述（200字以内）",    "overall_risk": "critical/high/medium/low",    "strengths": ["合同优点1", "合同优点2"],    "critical_issues": ["关键问题1", "关键问题2"],    "key_suggestions": ["核心建议1", "核心建议2", "核心建议3"],    "contract_type": "合同类型（如劳动合同/买卖合同/租赁合同等）"}}"""

def _call_ai(self, prompt: str, max_tokens: int = 4096) -> str:    """调用 Qwen3.8-Max API（显式关闭 thinking 以加速 JSON 输出）"""    try:        response = self.client.chat.completions.create(            model=self.model,            messages=[                {"role": "system", "content": "你是一位拥有15年经验的中国合同审查律师，"                                              "擅长发现合同中的风险条款、霸王条款和模糊表述。"                                              "请始终以有效的JSON格式返回分析结果，不要包含任何非JSON内容。"},                {"role": "user", "content": prompt},            ],            temperature=0.1,            max_tokens=max_tokens,            extra_body={"enable_thinking": False},  # 关键：关闭 thinking 加速 JSON 输出        )        content = response.choices[0].message.content        if content is None:            return "{}"        return content.strip()    except Exception as e:        print(f"[AI调用错误] {e}")        return "{}"

def _calculate_score(self, clauses: list) -> int:    if not clauses:        return 50    RISK_WEIGHTS = {"critical": 10, "high": 7, "medium": 4, "low": 1}    total_penalty = 0    for clause in clauses:        level = clause.get("risk_level", "low")        weight = self.RISK_WEIGHTS.get(level, 1)        risk_score = clause.get("risk_score", 5)        total_penalty += weight * (risk_score / 10.0)    max_penalty = len(clauses) * 10    return max(0, min(100, round(100 - (total_penalty / max_penalty * 100))))

function renderClauses(clauses) {    const riskLabels = { critical: '严重', high: '高危', medium: '中等', low: '低危' };    clauses.forEach(clause => {        const level = clause.risk_level || 'low';        const types = (clause.risk_types || []).map(t => `<span class="type-tag">${escapeHtml(t)}</span>`).join('');        const scorePercent = (clause.risk_score || 0) * 10;        card.innerHTML = `            <div class="clause-header">                <span class="clause-title">条款 ${escapeHtml(clause.original_id || '?')}：${escapeHtml(clause.title || '')}</span>                <span class="risk-badge ${level}">${riskLabels[level] || level}</span>            </div>            ${types ? `<div class="clause-types">${types}</div>` : ''}            <div class="clause-text">${escapeHtml(clause.original_text || '')}</div>            <div class="clause-section-label">🔍 风险分析</div>            <div class="clause-analysis">${escapeHtml(clause.analysis || '无分析')}</div>            <div class="clause-section-label">✏️ 修改建议</div>            <div class="clause-suggestion">${escapeHtml(clause.suggestion || '无建议')}</div>            ${clause.legal_basis ? `<div class="clause-legal">📖 ${escapeHtml(clause.legal_basis)}</div>` : ''}            <div class="risk-score-bar">                <div class="risk-score-fill" style="width:${scorePercent}%"></div>            </div>        `;    });}

// 环形饼图 - 风险等级分布const pieCtx = document.getElementById('riskPieChart');pieCtx._chart = new Chart(pieCtx, {    type: 'doughnut',    data: {        labels: ['严重', '高危', '中等', '低危'],        datasets: [{ data: values, backgroundColor: ['#dc2626', '#ea580c', '#ca8a04', '#16a34a'] }],    },});
// 柱状图 - 各条款风险评分const barCtx = document.getElementById('riskBarChart');barCtx._chart = new Chart(barCtx, {    type: 'bar',    data: {        labels: clauseLabels,        datasets: [{ label: '风险评分', data: clauseScores, backgroundColor: clauseColors }],    },    options: { scales: { y: { beginAtZero: true, max: 10 } } },});

def _build_comparison_table(ai_analysis: dict, human: dict) -> list:    ai_risk = ai_analysis.get("risk_counts", {})    human_risk = human.get("risk_counts", {})    def get_risk(d, key, cn_key):        return d.get(key, d.get(cn_key, 0))  # 兼容中英文 key    return [        {"dimension": "安全评分", "ai": f"{ai_analysis.get('overall_score', 0)}/100",         "human": f"{human['score']}/100", "winner": "ai" if ... else "human"},        {"dimension": "审查时间", "ai": ai_analysis.get("analysis_duration", "?"),         "human": human["duration"], "winner": "ai"},        {"dimension": "审查成本", "ai": "≈ ¥0.5",         "human": human["cost"], "winner": "ai"},        {"dimension": "严重风险发现", "ai": str(ai_critical),         "human": str(human_critical), "winner": "ai" if ai_critical >= human_critical else "human"},        ...    ]

def _get_human_benchmark() -> dict:    return {        "score": 78, "total_clauses": 25,        "risk_counts": {"critical": 2, "high": 3, "medium": 4, "low": 16},        "duration": "约 2-4 小时", "cost": "¥2,000 - ¥5,000", "issues_found": 9,        "notes": [            "人工审查受疲劳影响，可能遗漏低风险条款",            "不同律师审查标准不一致",            "审查周期长，影响业务进度",            "按时/按份计费，成本较高",        ],    }

@app.route("/api/analyze-demo/<task_id>", methods=["POST"])def analyze_demo(task_id):    if task_id not in analysis_store:        analysis_store[task_id] = {            "filename": "示例合同 - 技术咨询服务合同.txt",            "text": "",            "created_at": datetime.now().isoformat(),        }    demo_result = _get_demo_analysis()  # 内置 12 条示例分析数据    analysis_store[task_id]["analysis"] = demo_result    return jsonify(demo_result)

# 解决：显式关闭 thinkingextra_body={"enable_thinking": False}# 效果：从超时 10 分钟 → 101 秒完成全流程

# 解决：三管齐下# 1. 默认 max_tokens 4096 → 8192# 2. 条款提取专用 max_tokens=16384# 3. 新增 _repair_truncated_json() 截断修复：JSON 数组被截断时，#    自动找最后一个完整对象并闭合数组，避免全部丢失

# 解决 1：analyze_demo 端点自动创建 task 条目# 解决 2：兼容函数，两种 key 都尝试def _count_issues(risk_counts: dict) -> int:    en_total = sum(risk_counts.get(k, 0) for k in ["critical", "high", "medium"])    cn_total = sum(risk_counts.get(k, 0) for k in ["严重风险", "高风险", "中风险"])    return max(en_total, cn_total)

function escapeHtml(str) {    if (!str) return '';    const div = document.createElement('div');    div.textContent = str;    return div.innerHTML;}
selectFileBtn.addEventListener('click', (e) => {    e.stopPropagation();    fileInput.click();});