如何实现Chainlit聊天机器人获取Firebase用户CSV库存并转字典

原方案的核心问题在于Chainlit与Flask是两个独立运行的服务，彼此的会话（Session）不互通，无法通过Flask的Session共享库存数据；同时read_csv_from_firebase函数存在路径拼接错误，导致无法正确读取Firebase Storage中的文件。以下是两种可行的高效解决方案：

解决方案一：移除Flask中间层，Chainlit直接处理用户认证与库存加载（推荐） #

直接让Chainlit读取URL中的用户ID参数，从Firebase拉取对应库存，避免跨服务会话共享的问题。

1. 修改React端启动聊天逻辑 #

将用户ID作为URL参数直接传递给Chainlit页面，无需再调用Flask接口：

import { useAuthState } from 'react-firebase-hooks/auth';
import { auth } from "../firebase";

const Home = () => {
  const [user, loading, error] = useAuthState(auth);

  const handleClick = () => {
    if (loading) {
      console.log("Loading user information...");
      return;
    }
    if (error) {
      console.error("Error loading user information: ", error);
      return;
    }
    if (user) {
      // 直接跳转Chainlit页面并携带userId参数
      window.open(`http://localhost:8000?userId=${user.uid}`, "_blank");
    } else {
      console.log("No user is signed in.");
    }
  };

  // 组件渲染逻辑...
};

2. 修改Chainlit核心代码 #

移除Flask相关代码，在Chainlit会话启动时读取用户ID并加载库存：

import chainlit as cl
import pandas as pd
import firebase_admin
from firebase_admin import credentials, storage
from io import StringIO

# 初始化Firebase Admin SDK
cred = credentials.Certificate('./********-firebase-adminsdk-********.json')
firebase_admin.initialize_app(cred, {
    'storageBucket': '********.appspot.com'
})

storage_client = storage.bucket()

def read_csv_from_firebase(user_id):
    """从Firebase Storage读取用户库存CSV并转为字典"""
    try:
        blob = storage_client.blob(f"{user_id}/inventory.csv")
        csv_content = blob.download_as_text()
        # 将文本内容转为可读取的文件对象
        df = pd.read_csv(StringIO(csv_content))
        # 转为records格式（每条商品为一个字典，更适合LLM处理）
        return df.to_dict('records')
    except Exception as e:
        raise Exception(f"读取库存失败：{str(e)}")

# 会话启动钩子：加载用户专属库存
@cl.on_chat_start
async def start_chat():
    # 从URL参数获取用户ID
    query_params = cl.query_params
    user_id = query_params.get('userId')
    
    if not user_id:
        await cl.Message(content="请先登录并从仪表盘启动聊天").send()
        return
    
    try:
        inventory = read_csv_from_firebase(user_id)
        # 将库存存入Chainlit用户会话，供后续消息处理使用
        cl.user_session.set('inventory', inventory)
        await cl.Message(content=f"已加载您的库存，共{len(inventory)}件商品，现在可以开始咨询了！").send()
    except Exception as e:
        await cl.Message(content=str(e)).send()

# 消息处理钩子：使用库存生成回复
@cl.on_message
async def handle_message(message: cl.Message):
    inventory = cl.user_session.get('inventory')
    if not inventory:
        await cl.Message(content="未找到您的库存，请重新从仪表盘启动聊天").send()
        return
    
    # 调用SalesGPT业务逻辑，传入库存数据
    sales_agent = SalesGPT()
    ai_response = sales_agent.run(input=message.content, inventory=inventory)
    await cl.Message(content=ai_response).send()

解决方案二：保留Flask作为库存服务，Chainlit主动请求库存 #

如果已有Flask业务逻辑需要保留，可将Flask改为纯数据接口，Chainlit通过用户ID主动请求库存（不再依赖Session）。

1. 修改Flask接口 #

移除Session相关逻辑，改为根据用户ID返回库存：

from flask import Flask, request, jsonify
import pandas as pd
import firebase_admin
from firebase_admin import credentials, storage
from io import StringIO

app = Flask(__name__)

# 初始化Firebase Admin SDK
cred = credentials.Certificate('./********-firebase-adminsdk-********.json')
firebase_admin.initialize_app(cred, {
    'storageBucket': '********.appspot.com'
})

storage_client = storage.bucket()

def read_csv_from_firebase(user_id):
    blob = storage_client.blob(f"{user_id}/inventory.csv")
    csv_content = blob.download_as_text()
    df = pd.read_csv(StringIO(csv_content))
    return df.to_dict('records')

@app.route('/get-inventory', methods=['GET'])
def get_inventory():
    user_id = request.args.get('userId')
    if not user_id:
        return jsonify({"error": "缺少userId参数"}), 400
    
    try:
        inventory = read_csv_from_firebase(user_id)
        return jsonify({"inventory": inventory}), 200
    except Exception as e:
        return jsonify({"error": str(e)}), 500

if __name__ == '__main__':
    app.run(port=8001)  # 修改端口避免与Chainlit冲突

2. Chainlit调用Flask接口获取库存 #

import chainlit as cl
import requests

FLASK_API_URL = "http://localhost:8001/get-inventory"

@cl.on_chat_start
async def start_chat():
    query_params = cl.query_params
    user_id = query_params.get('userId')
    
    if not user_id:
        await cl.Message(content="请先登录并从仪表盘启动聊天").send()
        return
    
    try:
        # 调用Flask接口获取库存
        response = requests.get(FLASK_API_URL, params={"userId": user_id})
        response.raise_for_status()
        inventory = response.json()['inventory']
        
        cl.user_session.set('inventory', inventory)
        await cl.Message(content=f"已加载您的库存，共{len(inventory)}件商品，现在可以开始咨询了！").send()
    except Exception as e:
        await cl.Message(content=f"加载库存失败：{str(e)}").send()

# 消息处理逻辑同解决方案一
@cl.on_message
async def handle_message(message: cl.Message):
    inventory = cl.user_session.get('inventory')
    if not inventory:
        await cl.Message(content="未找到您的库存，请重新从仪表盘启动聊天").send()
        return
    
    sales_agent = SalesGPT()
    ai_response = sales_agent.run(input=message.content, inventory=inventory)
    await cl.Message(content=ai_response).send()

额外注意事项 #

1. 权限验证：生产环境需添加Firebase ID Token验证，防止恶意用户传入他人ID，可在React端获取ID Token后传递给Chainlit/Flask，后端验证Token合法性后再返回库存。
2. 错误处理：需补充文件不存在、权限不足等场景的错误提示。
3. Chainlit版本适配：部分版本中cl.query_params可能需替换为cl.context.session.get('query_params')，请参考Chainlit官方文档。

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