本教程详细介绍了如何将autogen多代理框架与nicegui web界面集成，以实现在web ui中实时显示代理对话内容。核心在于利用autogen代理的`register_reply`功能，通过注册自定义回调函数来拦截并处理代理间的消息，从而将这些消息同步到nicegui的聊天界面，解决autogen默认输出至终端而不在ui中显示的问题。
1. 引言
在开发基于大型语言模型（LLM）的应用程序时，AutoGen提供了一个强大的多代理会话框架。然而，当我们需要将AutoGen的对话输出集成到交互式Web界面（如NiceGUI）时，会遇到一个常见挑战：AutoGen代理的响应通常默认打印到控制台，而不是直接返回给前端UI进行渲染。本文将详细阐述如何通过利用AutoGen的register_reply机制，实现将代理的实时对话内容无缝同步到NiceGUI的Web界面。
2. 问题分析
在原始的NiceGUI与AutoGen集成尝试中，开发者可能会发现用户输入能被NiceGUI捕获并发送给AutoGen代理，但代理的响应却未能显示在Web界面上。user_proxy.initiate_chat()方法虽然启动了会话，但其返回值通常是会话的总结信息或最后一个消息的结构化表示，而非代理在会话过程中产生的每一条具体对话内容。这意味着我们需要一种更底层的方式来“监听”代理之间的交流。
3. 解决方案：利用 register_reply
AutoGen提供了一个名为 register_reply 的方法，允许开发者在代理接收或发送消息时注册自定义的回调函数。通过这种机制，我们可以拦截代理间的任何消息，并在消息传递时执行自定义逻辑，例如将其添加到NiceGUI的聊天消息列表中。
3.1 register_reply 工作原理
register_reply 方法允许您指定一个 reply_func，这个函数会在特定的代理（recipient）收到消息时被调用。其核心参数包括：
- recipient: 消息接收方代理。
- reply_func: 一个回调函数，它接收 recipient, messages, sender, config 等参数。
- config: 可选的配置字典。
在 reply_func 中，messages 参数是一个列表，其中 messages[-1] 包含了最新接收到的消息。我们可以从中提取消息的发送者名称和内容。
3.2 实现消息捕获回调函数
为了将AutoGen的消息同步到NiceGUI，我们需要定义一个全局可访问的列表（例如 nicegui_messages），并在回调函数中将捕获到的消息添加到这个列表中。
import autogen
from nicegui import ui, context
from uuid import uuid4
import asyncio
# 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
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
)
# 用于存储NiceGUI聊天消息的全局列表
nicegui_messages = []
# 定义回调函数，捕获代理消息并添加到NiceGUI消息列表
def capture_agent_message(recipient, messages, sender, config):
"""
当代理接收到消息时，此回调函数将被调用。
它将消息内容和发送者添加到NiceGUI的全局消息列表中。
"""
if messages and messages[-1]:
latest_message = messages[-1]
name = latest_message.get('name', sender.name) # 尝试获取消息中的名称，否则使用发送者名称
content = latest_message.get('content', '')
# 避免重复添加，并过滤掉可能不需要显示给用户的内部消息
# 这里可以根据实际需求进行更复杂的过滤
if content and name not in ["TERMINATE", "exit"]: # 简单的过滤示例
# 确保在NiceGUI的事件循环中更新UI
asyncio.create_task(add_message_to_nicegui(name, content))
return False, None # 返回False表示不阻止消息传递，None表示没有额外回复
async def add_message_to_nicegui(name, content):
"""异步将消息添加到NiceGUI的显示列表并刷新UI"""
nicegui_messages.append((name, content))
# 刷新NiceGUI UI，如果当前有活跃的客户端连接
if context.get_client().has_socket_connection:
chat_messages.refresh()
# 将回调函数注册到user_proxy和assistant，以便捕获它们之间的所有消息
# 注册到user_proxy，捕获它收到的来自assistant的消息
user_proxy.register_reply(
[autogen.Agent, None], # 捕获所有Agent发送给user_proxy的消息
reply_func=capture_agent_message,
config={"callback": None},
)
# 注册到assistant，捕获它收到的来自user_proxy的消息
assistant.register_reply(
[autogen.Agent, None], # 捕获所有Agent发送给assistant的消息
reply_func=capture_agent_message,
config={"callback": None},
)
@ui.page('/')
def main():
user_id = str(uuid4()) # Unique ID for each user session
@ui.refreshable
def chat_messages():
for name, text in nicegui_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)')
async def send():
user_message = task_input.value
if not user_message:
return
nicegui_messages.append(('You', user_message)) # Append user's message to the messages list
chat_messages.refresh() # Refresh chat messages to display the latest message
task_input.value = '' # Clear the input field after sending the message
try:
# 启动会话，但我们不再依赖其返回值来获取每条消息
# 而是通过注册的回调函数来捕获
await user_proxy.initiate_chat(assistant, message=user_message)
except Exception as e:
await add_message_to_nicegui('System Error', f"Error: {e}")
finally:
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(placeholder='输入你的消息...').props('clearable').classes('w-full').on('keydown.enter', send)
ui.button('发送', on_click=send)
ui.run(title='Chat with Albert')4. 代码详解与注意事项
4.1 注册回调函数
- user_proxy.register_reply(...) 和 assistant.register_reply(...): 我们将 capture_agent_message 函数注册到了 user_proxy 和 assistant 两个代理上。这意味着无论是 assistant 收到 user_proxy 的消息，还是 user_proxy 收到 assistant 的消息，capture_agent_message 都会被触发。这确保了我们能够捕获到双方的对话。
- [autogen.Agent, None]: 这个参数表示回调函数将应用于任何 autogen.Agent 类型的发送者，并且不限制接收者类型（None）。在我们的例子中，由于是注册在特定代理上，它会捕获所有发给该代理的消息。
4.2 capture_agent_message 函数
- messages[-1]: 包含了当前触发回调的最新消息字典。
- name = latest_message.get('name', sender.name): 尝试从消息字典中获取发送者名称。如果消息字典中没有 name 字段（例如，当消息由 initiate_chat 隐式发送时），则回退使用 sender.name。
- content = latest_message.get('content', ''): 提取消息的具体文本内容。
- asyncio.create_task(add_message_to_nicegui(name, content)): 关键点。NiceGUI在异步环境中运行，直接在回调函数中修改UI状态可能会导致线程问题。因此，我们使用 asyncio.create_task 将UI更新操作调度到NiceGUI的事件循环中执行，确保UI更新的线程安全和响应性。
- return False, None: register_reply 的回调函数需要返回一个元组 (bool, Any)。False 表示消息会继续正常传递给下一个处理器，None 表示没有额外的回复内容。
4.3 NiceGUI UI更新
- nicegui_messages: 一个全局列表，用于存储所有需要显示在UI上的聊天消息。
- @ui.refreshable def chat_messages(): NiceGUI的 @ui.refreshable 装饰器使得这个函数可以在数据变化时被手动刷新，从而更新UI。
- add_message_to_nicegui 函数负责将新消息添加到 nicegui_messages 列表，并调用 chat_messages.refresh() 来触发UI更新。
- ui.run_javascript('setTimeout(() => window.scrollTo(0, document.body.scrollHeight), 0)'): 确保聊天区域自动滚动到底部，显示最新消息。
4.4 错误处理
- 在 send 函数中，try...except 块用于捕获 initiate_chat 过程中可能发生的异常，并将错误信息显示在聊天界面中。
5. 总结
通过巧妙地利用AutoGen的 register_reply 机制，我们成功地解决了在NiceGUI Web界面中显示代理实时对话内容的挑战。这种方法不仅提供了对代理间消息的细粒度控制，还通过异步处理确保了UI的流畅性和响应性。这种集成模式为构建更丰富、更具交互性的基于AutoGen的Web应用程序奠定了基础。开发者可以根据此教程，进一步扩展和定制消息处理逻辑，例如实现消息过滤、格式化或与其他后端服务的集成。
