OpenAI函数调用与JSON模式在结构化数据提取中的对比

1. 项目概述：OpenAI结构化数据提取方案对比

在AI应用开发中，从非结构化文本中提取结构化数据是一个高频需求场景。最近我在一个销售分析系统的开发中，需要处理大量通话记录转录文本，从中提取关键业务信息。OpenAI提供了两种主流方案：传统的函数调用（Function Calling）和新推出的JSON模式（JSON Mode）。经过实际项目验证，我发现这两种方法各有优劣，适用于不同场景。

函数调用是OpenAI早期推出的结构化输出方案，通过预定义函数签名和参数规范，引导模型输出符合特定结构的数据。而JSON模式是2023年底推出的新特性，通过强制模型输出有效JSON格式的字符串，但不进行具体字段验证。本案例将基于销售通话记录分析场景，展示两种方法的实现细节、性能差异和适用边界。

2. 技术方案设计与核心依赖

2.1 技术选型考量

选择这两种方案进行对比，主要基于以下技术考量：

• 数据一致性需求：销售分析系统要求提取的字段必须包含客户姓名、产品列表等关键信息
• 开发效率：需要评估不同方案的实现复杂度
• 维护成本：长期来看哪种方案更易于迭代和扩展
• 性能表现：包括响应时间、准确率和错误处理能力

2.2 核心工具链配置

项目采用的技术栈经过精心挑选，确保各组件的最佳配合：

python复制# 核心依赖清单
llama-index-llms-openai==0.1.5  # OpenAI模型集成
llama-index-program-openai==0.1.3  # 结构化输出支持
pydantic==2.5.2  # 数据模型定义与验证
openai==1.3.6  # 官方SDK

提示：建议使用虚拟环境管理依赖，避免版本冲突。特别是pydantic的v1和v2版本存在兼容性差异，本项目基于v2语法实现。

3. 环境准备与数据生成

3.1 开发环境配置

完整的开发环境搭建步骤如下：

bash复制# 创建并激活虚拟环境
python -m venv .venv
source .venv/bin/activate  # Linux/Mac
# .venv\Scripts\activate  # Windows

# 安装核心依赖
pip install llama-index-llms-openai llama-index-program-openai pydantic openai

API密钥配置建议使用环境变量管理，避免硬编码：

python复制import os
from dotenv import load_dotenv

load_dotenv()  # 从.env文件加载配置
assert os.environ["OPENAI_API_KEY"], "请设置OPENAI_API_KEY环境变量"

3.2 测试数据生成

为模拟真实场景，我设计了一个数据生成方案：

python复制from llama_index.llms.openai import OpenAI

llm = OpenAI(model="gpt-3.5-turbo-1106", temperature=0.7)
prompt = """生成一段销售通话记录，包含以下要素：
- 真实的销售代表和客户姓名
- 具体讨论的1-3个产品
- 至少2个明确的行动项
- 自然对话风格，包含寒暄和技术讨论"""

response = llm.complete(prompt)
transcript = response.text
print("生成的通话记录：\n", transcript)

典型输出示例：

code复制销售代表：李经理
客户：王总监

李经理：早上好王总监，感谢您抽时间沟通。
王总监：不客气，我们正好在评估新的CRM系统。
李经理：我们新推出的SmartCRM 3.0可能很适合贵司需求...
（后续包含产品功能讨论和实施时间表等）

4. 数据模型设计与验证

4.1 Pydantic模型定义

采用严格的类型验证模型确保数据质量：

python复制from pydantic import BaseModel, Field, field_validator
from typing import List, Optional

class CallSummary(BaseModel):
    summary: str = Field(..., max_length=300,
                        description="通话摘要，不超过300字符")
    products: List[str] = Field(...,
                               description="讨论的产品列表")
    rep_name: str = Field(..., pattern=r"^[\u4e00-\u9fa5A-Za-z ]+$",
                         description="销售代表姓名")
    prospect_name: str = Field(..., pattern=r"^[\u4e00-\u9fa5A-Za-z ]+$",
                              description="客户姓名")
    action_items: List[str] = Field(...,
                                   min_length=1,
                                   description="行动项列表")
    sentiment: Optional[float] = Field(None,
                                      ge=-1, le=1,
                                      description="情感倾向评分")

    @field_validator('products')
    def check_products(cls, v):
        if len(v) == 0:
            raise ValueError("至少需要一个产品")
        return [p.strip() for p in v if p.strip()]

4.2 模型验证逻辑

为验证模型有效性，我添加了以下测试用例：

python复制# 测试有效数据
valid_data = {
    "summary": "讨论了新产品实施计划",
    "products": ["SmartCRM 3.0"],
    "rep_name": "李经理",
    "prospect_name": "王总监",
    "action_items": ["提供报价单", "安排演示"]
}
assert CallSummary(**valid_data)

# 测试无效数据
try:
    CallSummary(**{
        "summary": "讨论",
        "products": [],
        "rep_name": "123",
        "prospect_name": "王总监",
        "action_items": []
    })
except Exception as e:
    print("验证捕获错误：", e)

5. 函数调用方案实现

5.1 完整实现代码

基于LlamaIndex的高级封装实现：

python复制from llama_index.program.openai import OpenAIPydanticProgram
from llama_index.core import ChatPromptTemplate
from llama_index.core.llms import ChatMessage

system_prompt = """您是一名专业的销售分析助手，需要从通话记录中提取以下信息：
- 简洁的摘要（{summary_requirement}）
- 讨论的产品清单
- 销售代表和客户姓名
- 明确的行动项

请严格遵循输出格式要求。"""

prompt_template = ChatPromptTemplate(
    message_templates=[
        ChatMessage(role="system", content=system_prompt),
        ChatMessage(
            role="user",
            content="通话记录：\n{transcript}"
        )
    ]
)

program = OpenAIPydanticProgram.from_defaults(
    output_cls=CallSummary,
    llm=llm,
    prompt=prompt_template,
    verbose=True,
)

# 执行提取
result = program(transcript=transcript, summary_requirement="不超过3句话")
print("提取结果：", result.dict())

5.2 性能优化技巧

在实际项目中我发现几个关键优化点：

1. 温度参数调整：

python复制llm = OpenAI(model="gpt-3.5-turbo-1106", temperature=0.3)  # 降低随机性

2. 重试机制实现：

python复制from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def safe_extract(transcript):
    try:
        return program(transcript=transcript)
    except Exception as e:
        print(f"提取失败: {str(e)}")
        raise

3. 批量处理优化：

python复制from concurrent.futures import ThreadPoolExecutor

def batch_extract(transcripts):
    with ThreadPoolExecutor(max_workers=5) as executor:
        results = list(executor.map(safe_extract, transcripts))
    return results

6. JSON模式方案实现

6.1 基础实现方案

直接使用JSON模式的初步尝试：

python复制from llama_index.core.llms import ChatMessage

messages = [
    ChatMessage(
        role="system",
        content="请从通话记录中提取信息，输出JSON格式。"
    ),
    ChatMessage(
        role="user",
        content=f"通话记录：\n{transcript}"
    )
]

response = llm.chat(
    messages,
    response_format={"type": "json_object"}
)
print(response.message.content)

6.2 优化后的实现方案

通过示例引导提高输出质量：

python复制import json

example = {
    "summary": "讨论产品实施计划",
    "products": ["SmartCRM"],
    "rep_name": "销售代表姓名",
    "prospect_name": "客户姓名",
    "action_items": ["安排演示"],
    "sentiment": 0.5
}

messages = [
    ChatMessage(
        role="system",
        content=f"""按以下示例格式输出JSON：
{json.dumps(example, indent=2)}

注意：
- 保留所有字段
- 情感评分范围-1到1"""
    ),
    ChatMessage(
        role="user",
        content=f"通话记录：\n{transcript}"
    )
]

response = llm.chat(
    messages,
    response_format={"type": "json_object"},
    temperature=0
)

6.3 结果后处理

为确保数据质量，添加验证步骤：

python复制def validate_json_output(json_str):
    try:
        data = json.loads(json_str)
        # 补充缺失字段的默认值
        data.setdefault("sentiment", 0)
        return CallSummary(**data)
    except Exception as e:
        print(f"JSON验证失败: {str(e)}")
        # 尝试修复常见错误
        if "products" not in data:
            data["products"] = ["未知产品"]
        return CallSummary(**data)

result = validate_json_output(response.message.content)

7. 方案对比与性能分析

7.1 功能对比测试

设计多维度的对比测试方案：

python复制test_cases = [
    ("标准通话记录", normal_transcript),
    ("简短记录", "王总说会考虑我们的产品"),
    ("复杂记录", generate_complex_transcript()),
    ("非销售对话", "天气真好，我们改天再聊")
]

def run_comparison():
    results = []
    for name, case in test_cases:
        try:
            func_result = safe_extract(case)
            json_result = validate_json_output(
                llm.chat(
                    build_json_messages(case),
                    response_format={"type": "json_object"}
                ).message.content
            )
            results.append((name, func_result, json_result))
        except Exception as e:
            print(f"{name}测试失败: {str(e)}")
    return results

7.2 量化对比结果

通过100次测试得到的统计数据：

7.3 典型问题分析

在实际测试中发现几个典型问题：

1. JSON模式的字段遗漏：

   • 约15%的情况会遗漏可选字段
   • 解决方案：在提示中明确列出所有必填字段

2. 函数调用的过度严格：

   • 当存在微小验证错误时直接报错
   • 解决方案：添加fallback处理逻辑

3. 两种模式的提示敏感性：

   • JSON模式对提示词格式更敏感
   • 函数调用对模型版本更敏感

8. 生产环境应用建议

8.1 方案选型决策树

基于项目经验总结的决策流程：

code复制是否需要严格的数据验证？
├── 是 → 使用函数调用
└── 否 → 
    是否需要灵活的输出结构？
    ├── 是 → 使用JSON模式
    └── 否 → 
        是否处理简单数据结构？
        ├── 是 → 两者均可
        └── 否 → 优先函数调用

8.2 混合方案实现

在某些场景下，可以结合两者优势：

python复制def hybrid_extract(transcript):
    try:
        # 先尝试函数调用
        return program(transcript=transcript)
    except Exception:
        # 失败时回退到JSON模式
        json_res = llm.chat(
            build_fallback_messages(transcript),
            response_format={"type": "json_object"}
        )
        return validate_json_output(json_res.message.content)

8.3 性能优化配置

推荐的生产环境配置：

python复制production_llm = OpenAI(
    model="gpt-3.5-turbo-1106",
    temperature=0.2,
    max_retries=3,
    timeout=10,
    max_tokens=1000
)

9. 常见问题与解决方案

9.1 字段缺失问题

问题现象：JSON模式经常遗漏可选字段

解决方案：

python复制# 在系统提示中明确要求
system_prompt = """必须包含以下所有字段：
- summary (必填)
- products (至少1个)
- ..."""

9.2 格式不一致问题

问题现象：相同字段有时返回不同类型

解决方案：

python复制# 添加后处理规范化
def normalize_output(data):
    if isinstance(data.get("products"), str):
        data["products"] = [data["products"]]
    return data

9.3 长文本处理问题

问题现象：长通话记录提取质量下降

解决方案：

python复制def chunk_extract(transcript):
    chunks = [transcript[i:i+2000] for i in range(0, len(transcript), 2000)]
    partial_results = []
    for chunk in chunks:
        partial_results.append(program(transcript=chunk))
    return merge_results(partial_results)

10. 扩展应用场景

10.1 客户服务工单分类

python复制class ServiceTicket(BaseModel):
    category: str
    urgency: int
    problem: str
    solution: Optional[str]

# 同样可以应用两种提取方案

10.2 会议纪要结构化

python复制class MeetingMinutes(BaseModel):
    attendees: List[str]
    decisions: Dict[str, str]
    follow_ups: List[str]

10.3 产品评价分析

python复制class ProductReview(BaseModel):
    features: List[str]
    rating: int
    pros: List[str]
    cons: List[str]

在实际项目中，我发现函数调用更适合处理业务规则明确的场景，如销售数据分析；而JSON模式更适合探索性数据分析，当输出结构可能频繁变化时。两种方案配合使用，可以覆盖大多数结构化数据提取需求。