LangChain+ChatGPT+Pydantic结构化输出异常问题排查

问题梳理 #

• ChatGPT返回的日期未转换为datetime.date格式，保留文档原始格式
• 文档中出现"Lastname"时，未映射到Pydantic Enum定义的Surname
• 对LangChain链的组合方式存疑

核心原因与解决方案 #

1. 强化Prompt的格式约束 #

LLM对结构化输出的遵循度依赖明确指令，当前提示过于模糊，需补充以下细节：

• 明确要求输出严格符合Pydantic Schema的JSON格式
• 指定日期必须转换为YYYY-MM-DD格式（对应datetime.date的序列化标准格式）
• 明确枚举映射规则：文档中的"Lastname""Family Name"等均需映射为枚举值Surname

修改后的prompt_messages示例：

prompt_messages = [
    SystemMessage(
        content=(
            "你是专业的结构化信息提取工具，必须严格按照指定的JSON格式输出结果。"
            "日期必须转换为YYYY-MM-DD格式，文档中的Lastname/Family Name需映射为枚举值Surname，Given Name/First Name映射为Name。"
            "输出仅包含符合要求的JSON，不要添加任何额外解释。"
        )
    ),
    HumanMessage(content="基于以下上下文回答问题："),
    HumanMessagePromptTemplate.from_template("{context}"),
    HumanMessagePromptTemplate.from_template("问题：{question}"),
]

2. 优化Pydantic模型的描述性 #

给Enum和字段补充更明确的业务规则描述，帮助LLM理解映射逻辑：

from enum import Enum
from pydantic import BaseModel, Field
import datetime

class NameEnum(Enum):
    Name = 'Name'
    Surname = 'Surname'

class DocumentSchema(BaseModel):
    date: datetime.date = Field(..., description='文档日期，必须转换为YYYY-MM-DD格式的日期')
    name: NameEnum = Field(..., description='类型：Name对应名字（Given Name/First Name），Surname对应姓氏（Lastname/Family Name）')

3. 验证链的配置正确性 #

当前链的组合逻辑是正确的，但需注意两个细节：

• create_structured_output_chain会自动为LLM添加函数调用约束，确保输出结构，无需手动构建冗余提示
• 保持temperature=0（已设置），减少LLM输出的随机性，提升格式一致性

4. 添加容错处理（可选） #

为避免格式异常导致程序崩溃，可添加Pydantic解析的容错逻辑：

try:
    data = retrieval_qa_pydantic.run(question)
    doc_data = DocumentSchema.parse_raw(data)
except Exception as e:
    print(f"格式解析错误：{e}")
    # 可根据需求添加重试或降级逻辑

完整修改后的代码示例 #

FILE_PATH = 'foo.pdf'

from enum import Enum
from pydantic import BaseModel, Field
import datetime
from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, PromptTemplate, HumanMessagePromptTemplate
from langchain.schema import SystemMessage, HumanMessage
from langchain.chains import create_structured_output_chain, StuffDocumentsChain, RetrievalQA

class NameEnum(Enum):
    Name = 'Name'
    Surname = 'Surname'

class DocumentSchema(BaseModel):
    date: datetime.date = Field(..., description='文档日期，必须转换为YYYY-MM-DD格式的日期')
    name: NameEnum = Field(..., description='类型：Name对应名字（Given Name/First Name），Surname对应姓氏（Lastname/Family Name）')

def main():
    loader = PyPDFLoader(FILE_PATH)
    data = loader.load()
    text_splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=10)
    all_splits = text_splitter.split_documents(data)
    vectorstore = Chroma.from_documents(documents=all_splits, embedding=OpenAIEmbeddings())
    llm = ChatOpenAI(model_name='gpt-3.5-turbo', temperature=0)
    question = """What is the date on the document?
        Is it about a name or surname?
    """

    doc_prompt = PromptTemplate(
        template="Content: {page_content}\nSource: {source}",
        input_variables=["page_content", "source"],
    )
    prompt_messages = [
        SystemMessage(
            content=(
                "你是专业的结构化信息提取工具，必须严格按照指定的JSON格式输出结果。"
                "日期必须转换为YYYY-MM-DD格式，文档中的Lastname/Family Name需映射为枚举值Surname，Given Name/First Name映射为Name。"
                "输出仅包含符合要求的JSON，不要添加任何额外解释。"
            )
        ),
        HumanMessage(content="基于以下上下文回答问题："),
        HumanMessagePromptTemplate.from_template("{context}"),
        HumanMessagePromptTemplate.from_template("问题：{question}"),
    ]

    chain_prompt = ChatPromptTemplate(messages=prompt_messages)

    chain = create_structured_output_chain(output_schema=DocumentSchema, llm=llm, prompt=chain_prompt)
    final_qa_chain_pydantic = StuffDocumentsChain(
        llm_chain=chain,
        document_variable_name="context",
        document_prompt=doc_prompt,
    )
    retrieval_qa_pydantic = RetrievalQA(
        retriever=vectorstore.as_retriever(), combine_documents_chain=final_qa_chain_pydantic
    )
    
    try:
        data = retrieval_qa_pydantic.run(question)
        doc_data = DocumentSchema.parse_raw(data)
        print(doc_data)
    except Exception as e:
        print(f"解析失败：{str(e)}")

if __name__ == "__main__":
    main()

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