AutoGen Core × Chainlit 实战：为单 Agent 与多 Agent 群聊构建流式 Web 聊天界面

AutoGen Core × Chainlit 实战为单 Agent 与多 Agent 群聊构建流式 Web 聊天界面【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogencore_chainlit是 AutoGen Python 仓库中官方提供的一个端到端示例目录位于 python/samples/core_chainlit演示如何用AutoGen Coreautogen-core的单线程 Agent 运行时对接Chainlit网页聊天框架实现「用户浏览器提问 → Agent 后台推理 → token 实时流式回显」的完整链路。读完本文你将掌握如何为单个 AssistantAgent 搭建可流式输出的聊天页如何把「Assistant Critic」两个 Agent 组成的群聊团队接入同一界面以及如何借助ClosureAgent、消息 Topic 与cl.user_session在 Chainlit 会话内管理运行时与输出队列。示例组成与运行机制概览根据示例的 READMEcore_chainlit旨在演示 Chainlit 与单线程 Agent 运行时SingleThreadedAgentRuntime集成的常见用法全部代码都运行在进程内、同一个事件循环中。它覆盖了六个关键组件单 Agent 场景一个在 Chainlit 环境中独立运行的 Agent可直接回复用户输入并能调用工具示例内置了一个查询天气的函数群聊场景一个双 Agent 的 Group Chat其中Assistant Agent响应用户输入Critic Agent负责对 Assistant 的输出进行反思与点评两者通过一个群聊管理器轮转发言Closure Agent闭包 Agent用一个普通异步闭包来定义一个收尾 Agent把所有输出消息聚合到一个 asyncio 输出队列中从而把「运行时的消息」搬回「UI 的可见世界」Token 流式传输把模型客户端的流式分块逐个转发到前端界面会话管理在 Chainlit 的用户会话cl.user_session中保存 runtime 与输出队列保证每个浏览器会话都有独立的 Agent 运行时。两个示例的入口文件分别是 app_agent.py单 Agent与 app_team.py双 Agent 群聊公共的 Agent 实现抽在 SimpleAssistantAgent.py 中。环境要求与安装示例要求Python 3.8 及以上版本并安装以下依赖安装命令来自 READMEpip install -U chainlit autogen-core autogen-ext[openai] pyyaml说明chainlit负责把 Python 函数包装成可在浏览器中交互的聊天 UIautogen-coreAutoGen Core 的 Agent 运行时核心库autogen-ext[openai]OpenAI 兼容的模型扩展包其中[openai]extra 会一并安装 OpenAI SDK 相关依赖pyyaml用于解析model_config.yaml模型配置文件。如果改用其他模型提供商只需要为autogen-ext安装对应的 extra例如 Anthropic、Gemini、Ollama 等提供商各有专属 extra并在配置文件中切换provider整体代码无需改动。模型配置的写法可参考本仓库各模型扩展的配置模板与 模型配置示例。模型配置从模板生成 model_config.yaml运行时通过ChatCompletionClient.load_component(model_config)见 app_agent.py从 YAML 字典加载模型客户端因此你需要在运行示例前创建model_config.yaml。直接复制模板即可cp model_config_template.yaml model_config.yaml模板 model_config_template.yaml 内给出了三种受支持的配置形态1. OpenAIAPI Keyprovider: autogen_ext.models.openai.OpenAIChatCompletionClient config: model: gpt-4o api_key: REPLACE_WITH_YOUR_API_KEY2. Azure OpenAIAPI Keyprovider: autogen_ext.models.openai.AzureOpenAIChatCompletionClient config: model: gpt-4o azure_endpoint: https://{your-custom-endpoint}.openai.azure.com/ azure_deployment: {your-azure-deployment} api_version: {your-api-version} api_key: REPLACE_WITH_YOUR_API_KEY3. Azure OpenAIEntra / AD Tokenprovider: autogen_ext.models.openai.AzureOpenAIChatCompletionClient config: model: gpt-4o azure_endpoint: https://{your-custom-endpoint}.openai.azure.com/ azure_deployment: {your-azure-deployment} api_version: {your-api-version} azure_ad_token_provider: provider: autogen_ext.auth.azure.AzureTokenProvider config: provider_kind: DefaultAzureCredential scopes: - https://cognitiveservices.azure.com/.default注意实际的api_key、azure_endpoint、azure_deployment、api_version等值都需要替换为你自己的凭据。运行时采用组件化声明式配置provider指向模型客户端类的完整限定名config下的字段与对应构造函数参数一一对应这正是 AutoGen Core 组件系统在模型接入上的典型用法。示例一单 Agent 聊天界面app_agent.py首先进入示例目录启动单 Agent 界面cd python/samples/core_chainlit chainlit run app_agent.pychainlit run会启动 Chainlit 自带的服务并自动打开网页聊天窗口。第一个消息界面直接与一个SimpleAssistantAgent对话其系统提示词为 You are a helpful assistant并装配了一个get_weather工具app_agent.py 通过FunctionTool(get_weather, descriptionGet weather tool.)包装模型有需要时可发起函数调用、执行后再次推理给出自然语言答复。会话启动注册运行时、工具与输出管道启动逻辑集中在cl.on_chat_start装饰的start_chat()中app_agent.py按顺序完成五件事读取配置并创建模型客户端用yaml.safe_load打开model_config.yaml再交给ChatCompletionClient.load_component完成反序列化创建并保存单线程运行时runtime SingleThreadedAgentRuntime()随后cl.user_session.set(run_time, runtime)把它挂到当前用户会话。SingleThreadedAgentRuntime是autogen-core提供的内存级进程内运行时实现见 python/packages/autogen-core/src/autogen_core/_single_threaded_agent_runtime.py消息在同一进程内异步流转无需任何外部基础设施特别适合嵌入 Chainlit 这类 Web 应用创建输出队列queue_stream asyncio.Queue[StreamResult]()同样存入用户会话键queue_stream。它是「Agent 侧生产、UI 侧消费」的桥梁注册业务 Agent调用SimpleAssistantAgent.register(runtime, weather_agent, lambda: ...)。Agent 实例的创建采用了惰性工厂模式lambda以保证运行时在需要时才实例化该类型构造参数model_client_streamTrue开启模型客户端流式输出reflect_on_tool_useTrue让 Agent 在工具执行后再次调用模型、把工具结果整合进最终答复注册 Closure Agent 并启动运行时通过ClosureAgent.register_closure(...)注册聚合输出结果的收尾 Agent详见下文最后runtime.start()让运行时在后台开始投递消息。群聊之外的输入链路starter 与消息处理页面在会话开始时还通过cl.set_starters注入了一组「开场白」快捷按钮app_agent.py如 Greetings 与 Weather后者会把示例消息 Find the weather in New York City. 填入输入框天然触发一次工具调用。用户真正发送消息时走cl.on_message的chat()app_agent.py从会话中取回runtime与queue用asyncio.create_task(runtime.send_message(UserMessage(...), AgentId(weather_agent, default)))把用户文本作为UserMessage发给weather_agent随后进入while True循环消费队列当stream_msg.content是普通str时调用ui_resp.stream_token(...)把文本增量逐字推送到浏览器当收到CreateResult即模型流式结束时的最终结果对象时调用ui_resp.send()落定本条消息并跳出循环。流式输出是如何从模型一路走到浏览器的流式关键在 SimpleAssistantAgent.py 中模型层使用self._model_client.create_stream(messages, tools, ...)SimpleAssistantAgent.py逐块产出数据文本块是str终态对象是CreateResultAgent 的handle_user_messageSimpleAssistantAgent.py遍历这些块凡是str块就await self.runtime.publish_message(StreamResult(contentchunk, sourceself.id.type), topic_idtask_results_topic_id)发布到task-results主题若是工具调用则解析FunctionCall、调用tool.run_json执行并记录FunctionExecutionResultMessage后再调模型做反思轮订阅在task-results主题上的 Closure Agent 收到每个StreamResult后把它写入 Chainlit 会话中的asyncio.Queue见下节UI 侧chat()从同一队列取值并stream_token上屏。值得指出SimpleAssistantAgent在仓库中是被单 Agent 与群聊两个示例共用的通用实现。它内部使用RoutedAgentmessage_handler的消息分发模型同一个类里通过多个 handler 分别处理UserMessage直接对话、GroupChatMessage群聊广播与RequestToSpeak获准发言这也解释了为什么它能同时服务于两种场景。示例二Assistant Critic 双 Agent 群聊app_team.py第二个示例把「团队」接入聊天界面chainlit run app_team.py -hREADME 中以-h参数调用用于展示 Chainlit 命令行帮助正式运行直接执行chainlit run app_team.py即可。团队包含两个 Agent系统提示词分别为app_team.pyAssistantYou are a helpful assistant —— 负责对用户请求给出解答CriticYou are a critic. Provide constructive feedback. Respond with APPROVE if your feedback has been addressed. —— 负责点评 Assistant 的回答直到反馈被采纳并回复APPROVE为止。基于 Topic 的消息路由与订阅app_team.py不使用直接寻址的send_message而是完全基于主题发布/订阅publish/subscribe组织消息流定义了三类业务主题与一个输出主题assistant_topic_type assistant critic_topic_type critic group_chat_topic_type group_chat TASK_RESULTS_TOPIC_TYPE task-results # 输出回 UI 的专用主题注册完 Agent 后代码显式为运行时添加订阅await runtime.add_subscription(TypeSubscription(topic_typeassistant_topic_type, agent_typeassistant_agent_type.type)) await runtime.add_subscription(TypeSubscription(topic_typegroup_chat_topic_type, agent_typeassistant_agent_type.type))即assistant与group_chat两类主题上发布的GroupChatMessage都会被路由给 AssistantCritic 同理订阅critic与group_chatapp_team.py。从 TypeSubscription 源码 可以看到它的语义is_match仅按topic_id.type判断map_to_agent则用主题的source作为 Agent 实例的 key——也就是说TypeSubscription(topic_type..., agent_type...)会让「每个 source 拥有自己的 Agent 实例」天然支持多会话隔离。群聊管理器简单的轮转调度群聊的调度核心是自定义的GroupChatManager(RoutedAgent)app_team.py其handle_message逻辑清晰可读把收到的GroupChatMessage.body一个UserMessage追加到self._chat_history终止条件若发言方是User且内容以approve结尾做了小写化与标点剥离处理向task-results主题发布StreamResult(contentstop, ...)通知 UI 结束本轮若发言方是Critic且回复以approve结尾则发布AssistantMessage(contentTask Finished)表示任务达成轮转发言否则用一个_previous_participant_idx 1并对成员数取模的简单 round-robin 算法选出下一位发言人向其专属主题发布RequestToSpeak()空消息触发该 Agent 生成内容。值得注意的是GroupChatManager把终止判据显式建模在消息语义里Critic 一旦「认可」就结束对话这是样例中群聊收敛的方式读者在自己设计中可替换为更复杂的终止条件。多轮回复的 UI 聚合pass_msg_to_ui群聊中 Assistant 与 Critic 会交替产出多条消息因此app_team.py的 UI 消费逻辑升级为独立的pass_msg_to_ui()协程app_team.py首次拿到文本块时创建一条以发言人名字为前缀的消息ui_resp cl.Message(contentstream_msg.source : )随后的str块持续stream_token追加收到CreateResult表示该 Agent 本轮结束ui_resp.send()把整条消息固化到页面然后重置状态等待下一位发言者收到内容为stop的终止消息或无法识别的负载时break结束本轮群聊循环。而cl.on_message的入口app_team.py则为每次用户发言生成一个随机会话 idsession_id str(uuid.uuid4())把GroupChatMessage(UserMessage(content..., sourceUser))发布到TopicId(typegroup_chat_topic_type, sourcesession_id)主题上然后启动pass_msg_to_ui()消费队列直至结束。用source隔离会话正是上一节TypeSubscription.map_to_agent语义的落地运用。三个关键机制源码级解析1. ClosureAgent用闭包把结果搬出运行时在 AutoGen Core 中定义 Agent 通常要继承RoutedAgent并写 handler 类但如果只是想「收个尾、把消息送出去」可以用更轻量的ClosureAgent。在 ClosureAgent 实现 中可以看到register_closure的完整签名与语义await ClosureAgent.register_closure( runtime, CLOSURE_AGENT_TYPE, output_result, subscriptionslambda: [TypeSubscription(topic_typeTASK_RESULTS_TOPIC_TYPE, agent_typeCLOSURE_AGENT_TYPE)] )闭包签名固定为三个参数(_agent: ClosureContext, message: StreamResult, ctx: MessageContext)。其中第二个参数message的类型注解会被自动解析用于确定该 Agent 能处理的消息类型_closure_agent.py中的get_handled_types_from_closure通过inspect.getfullargspec与get_type_hints完成并为这些类型注册对应的序列化器ClosureContext是运行时句柄协议源码中的Protocol闭包内可用ctx.send_message/ctx.publish_message继续向运行时发消息subscriptions参数传入一个返回订阅列表的延迟求值工厂示例用它把 Closure Agent 挂到task-results主题上——任何 Agent 发布到该主题的StreamResult都会触发闭包output_result闭包本体的作用被极致简化从 Chainlit 会话取出输出队列并await queue.put(message)。这一句把「运行时消息空间」与「UI 消费空间」打通是整个流式管道返程的关键一环。2. StreamResult流式管道的统一信封单 Agent 与群聊场景为何能共用一套 UI 消费逻辑因为 SimpleAssistantAgent.py 用 Pydantic 定义了统一的消息信封class StreamResult(BaseModel): content: str | CreateResult | AssistantMessage source: strcontent有且仅有三种取值UI 侧据此分流str 可上屏的文本增量CreateResult 本轮生成的终结信号AssistantMessage 群聊的结束/汇总消息。source则标记是哪个 Agent 说的供 UI 显示发言人前缀。3. cl.user_sessionChainlit 会话内的状态管理两个示例都遵循同一套状态管理约定runtime、queue_stream等在cl.on_chat_start时初始化并写入cl.user_session在cl.on_message中再取回。这样每个用户会话持有独立的运行时与队列互不串扰cl.user_session的生命周期与 Chainlit 的 web 会话绑定示例通过它优雅地解决了「无状态 Web 请求」与「有状态 Agent 运行时」之间的张力。常见问题与排障思路找不到model_config.yaml程序以相对路径open(model_config.yaml, r)读取因此必须在python/samples/core_chainlit目录下启动chainlit runREADME 也专门提示了这一点或用模板复制出真实配置文件收到RuntimeError(No final model result in streaming mode.)来自 SimpleAssistantAgent.py说明流式调用全程未产生CreateResult终态。通常是模型配置不可用或上游服务异常请先核对model_config.yaml的 key/endpoint消息到了浏览器但迟迟不上屏 / 不落定检查queue的消费分支是否齐全——只有收到CreateResult才会ui_resp.send()只有收到stop终止消息才会退出群聊循环换模型厂商修改model_config.yaml的provider/config即可ChatCompletionClient.load_component会根据provider字段反序列化出对应客户端无需改动任何 Python 代码。总结core_chainlit虽然是一个示例级工程却集中展示了 AutoGen Core 开发交互式 Agent 应用的几条核心范式用SingleThreadedAgentRuntime把运行时嵌进 Web 进程、用ClosureAgent asyncio 队列架起「Agent 结果 → UI 渲染」的流式通道、用TypeSubscription与 Topic 实现基于发布订阅的多 Agent 群聊路由并用 Chainlit 的cl.user_session完成会话级状态管理。对照 app_agent.py、app_team.py 与 SimpleAssistantAgent.py 三个文件逐行阅读再配合 ClosureAgent 与 TypeSubscription 的底层实现即可快速迁移出你自己的 Chainlit AutoGen Core 聊天应用。【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogen创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考