Flask集成Langchain与AzureOpenAI流式响应报错问题咨询

问题描述 #

使用Flask开发REST API，集成Langchain与AzureOpenAI时，llm.stream无法正常工作，报错TypeError: 'function' object is not iterable，需要实现AzureOpenAI的响应流式输出。

示例代码 #

@app.route('/stream2', methods=['GET'])
def stream2():
    try:
        user_query = request.json.get('user_query')
        if not user_query:
            return "No user query provided", 400
        
        callback_handler = StreamHandler()
        callback_manager = CallbackManager([callback_handler])
        llm = AzureChatOpenAI(
            azure_endpoint=AZURE_OPENAI_ENDPOINT,
            openai_api_version=OPENAI_API_VERSION,
            deployment_name=OPENAI_DEPLOYMENT_NAME,
            openai_api_key=OPENAI_API_KEY,
            openai_api_type=OPENAI_API_TYPE,
            model_name=OPENAI_MODEL_NAME,
            streaming=True,
            model_kwargs={
                "logprobs": None,
                "best_of": None,
                "echo": None
            },
            #callback_manager=callback_manager, 
            temperature=0)
        
        @stream_with_context
        async def generate():
            async for chunk in llm.stream(user_query):
                yield chunk
        
        return Response(generate(), mimetype='text/event-stream')
    except Exception as e:
        logger.error(f"An error occurred: {e}")
        return "An error occurred", 500

错误信息 #

Traceback (most recent call last):
  File "c:\Users\x\repos\y-chatbot\backend-flask-appservice\.venv\lib\site-packages\werkzeug\serving.py", line 362, in run_wsgi
    execute(self.server.app)
  File "c:\Users\x\repos\y-chatbot\backend-flask-appservice\.venv\lib\site-packages\werkzeug\serving.py", line 325, in execute
    for data in application_iter:
  File "c:\Users\x\repos\y-chatbot\backend-flask-appservice\.venv\lib\site-packages\werkzeug\wsgi.py", line 256, in __next__
    return self._next()
  File "c:\Users\x\repos\y-chatbot\backend-flask-appservice\.venv\lib\site-packages\werkzeug\wrappers\response.py", line 32, in _iter_encoded        
    for item in iterable:
TypeError: 'function' object is not iterable

系统信息 #

langchain==0.1.0
langchain-community==0.0.12
langchain-core==0.1.12
langchainhub==0.1.14

问题原因 #

1. 请求方法不匹配：路由使用GET方法，但尝试读取request.json（JSON请求体），GET请求通常不携带请求体，无法正确获取用户查询。
2. 异步同步混淆：Langchain 0.1.x版本中AzureChatOpenAI.stream是同步方法，返回同步迭代器，代码中错误使用async for遍历；同时Flask默认是同步环境，异步生成器无法被正确迭代。
3. 返回格式问题：直接返回Chunk对象不符合SSE（Server-Sent Events）格式，且生成器处理方式错误导致迭代失败。

修复后的代码 #

from flask import Flask, request, Response, stream_with_context
from langchain_community.chat_models import AzureChatOpenAI
import logging

logger = logging.getLogger(__name__)
app = Flask(__name__)

# 替换为你的Azure OpenAI配置
AZURE_OPENAI_ENDPOINT = "你的Azure端点"
OPENAI_API_VERSION = "API版本"
OPENAI_DEPLOYMENT_NAME = "部署名称"
OPENAI_API_KEY = "你的API密钥"
OPENAI_API_TYPE = "azure"
OPENAI_MODEL_NAME = "模型名称"

@app.route('/stream2', methods=['POST'])
def stream2():
    try:
        user_query = request.json.get('user_query')
        if not user_query:
            return "No user query provided", 400
        
        llm = AzureChatOpenAI(
            azure_endpoint=AZURE_OPENAI_ENDPOINT,
            openai_api_version=OPENAI_API_VERSION,
            deployment_name=OPENAI_DEPLOYMENT_NAME,
            openai_api_key=OPENAI_API_KEY,
            openai_api_type=OPENAI_API_TYPE,
            model_name=OPENAI_MODEL_NAME,
            streaming=True,
            temperature=0,
            model_kwargs={
                "logprobs": None,
                "best_of": None,
                "echo": None
            }
        )
        
        @stream_with_context
        def generate():
            # 遍历同步流式响应，提取内容并按SSE格式输出
            for chunk in llm.stream(user_query):
                if chunk.content:  # 过滤空内容的Chunk
                    yield f"data: {chunk.content}\n\n"
        
        return Response(generate(), mimetype='text/event-stream')
    except Exception as e:
        logger.error(f"An error occurred: {e}")
        return "An error occurred", 500

关键调整说明 #

• 修改请求方法为POST：确保能正确接收JSON格式的用户查询参数。
• 使用同步生成器：适配Langchain同步的stream方法，用普通for循环遍历Chunk，避免异步迭代错误。
• 符合SSE格式输出：将Chunk中的content按data: 内容\n\n格式返回，前端可直接通过SSE API监听流式数据。
• 过滤空内容Chunk：避免输出无效的空数据，提升流式体验。

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