Langchain批量LLM多条目JSON输出解析问题咨询

核心问题分析 #

你当前的问题在于PydanticOutputParser绑定的是单个metdata对象，但批量请求后LLM返回的是JSON数组，解析器无法匹配；同时还存在LLM返回损坏JSON的情况。Langchain完全支持批量场景，不需要完全自行实现，只需调整模型和解析器配置即可。

解决方案步骤 #

1. 调整Pydantic模型为批量结构 #

定义一个包含多个metdata对象的容器模型，让解析器能识别JSON数组：

from langchain.pydantic_v1 import BaseModel, Field
from typing import List

# 单个运动队元数据模型（保持原有字段定义）
class metdata(BaseModel):
    Sport_team: str = Field(description="Sport_team")
    stadium: str = Field(description="stadium")
    coach: str = Field(description="coach")
    city: str = Field(description="city")
    country: str = Field(description="country")
    website: str = Field(description="website")

# 批量结果容器模型，适配多条目返回
class TeamMetadataList(BaseModel):
    teams: List[metdata] = Field(description="List of metadata for each sport team")

2. 配置支持批量的解析器+错误修复 #

使用PydanticOutputParser绑定批量模型，再结合OutputFixingParser自动修复损坏的JSON格式：

from langchain.output_parsers import PydanticOutputParser, OutputFixingParser

# 初始化批量解析器
base_parser = PydanticOutputParser(pydantic_object=TeamMetadataList)
# 添加错误修复层：解析失败时自动调用LLM修正输出格式
fixing_parser = OutputFixingParser.from_llm(llm=model, parser=base_parser)

3. 优化提示词，明确批量输出要求 #

修改系统提示词，清晰指定输出格式，避免LLM返回歧义内容：

system_prompt = """
处理给定的运动队列表，为每个队伍返回指定元数据字段。
输出必须是严格的JSON格式，结构为：
{
  "teams": [
    {"Sport_team": "队伍名", "stadium": "球场名", "coach": "教练名", "city": "城市", "country": "国家", "website": "官网"},
    ...
  ]
}
确保每个队伍的所有请求字段都被填充，不要输出任何JSON以外的内容。
{format_instructions}
"""

4. 修改解析调用与结果处理逻辑 #

更新get_completion_with_parser使用带修复的解析器，并调整结果转换为DataFrame的逻辑：

def get_completion_with_parser(prompt, parser, callback_info=False):
    with get_openai_callback() as cb:
        output = model(prompt)
    output_content = output.content
    # 使用带自动修复的解析器处理批量结果
    response = parser.parse(output_content)

    if callback_info:
        return response, cb.total_tokens, cb.total_cost
    return response

# 在循环中处理返回结果
response, total_tokens, total_cost = gpt_teams_metadata(team_names=team_names, metadata_fields=metadata_fields)
# 从批量模型中提取单个条目并转为DataFrame
team_data = [team.dict() for team in response.teams]
response_df = pd.DataFrame(team_data)

5. 手动回退逻辑（可选） #

如果OutputFixingParser仍失败，可添加重试机制捕获异常：

def get_completion_with_parser(prompt, parser, callback_info=False, max_retries=2):
    retry_count = 0
    while retry_count < max_retries:
        try:
            with get_openai_callback() as cb:
                output = model(prompt)
            output_content = output.content
            response = parser.parse(output_content)
            if callback_info:
                return response, cb.total_tokens, cb.total_cost
            return response
        except Exception as e:
            retry_count += 1
            print(f"解析失败，重试第{retry_count}次: {str(e)}")
            time.sleep(10)
    # 重试耗尽后记录错误，返回空数据或原始内容
    print("重试次数耗尽，解析失败")
    return None, 0, 0

内容的提问来源于stack exchange，提问作者Angle