企业官网建设流程全解析

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

【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogen

core_chainlit是 AutoGen Python 仓库中官方提供的一个端到端示例（目录位于 python/samples/core_chainlit），演示如何用AutoGen Core（autogen-core）的单线程 Agent 运行时，对接Chainlit网页聊天框架，实现「用户浏览器提问 → Agent 后台推理 → token 实时流式回显」的完整链路。读完本文你将掌握：如何为单个 AssistantAgent 搭建可流式输出的聊天页，如何把「Assistant + Critic」两个 Agent 组成的群聊团队接入同一界面，以及如何借助ClosureAgent、消息 Topic 与cl.user_session在 Chainlit 会话内管理运行时与输出队列。

示例组成与运行机制概览

根据示例的 README，core_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 及以上版本，并安装以下依赖（安装命令来自 README）：

pip install -U chainlit autogen-core autogen-ext[openai] pyyaml

说明：

• chainlit：负责把 Python 函数包装成可在浏览器中交互的聊天 UI；
• autogen-core：AutoGen 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. OpenAI（API Key）

provider: autogen_ext.models.openai.OpenAIChatCompletionClient config: model: gpt-4o api_key: REPLACE_WITH_YOUR_API_KEY

2. Azure OpenAI（API Key）

provider: 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_KEY

3. Azure OpenAI（Entra / AD Token）

provider: 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.py

chainlit run会启动 Chainlit 自带的服务并自动打开网页聊天窗口。第一个消息界面直接与一个SimpleAssistantAgent对话，其系统提示词为 "You are a helpful assistant"，并装配了一个get_weather工具（app_agent.py 通过FunctionTool(get_weather, description="Get weather tool.")包装），模型有需要时可发起函数调用、执行后再次推理给出自然语言答复。

会话启动：注册运行时、工具与输出管道

启动逻辑集中在@cl.on_chat_start装饰的start_chat()中（app_agent.py），按顺序完成五件事：

1. 读取配置并创建模型客户端：用yaml.safe_load打开model_config.yaml，再交给ChatCompletionClient.load_component完成反序列化；
2. 创建并保存单线程运行时：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 应用；
3. 创建输出队列：queue_stream = asyncio.Queue[StreamResult]()，同样存入用户会话键"queue_stream"。它是「Agent 侧生产、UI 侧消费」的桥梁；
4. 注册业务 Agent：调用SimpleAssistantAgent.register(runtime, "weather_agent", lambda: ...)。Agent 实例的创建采用了惰性工厂模式（lambda），以保证运行时在需要时才实例化该类型；构造参数model_client_stream=True开启模型客户端流式输出，reflect_on_tool_use=True让 Agent 在工具执行后再次调用模型、把工具结果整合进最终答复；
5. 注册 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，终态对象是CreateResult；
• Agent 的handle_user_message（SimpleAssistantAgent.py）遍历这些块，凡是str块就await self.runtime.publish_message(StreamResult(content=chunk, source=self.id.type), topic_id=task_results_topic_id)发布到task-results主题；若是工具调用，则解析FunctionCall、调用tool.run_json执行并记录FunctionExecutionResultMessage后再调模型做反思轮；
• 订阅在task-results主题上的 Closure Agent 收到每个StreamResult后，把它写入 Chainlit 会话中的asyncio.Queue（见下节）；
• UI 侧chat()从同一队列取值并stream_token上屏。

值得指出：SimpleAssistantAgent在仓库中是被单 Agent 与群聊两个示例共用的通用实现。它内部使用RoutedAgent+@message_handler的消息分发模型，同一个类里通过多个 handler 分别处理UserMessage（直接对话）、GroupChatMessage（群聊广播）与RequestToSpeak（获准发言），这也解释了为什么它能同时服务于两种场景。

示例二：Assistant + Critic 双 Agent 群聊（app_team.py）

第二个示例把「团队」接入聊天界面：

chainlit run app_team.py -h

（README 中以-h参数调用，用于展示 Chainlit 命令行帮助；正式运行直接执行chainlit run app_team.py即可。）

团队包含两个 Agent，系统提示词分别为（app_team.py）：

• Assistant："You are a helpful assistant" —— 负责对用户请求给出解答；
• Critic："You 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_type=assistant_topic_type, agent_type=assistant_agent_type.type)) await runtime.add_subscription(TypeSubscription(topic_type=group_chat_topic_type, agent_type=assistant_agent_type.type))

即assistant与group_chat两类主题上发布的GroupChatMessage都会被路由给 Assistant；Critic 同理订阅critic与group_chat（app_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逻辑清晰可读：

1. 把收到的GroupChatMessage.body（一个UserMessage）追加到self._chat_history；
2. 终止条件：若发言方是User且内容以approve结尾（做了小写化与标点剥离处理），向task-results主题发布StreamResult(content="stop", ...)，通知 UI 结束本轮；若发言方是Critic且回复以approve结尾，则发布AssistantMessage(content="Task Finished")表示任务达成；
3. 轮转发言：否则用一个_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(content=stream_msg.source + ": ")）；
• 随后的str块持续stream_token追加；
• 收到CreateResult表示该 Agent 本轮结束，ui_resp.send()把整条消息固化到页面，然后重置状态等待下一位发言者；
• 收到内容为"stop"的终止消息或无法识别的负载时break，结束本轮群聊循环。

而@cl.on_message的入口（app_team.py）则为每次用户发言生成一个随机会话 id（session_id = str(uuid.uuid4())），把GroupChatMessage(UserMessage(content=..., source="User"))发布到TopicId(type=group_chat_topic_type, source=session_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, subscriptions=lambda: [TypeSubscription(topic_type=TASK_RESULTS_TOPIC_TYPE, agent_type=CLOSURE_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: str

content有且仅有三种取值，UI 侧据此分流：str= 可上屏的文本增量；CreateResult= 本轮生成的终结信号；AssistantMessage= 群聊的结束/汇总消息。source则标记是哪个 Agent 说的，供 UI 显示发言人前缀。

3. cl.user_session：Chainlit 会话内的状态管理

两个示例都遵循同一套状态管理约定：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 run（README 也专门提示了这一点），或用模板复制出真实配置文件；
• 收到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