如何在Chainlit应用中直接获取UI反馈并全局访问？

Chainlit没有内置的全局访问反馈对象的函数，但可以通过扩展自定义数据层实现需求，以下是几种实用方案：

方案1：全局变量存储（简单场景） #

定义全局容器存储所有反馈，在upsert_feedback中写入，其他代码片段直接读取该容器。

修改你的代码如下：

import chainlit as cl
import chainlit.data as cl_data

# 全局存储反馈的容器
feedback_store = []

class CustomDataLayer(cl_data.BaseDataLayer):
    async def upsert_feedback(self, feedback: cl_data.Feedback) -> str:
        # 将反馈存入全局容器
        feedback_store.append(feedback)
        # 保留原有逻辑同步到Literal AI
        return await super().upsert_feedback(feedback)

cl_data._data_layer = CustomDataLayer()

# 在其他函数中访问反馈示例
@cl.on_message
async def on_message(message: cl.Message):
    if feedback_store:
        print("已收集的反馈：")
        for fb in feedback_store:
            print(f"评分：{fb.score}, 评论：{fb.comment}, 会话ID：{fb.thread_id}")
    # 其他业务逻辑
    await cl.Message(content="收到消息").send()

方案2：本地文件持久化（需永久存储时） #

如果需要把反馈永久保存到本地，可在upsert_feedback中将反馈写入JSON文件，其他地方读取该文件。

示例代码：

import chainlit as cl
import chainlit.data as cl_data
import json
from pathlib import Path

# 定义存储文件路径
FEEDBACK_FILE = Path("feedback_records.json")

# 初始化文件（不存在则创建空列表）
if not FEEDBACK_FILE.exists():
    with open(FEEDBACK_FILE, "w", encoding="utf-8") as f:
        json.dump([], f)

class CustomDataLayer(cl_data.BaseDataLayer):
    async def upsert_feedback(self, feedback: cl_data.Feedback) -> str:
        # 将反馈转为可序列化字典
        feedback_dict = {
            "score": feedback.score,
            "comment": feedback.comment,
            "thread_id": feedback.thread_id,
            "created_at": feedback.created_at.isoformat(),
            "user_id": feedback.user_id
        }
        # 读取现有数据并追加新反馈
        with open(FEEDBACK_FILE, "r+", encoding="utf-8") as f:
            data = json.load(f)
            data.append(feedback_dict)
            f.seek(0)
            json.dump(data, f, indent=2, ensure_ascii=False)
        # 保留原有逻辑同步到Literal AI
        return await super().upsert_feedback(feedback)

cl_data._data_layer = CustomDataLayer()

# 在其他函数中读取本地存储的反馈示例
@cl.button_callback("view_feedback")
async def view_feedback():
    with open(FEEDBACK_FILE, "r", encoding="utf-8") as f:
        feedbacks = json.load(f)
    feedback_text = "\n".join([f"会话{fb['thread_id']}：{fb['comment']}（评分：{fb['score']}）" for fb in feedbacks])
    await cl.Message(content=f"本地存储的反馈：\n{feedback_text}").send()

方案3：异步队列通知（复杂异步场景） #

如果需要在异步任务中实时处理反馈，可用asyncio.Queue传递反馈对象，其他任务监听队列获取新反馈。

示例代码：

import chainlit as cl
import chainlit.data as cl_data
import asyncio

# 创建异步队列
feedback_queue = asyncio.Queue()

class CustomDataLayer(cl_data.BaseDataLayer):
    async def upsert_feedback(self, feedback: cl_data.Feedback) -> str:
        # 将反馈放入队列
        await feedback_queue.put(feedback)
        # 保留原有逻辑同步到Literal AI
        return await super().upsert_feedback(feedback)

cl_data._data_layer = CustomDataLayer()

# 后台监听队列的任务
@cl.on_chat_start
async def start_feedback_listener():
    async def listen_feedback():
        while True:
            feedback = await feedback_queue.get()
            # 在这里处理新反馈，比如存入数据库、触发其他业务逻辑
            print(f"收到新反馈：会话{feedback.thread_id}，评论：{feedback.comment}")
            feedback_queue.task_done()
    # 创建后台任务
    asyncio.create_task(listen_feedback())

注意事项 #

• 全局变量方案适合单进程场景，多进程部署需改用共享存储（如Redis）
• 本地文件存储要注意文件锁问题，高并发场景建议用数据库代替
• 所有方案均保留super().upsert_feedback(feedback)，确保反馈同步到Literal AI看板不受影响

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