FastAPI集成LangChain：invoke接口返回空输出，流式接口正常

问题场景 #

• 基于LangChain构建客户反馈分类应用，使用AnalysisAttributes Pydantic类定义分类属性，本地调用full_pipeline.invoke可正常得到预期结果
• 通过add_routes将pipeline部署到FastAPI后，流式接口/categorize_feedback/stream/能正常返回结果，但invoke接口/categorize_feedback/invoke/返回空输出{'output': {}}
• 移除Pydantic输出解析器后，invoke接口恢复正常，当前临时采用自定义FastAPI接口实现功能

可能原因 #

• LangChain的add_routes方法在处理带有Pydantic输出解析器的pipeline时，存在序列化兼容问题，解析后的Pydantic对象未被正确转换为可返回的JSON格式
• add_routes默认的响应处理逻辑无法识别解析后的Pydantic模型实例，导致输出被清空

解决方案 #

方案一：保留自定义FastAPI接口（优化版） #

将pipeline初始化移至接口外，避免每次请求重复创建实例，提升性能：

# 提前初始化pipeline
full_pipeline = prompt | model | parser

@app.post('/categorize_feedback')
async def categorize_feedback(input_data: FeedbackInput) -> AnalysisAttributes:
    output = full_pipeline.invoke(input_data.model_dump())
    return output

方案二：修复add_routes的解析器兼容问题 #

1. 升级LangChain版本：旧版本可能存在add_routes与Pydantic解析器的兼容bug，升级至最新版本尝试解决
2. 指定响应模型与响应类：在add_routes时显式声明响应模型，强制正确序列化：

from fastapi.responses import JSONResponse

add_routes(
    app,
    full_pipeline,
    path="/categorize_feedback",
    response_model=AnalysisAttributes,
    response_class=JSONResponse
)

3. 添加序列化转换步骤：在pipeline中增加将Pydantic对象转为字典的步骤，适配add_routes的处理逻辑：

# 修改pipeline，增加字典转换
full_pipeline = prompt | model | parser | (lambda x: x.dict())

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