在NiceGUI中集成AutoGen并捕获Agent响应的教程

本教程旨在解决在nicegui应用中集成autogen时，如何实时捕获并显示agent对话响应的问题。通过深入解析autogen的`register_reply`机制，我们将展示如何注册自定义回调函数来拦截并处理agent的每一条消息，从而将autogen的强大对话能力无缝桥接到nicegui的交互式聊天界面中，确保所有对话内容都能在前端准确呈现。

1. 引言：AutoGen与NiceGUI的融合挑战

在构建基于大型语言模型（LLM）的交互式应用时，我们经常需要结合强大的LLM编排框架（如AutoGen）与灵活的Web UI框架（如NiceGUI）。AutoGen以其多Agent协作能力著称，能够模拟复杂的对话和任务执行流程。然而，当我们将AutoGen集成到NiceGUI等Web界面时，一个常见的问题是如何将AutoGen Agent在后台产生的实时对话内容准确、及时地呈现在用户界面上。AutoGen的initiate_chat方法通常会返回一个包含最终对话结果的对象，但它并不直接提供一个简单的方式来获取对话过程中Agent的每一步回复，而这些中间步骤对于构建流畅的用户体验至关重要。

2. AutoGen与NiceGUI基础设置

首先，我们来看一下AutoGen Agent和NiceGUI界面的基本配置。以下代码展示了如何初始化AutoGen Agent以及一个简单的NiceGUI聊天界面结构。

import autogen
from nicegui import ui, context
from uuid import uuid4

# 1. AutoGen 配置
config_list = [
    {
        'model': 'gpt-4',
        'api_key': 'YOUR_API_KEY' # 请替换为您的实际API密钥
    }
]
llm_config = {
    'seed': 42,
    'config_list': config_list,
    'temperature': 0.2
}

# 2. 初始化AutoGen Agent
assistant = autogen.AssistantAgent(name='Albert', llm_config=llm_config)
user_proxy = autogen.UserProxyAgent(
    name='user_proxy',
    human_input_mode="NEVER", # 在NiceGUI中，用户输入通过UI提供，Agent无需直接请求
    max_consecutive_auto_reply=10, # 允许更多轮次自动回复
    is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
    code_execution_config={"work_dir": "web"}, # 设定代码执行的工作目录
    llm_config=llm_config
)

# 3. NiceGUI 界面定义
@ui.page('/')
def main():
    # 存储聊天消息的列表，用于UI显示
    messages = []
    user_id = str(uuid4()) # 为每个用户会话生成唯一ID

    @ui.refreshable
    def chat_messages():
        """刷新聊天消息显示区域"""
        for name, text in messages:
            ui.chat_message(text=text, name=name, sent=name == 'You')
        if context.get_client().has_socket_connection:
            # 自动滚动到底部
            ui.run_javascript('setTimeout(() => window.scrollTo(0, document.body.scrollHeight), 0)')

    # ... (发送消息的逻辑将在后面修改)
    # ... (UI布局)

在上述代码中，我们配置了AssistantAgent和UserProxyAgent。UserProxyAgent的human_input_mode设置为"NEVER"，因为用户输入将通过NiceGUI的文本框提供。NiceGUI部分则定义了一个messages列表来存储对话历史，并通过@ui.refreshable装饰器实现消息的动态更新和滚动。

3. 核心挑战：捕获Agent的实时响应

最初的尝试可能是在send函数中调用user_proxy.initiate_chat后，尝试从其返回值中提取Agent的响应。然而，initiate_chat返回的是一个ChatResult对象，其中包含了整个对话的概要和最终状态，但并不直接提供Agent在对话过程中每一步的详细回复内容。这意味着，如果Agent进行了多轮思考或与UserProxyAgent进行了多次交互，这些中间过程将不会被NiceGUI捕获并显示，用户只能看到最终结果，这严重影响了用户体验。

例如，原始代码中尝试这样获取响应：

# 原始的尝试
response = await user_proxy.initiate_chat(assistant, message=user_message)
if response and 'content' in response[0]: # 这里的response[0]通常是最后一条消息
    assistant_response = response[0]['content']
    messages.append(('Albert', assistant_response))

这种方法只能捕获到initiate_chat结束后，对话中最后一条消息的内容，无法实时跟踪整个对话流程。

4. 解决方案：利用register_reply机制

AutoGen提供了一个强大的回调机制——register_reply，它允许我们在Agent发送或接收消息时注册一个自定义函数。这个函数会在每次消息交换时被调用，从而使我们能够实时捕获并处理Agent的对话内容。

4.1 register_reply的工作原理

register_reply方法允许你为Agent注册一个或多个回调函数。每当Agent准备发送回复或接收到消息时，这些回调函数就会被触发。其基本签名如下：

agent.register_reply(
    trigger: Union[Type[Agent], Agent, List[Union[Type[Agent], Agent]], None],
    reply_func: Callable,
    config: Optional[Any] = None,
    reset_config: Optional[Callable] = None,
    position: Optional[int] = None
)

其中：

trigger: 指定何时触发回调。可以是一个Agent实例、Agent类型、Agent列表，或者None（表示对所有消息都触发）。
reply_func: 这是一个可调用对象（函数），它将在消息交换时被执行。它的签名通常是 (recipient, messages, sender, config)。
- recipient: 接收当前消息的Agent。
- messages: 一个列表，包含当前对话回合中所有的消息（通常是最新消息）。
- sender: 发送当前消息的Agent。
- config: 注册时传递的配置字典。
reply_func的返回值：回调函数需要返回一个元组 (bool, Optional[str])。
- True表示回调函数已经处理了回复，Agent无需再生成回复。
- False表示Agent应继续其默认的回复生成逻辑。
- Optional[str]是回调函数生成的回复内容，如果返回True，则此内容将作为Agent的回复。

4.2 关键回调函数设计

为了将Agent的实时消息显示在NiceGUI界面上，我们需要设计一个回调函数，它能够从messages参数中提取出最新的消息内容和发送者，然后将其添加到NiceGUI的messages列表中，并触发UI刷新。

# 定义一个辅助函数来更新UI消息列表并刷新
def append_and_refresh_ui_messages(sender_name: str, content: str):
    """将消息添加到UI列表并刷新NiceGUI界面"""
    messages.append((sender_name, content))
    chat_messages.refresh()

# AutoGen回调函数
def autogen_reply_callback(recipient, current_messages, sender, config):
    """
    AutoGen Agent消息回调函数，用于捕获Agent的实时回复。
    """
    # current_messages 是一个列表，包含当前回合的所有消息。
    # 最后一条消息是刚刚发送或接收的消息。
    if current_messages:
        last_message = current_messages[-1]
        sender_name = last_message.get('name', sender.name) # 优先使用消息中的name，否则使用sender的name
        content = last_message.get('content', '')

        if content:
            # 使用NiceGUI的run_sync将UI更新操作调度到主线程
            ui.run_sync(lambda: append_and_refresh_ui_messages(sender_name, content))

    # 返回 False, None 表示Agent应继续其默认的回复生成逻辑
    return False, None

注意： 由于AutoGen的Agent对话可能在后台线程中进行，而NiceGUI的UI更新必须在主线程执行，因此我们使用ui.run_sync来安全地调度append_and_refresh_ui_messages函数。

4.3 注册回调函数

现在，我们将这个回调函数注册到assistant和user_proxy这两个Agent上。这样，无论是assistant回复user_proxy，还是user_proxy向assistant发送消息，我们都能捕获到。

# 在 main() 函数内部，在 Agent 初始化之后，NiceGUI UI定义之前
# 注册回调函数
assistant.register_reply(
    [autogen.Agent, None], # 捕获assistant发送给任何Agent的消息
    reply_func=autogen_reply_callback,
    config={"callback": None},
)
user_proxy.register_reply(
    [autogen.Agent, None], # 捕获user_proxy发送给任何Agent的消息
    reply_func=autogen_reply_callback,
    config={"callback": None},
)

5. 整合到NiceGUI聊天界面

将register_reply回调机制整合到NiceGUI的main函数中，并调整send函数。现在，send函数只需要负责接收用户输入并启动AutoGen对话，具体的Agent回复将由autogen_reply_callback处理。

import autogen
from nicegui import ui, context
from uuid import uuid4

# AutoGen Configuration
config_list = [
    {
        'model': 'gpt-4',
        'api_key': 'YOUR_API_KEY' # 替换为您的实际API密钥
    }
]
llm_config = {
    'seed': 42,
    'config_list': config_list,
    'temperature': 0.2
}

# Initialize AutoGen Agents (在 main() 外部初始化，以便在回调中访问)
assistant = autogen.AssistantAgent(name='Albert', llm_config=llm_config)
user_proxy = autogen.UserProxyAgent(
    name='user_proxy',
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
    is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
    code_execution_config={"work_dir": "web"},
    llm_config=llm_config
)

@ui.page('/')
def main():
    messages = [] # 存储聊天消息的列表
    user_id = str(uuid4())

    @ui.refreshable
    def chat_messages():
        for name, text in messages:
            ui.chat_message(text=text, name=name, sent=name == 'You')
        if context.get_client().has_socket_connection:
            ui.run_javascript('setTimeout(() => window.scrollTo(0, document.body.scrollHeight), 0)')

    # 定义辅助函数和回调函数，它们需要访问到 main() 作用域内的 messages 和 chat_messages
    def append_and_refresh_ui_messages(sender_name: str, content: str):
        messages.append((sender_name, content))
        chat_messages.refresh()

    def autogen_reply_callback(recipient, current_messages, sender, config):
        if current_messages:
            last_message = current_messages[-1]
            sender_name = last_message.get('name', sender.name)
            content = last_message.get('content', '')

            if content:
                ui.run_sync(lambda: append_and_refresh_ui_messages(sender_name, content))
        return False, None

    # 注册回调函数
    assistant.register_reply(
        [autogen.Agent, None],
        reply_func=autogen_reply_callback,
        config={"callback": None},
    )
    user_proxy.register_reply(
        [autogen.Agent, None],
        reply_func=autogen_reply_callback,
        config={"callback": None},
    )

    async def send():
        user_message = task_input.value
        if not user_message.strip():
            return # 防止发送空消息

        # 用户自己的消息直接添加到UI列表
        append_and_refresh_ui_messages('You', user_message)
        task_input.value = '' # 清空输入框

        try:
            # 启动AutoGen对话，Agent的回复将通过回调函数处理
            await user_proxy.initiate_chat(assistant, message=user_message)
        except Exception as e:
            # 异常处理，显示错误信息
            ui.run_sync(lambda: append_and_refresh_ui_messages('Albert', f"Error: {e}"))
        finally:
            # 确保在对话结束后刷新UI（虽然回调已经刷新，但以防万一）
            ui.run_sync(chat_messages.refresh)


    with ui.scroll_area().classes('w-full h-60 p-3 bg-white overflow-auto'):
        chat_messages()

    with ui.footer().style('position: fixed; left: 0; bottom: 0; width: 100%; background: white; padding: 10px; box-shadow: 0 -2px 5px rgba(0,0,0,0.1);'):
        task_input = ui.input().style('width: calc(100% - 100px);').on('keydown.enter', send) # 支持回车发送
        ui.button('Send', on_click=send).style('width: 90px;')

ui.run(title='Chat with Albert')

6. 注意事项与最佳实践

Agent初始化位置：将AutoGen Agent的初始化放在ui.page装饰的main()函数外部，可以确保Agent实例在整个应用生命周期中只被创建一次，并能在main()函数内部的回调中被访问。
回调函数的线程安全：由于AutoGen Agent的运行可能涉及多线程，而NiceGUI的UI更新必须在主线程进行，因此使用ui.run_sync()是至关重要的。它确保了对messages列表和chat_messages.refresh()的调用是在NiceGUI的主事件循环中安全执行的。
reply_func的返回值：在我们的场景中，回调函数仅用于监听和显示消息，不应干预Agent的默认回复逻辑，因此返回False, None是正确的选择。如果需要自定义Agent的回复行为，可以返回True和一个字符串作为新的回复。
max_consecutive_auto_reply：为了让Agent能够进行多轮次的自主对话，将UserProxyAgent的max_consecutive_auto_reply设置为一个较大的值（例如10），可以避免对话过早终止。
错误处理：在send函数中使用try...except块来捕获initiate_chat可能抛出的异常，并将错误信息显示在UI上，提升用户体验。
消息来源识别：在autogen_reply_callback中，last_message.get('name', sender.name)能够更准确地识别消息的发送者，因为有时消息字典中会包含name字段，它可能比sender.name更具描述性。
终止消息：is_termination_msg参数对于控制Agent何时停止对话至关重要。确保您的Agent在完成任务后发送一个符合终止条件的字符串（例如TERMINATE）。

7.

# javascript # java # 前端 # app # 回调函数 # ai # proxy # win # gpt # gpt-4 # 作用域 # try # 字符串 # bool # 循环 # 线程 # 多线程 # 主线程 # 对象 # 事件 # ui # 回调 # 您的 # 自定义 # 是一个 # 至关重要 # 是在 # 将在 # 都能 # 返回值 # 新和

返回目录在线咨询

上一篇：门户网站开发：构建高效信息平台的关键策略
下一篇：C语言实现数据结构串（堆分配存储表示法）实例详解

您的项目需求

*请认真填写需求信息，我们会在24小时内与您取得联系。

全网整合营销服务商