使用通义千问 API 实现非结构化数据提取与分析

3.4 结果解析与验证

由于 LLM 可能会偶尔输出 Markdown 代码块标记（如 json ... ），在解析前需要进行清洗。同时，必须验证 JSON 的合法性。

def parse_llm_response(response_text):
    if not response_text:
        return []
    
    # 清理可能的 markdown 标记
    cleaned_text = response_text.strip()
    if cleaned_text.startswith("```json") or cleaned_text.startswith("```"):
        cleaned_text = cleaned_text.split("\n", 1)[-1].rsplit("```", 1)[0].strip()
    
    try:
        data = json.loads(cleaned_text)
        # 简单的数据校验逻辑
        if isinstance(data, list):
            return data
        elif isinstance(data, dict):
            return [data]
        else:
            return []
    except json.JSONDecodeError as e:
        print(f"JSON 解析错误：{e}")
        return []

4. 完整脚本示例

将上述步骤整合为一个完整的可运行脚本。

import dashscope
import json
import os
from dotenv import load_dotenv

# 加载环境变量中的 API_KEY，避免硬编码
load_dotenv()
API_KEY = os.getenv("DASHSCOPE_API_KEY")

if not API_KEY:
    raise ValueError("请在 .env 文件中配置 DASHSCOPE_API_KEY")

dashscope.api_key = API_KEY

def extract_data_from_html(html_content):
    prompt = f"""
    你是一个数据分析助手。请从以下 HTML 内容中提取结构化数据。
    目标字段：name (string), picture (string), info (string), score (float), commentsNumber (int)。
    输出格式：JSON Array。
    
    HTML Content:
    {html_content}
    """
    
    messages = [{'role': 'user', 'content': prompt}]
    
    try:
        response = Generation.call(
            model=dashscope.Generation.Models.qwen_turbo,
            messages=messages,
            result_format='message'
        )
        
        if response.status_code == 200:
            raw_output = response.output.choices[0].message.content
            return clean_and_parse_json(raw_output)
        else:
            print(f"Error: {response.code}, {response.message}")
            return None
            
    except Exception as e:
        print(f"Exception occurred: {e}")
        return None

def clean_and_parse_json(text):
    text = text.strip()
    if text.startswith("```json"):
        text = text.replace("```json", "").replace("```", "").strip()
    elif text.startswith("```"):
        text = text.replace("```", "").strip()
    
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return []

if __name__ == "__main__":
    sample_html = "<div><h3>测试电影</h3><p>简介</p></div>"
    result = extract_data_from_html(sample_html)
    print(json.dumps(result, ensure_ascii=False, indent=2))

5. 最佳实践与注意事项

5.1 安全性

• 密钥管理：永远不要将 API Key 直接写在代码中。建议使用环境变量（如 .env 文件配合 python-dotenv）或云平台的密钥管理服务。
• 权限控制：为不同的应用场景创建独立的 API Key，以便追踪用量和限制权限。

5.2 性能优化

• Token 消耗：注意输入内容的长度。过长的 HTML 可能导致 Token 超限或成本增加。建议先进行预处理，只保留必要标签。
• 并发处理：对于大量数据，可以使用 asyncio 或线程池并发调用 API，但需注意 API 的 QPS（每秒查询率）限制。

5.3 容错机制

• 重试策略：网络波动可能导致请求失败。建议在调用外层添加指数退避重试机制。
• 格式校验：LLM 的输出并非 100% 可靠，务必在业务层对返回的 JSON 进行 Schema 校验，防止下游程序崩溃。

6. 总结

通过集成通义千问 API，我们可以将传统的数据提取工作流升级为智能化的数据处理流程。这种方法不仅降低了对特定网页结构的依赖，还能处理更复杂的语义信息。在实际生产中，建议结合规则引擎与大模型能力，构建混合式的数据处理方案，以平衡准确性、成本与效率。

开发者应持续关注模型版本的更新，并根据实际业务需求调整 Prompt 策略，以获得最佳的提取效果。