使用 FastAPI 传输 LangChain OpenAI 响应 [重复]

我想将 OpenAI 的响应直接传输到 FastAPI 的端点。

Code:

在我的threads_handler.py（位于单独的文件夹中）中，我有以下函数askQuestion()

def askQuestion(self, collection_id, question):
    collection_name = "collection-" + str(collection_id)
    self.llm = ChatOpenAI(model_name=self.model_name, temperature=self.temperature, openai_api_key=os.environ.get('OPENAI_API_KEY'), streaming=True, verbose=VERBOSE, callback_manager=CallbackManager([MyCustomHandler()]))
    self.memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True,  output_key='answer')
    
    chroma_Vectorstore = Chroma(collection_name=collection_name, embedding_function=self.embeddingsOpenAi, client=self.chroma_client)

    self.chain = ConversationalRetrievalChain.from_llm(self.llm, chroma_Vectorstore.as_retriever(similarity_search_with_score=True),
                                                        return_source_documents=True,verbose=VERBOSE, 
                                                        memory=self.memory)
    
    result = self.chain({"question": question})
    return result['answer']

如您所见，我指定callback_manager=CallbackManager([MyCustomHandler()])) --> 我在这里定义了它：

q = queue.Queue()
stop_item = "###finish###"

class MyCustomHandler(StreamingStdOutCallbackHandler):
    def on_llm_start(
        self, serialized: dict[str, any], prompts: list[str], **kwargs: any
    ) -> None:
        """Run when LLM starts running."""
        with q.mutex:
            q.queue.clear()

    def on_llm_new_token(self, token: str, **kwargs: any) -> None:
        """Run on new LLM token. Only available when streaming is enabled."""
        sys.stdout.write(token)
        sys.stdout.flush()
        q.put(token)

FastAPI 中的实际路由定义如下：

@router.post("/collection/{collection_id}/ask_question")
async def ask_question(collection_id: str, request: Request):
    try:
        form_data = await request.form()
        question = form_data["question"]

        def generate_tokens():  
            result = thread_handler.askQuestion(collection_id, question)
            for token in result:
                yield token
        return StreamingResponse(generate_tokens(), media_type="application/json")


    except requests.exceptions.ConnectionError as e:
        raise HTTPException(status_code=500, detail="Error connecting to the server")
    except Exception as e:
        raise HTTPException(status_code=404, detail=str(e))

流式传输只是不想工作，但是响应正在终端中传输。但我希望它直接流式传输到 api

我在终端中使用以下 api 调用测试流式传输：

curl -q -N -X POST -F 'question=What are the aspects in data science?' http://0.0.0.0/api/v1/collection/52/ask_question

我可以看到您已经从 FastAPI 形成并返回了 StreamingResponse，但是，我觉得您没有考虑到您可能也需要对 cURL 请求进行一些更改。尝试按上述方式更改您的请求，并检查控制台中的输出。

import requests

url = 'your endpoint here'
headers = {
    'accept': 'application/json',
    'Content-Type': 'application/json'
}
data = {
    'input_text': 'your question here'
}

response = requests.post(url, headers=headers, json=data, stream=True)

for chunk in response.iter_content(chunk_size=1024):
    if chunk:
        print(chunk.decode('utf-8'), end = "", flush = True)