迁移至Litestar后OpenAI无法返回补全，出现500内部服务器错误

核心问题 #

1. 返回类型不匹配：get_answer函数声明返回str，但实际返回字典{"answer": answer}，Litestar的类型校验会抛出错误，导致500状态码。
2. 同步阻塞调用阻塞异步事件循环：openai.Completion.create是同步阻塞方法，在Litestar的异步路由中直接调用会卡住事件循环，引发超时或内部错误。

具体修复步骤 #

1. 修正返回类型注解 #

修改get_answer函数的返回类型注解为dict，与实际返回值类型一致：

@get("/support/{question:str}")
async def get_answer(question: str) -> dict:  # 替换原str为dict
    # 函数内容后续调整

2. 用线程池包装同步OpenAI调用 #

在异步上下文里，将同步的OpenAI调用放到线程池中执行，避免阻塞事件循环：
先导入asyncio：

import asyncio

再替换原OpenAI调用代码：

# 替换原completion = openai.Completion.create(...)部分
loop = asyncio.get_event_loop()
completion = await loop.run_in_executor(
    None,
    lambda: openai.Completion.create(
        model="text-davinci-003",
        prompt=prompt,
        max_tokens=1000
    )
)

3. 优化Pinecone初始化（可选但推荐） #

将Pinecone的初始化逻辑移到函数外部，避免每次请求重复初始化：

load_dotenv()

embeddings = OpenAIEmbeddings()

# 仅初始化一次Pinecone
pinecone.init(
    api_key=os.getenv("PINECONE_API_KEY"),
    environment=os.environ.get('PINECONE_ENVIRONMENT'),
)
index_name = os.environ.get('PINECONE_INDEX_NAME')

4. 开启调试模式排查问题（可选） #

创建Litestar应用时开启调试模式，控制台会输出详细错误栈，方便定位问题：

app = Litestar([index, get_answer], debug=True)

修复后的完整代码示例 #

from dotenv import load_dotenv
from litestar import Litestar, get
import os
import pinecone
import asyncio
from langchain.embeddings.openai import OpenAIEmbeddings
from langchain.vectorstores import Pinecone
import openai

__all__ = (
    "index",
    "support",
)

load_dotenv()

embeddings = OpenAIEmbeddings()

# 初始化Pinecone仅一次
pinecone.init(
    api_key=os.getenv("PINECONE_API_KEY"),
    environment=os.environ.get('PINECONE_ENVIRONMENT'),
)
index_name = os.environ.get('PINECONE_INDEX_NAME')

@get("/")
async def index() -> str:
    return "Тестовый запрос выполнен. Чтобы получить ответ, воспользуйтесь командой /support/{вопрос%20вопрос}."

@get("/support/{question:str}")
async def get_answer(question: str) -> dict:
    k = 2
    docsearch = Pinecone.from_existing_index(index_name, embeddings)
    res = docsearch.similarity_search_with_score(question, k=k)

    prompt = f'''
    Use text below to compile an answer:
    {[x for x in res]}
    '''

    # 线程池执行同步OpenAI调用
    loop = asyncio.get_event_loop()
    completion = await loop.run_in_executor(
        None,
        lambda: openai.Completion.create(
            model="text-davinci-003",
            prompt=prompt,
            max_tokens=1000
        )
    )

    answer = completion.choices[0].text
    return {"answer": answer}

app = Litestar([index, get_answer], debug=True)

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