企业官网建设流程全解析

使用 FastAPI 构建 AutoGen AgentChat Web 聊天应用：单 Agent 与多 Agent 团队完整实战

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

导读

本文将基于仓库 python/samples/agentchat_fastapi 示例目录，完整讲解如何用 FastAPI 把 Microsoft AutoGen 的 AgentChat 封装为浏览器可用的 Web 聊天服务：既包含单个AssistantAgent的 HTTP 对话接口，也包含由多个智能体组成、经 WebSocket 双向通信的RoundRobinGroupChat多智能体团队。你将掌握 AgentChat 中AssistantAgent、带自定义 WebSocket 输入函数的UserProxyAgent、RoundRobinGroupChat的编排方式，以及通过save_state/load_state实现跨服务重启的会话状态持久化，最终能够动手复现并在浏览器中运行你自己的智能体聊天页面。

一、示例概览与核心特性

该示例演示的是一个"最小但完整"的对话式 AI 应用，其用到的 AgentChat 核心特性非常具有代表性：

示例目录结构如下：

python/samples/agentchat_fastapi/ ├── README.md # 示例说明（本文主文档） ├── model_config_template.yaml# 模型配置模板（含 OpenAI / Azure OpenAI 多种认证方式） ├── app_agent.py # 单 Agent 聊天服务（端口 8001） ├── app_agent.html # 单 Agent 聊天前端页面 ├── app_team.py # 多 Agent 团队聊天服务（端口 8002，WebSocket） └── app_team.html # 团队聊天前端页面

需要特别强调的是，示例对模型无关配置、WebSocket 人机交互与文件化状态持久化三件事做了很好的示范，这三件事正是把 AgentChat 从命令行脚本升级为可用 Web 产品的关键。

二、环境准备与依赖安装

示例的运行依赖 AgentChat、模型扩展包与 Web 服务框架。在包含 python/pyproject.toml 的仓库环境中安装命令如下（也等价于在任何独立项目环境执行）：

pip install -U "autogen-agentchat" "autogen-ext[openai]" "fastapi" "uvicorn[standard]" "PyYAML"

逐项拆解：

• autogen-agentchat：提供AssistantAgent、UserProxyAgent、RoundRobinGroupChat、save_state/load_state等高层 AgentChat API；
• autogen-ext[openai]：提供 OpenAI / Azure OpenAI 模型客户端（示例读取模型配置文件时的 provider 解析即依赖该扩展）；
• fastapi与uvicorn[standard]：提供 HTTP + WebSocket 服务能力（standard额外包含 WebSocket 所需的websockets等依赖）；
• PyYAML：用于解析model_config.yaml模型配置文件。

说明：示例默认使用 OpenAI 系模型。如需使用其他模型（如本地 Ollama、Gemini 等），应选用对应的autogen-ext扩展并提供对应的模型配置，同时模型客户端需实现统一的ChatCompletionClient接口（其接口定义见 python/packages/autogen-core）。

三、模型配置：model_config.yaml

app_agent.py与app_team.py都在运行时读取同目录下的model_config.yaml，再通过ChatCompletionClient.load_component(model_config)以组件化声明式配置的方式创建模型客户端（见 app_agent.py 的get_agent函数）。

注意：示例代码中使用的是相对路径"model_config.yaml"，因此必须把配置文件放在与启动脚本相同的目录（即示例目录python/samples/agentchat_fastapi内）。

仓库已提供可直接拷贝改写的模板 python/samples/agentchat_fastapi/model_config_template.yaml，请将拷贝后的文件命名为model_config.yaml。模板实际支持三种配置方式：

1. OpenAI API Key

# Use Open AI with key provider: autogen_ext.models.openai.OpenAIChatCompletionClient config: model: gpt-4o api_key: REPLACE_WITH_YOUR_API_KEY

2. Azure OpenAI（API Key 认证）

# Use Azure Open AI with 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（Azure AD Token 认证）

# Use Azure OpenAI with AD token provider. 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

各参数作用速查：

配置文件最终通过yaml.safe_load解析为字典，load_component会依据provider字段自动实例化对应组件，因此应用代码无需关心底层是 OpenAI 还是 Azure OpenAI——这正是该示例所体现的"模型无关"设计。

四、单 Agent 聊天服务：app_agent.py

启动命令（在示例目录内执行）：

python app_agent.py

浏览器访问 http://localhost:8001 即可开始对话。该服务基于HTTP POST实现一问一答，不依赖 WebSocket。

4.1 整体代码结构

服务端 app_agent.py 的关键组成如下：

from autogen_agentchat.agents import AssistantAgent from autogen_agentchat.messages import TextMessage from autogen_core import CancellationToken from autogen_core.models import ChatCompletionClient from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles app = FastAPI() # 允许跨域访问 app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 挂载当前目录为静态资源 app.mount("/static", StaticFiles(directory="."), name="static")

其中：

• 全开放的 CORS 配置仅为本地演示方便；生产环境应收紧allow_origins；
• /static使前端页面能加载同目录静态资源；
• 根路径/直接返回app_agent.html页面文件。

4.2 状态文件与 Agent 构建

model_config_path = "model_config.yaml" state_path = "agent_state.json" history_path = "agent_history.json" async def get_agent() -> AssistantAgent: """Get the assistant agent, load state from file.""" async with aiofiles.open(model_config_path, "r") as file: model_config = yaml.safe_load(await file.read()) model_client = ChatCompletionClient.load_component(model_config) agent = AssistantAgent( name="assistant", model_client=model_client, system_message="You are a helpful assistant.", ) if not os.path.exists(state_path): return agent # 首次运行时无状态文件，直接返回 async with aiofiles.open(state_path, "r") as file: state = json.loads(await file.read()) await agent.load_state(state) return agent

这一段揭示了三个重要设计：

1. Agent 每次请求都会被重建：由于模型客户端与 Agent 均为轻量对象，接口每收到一次消息就调用get_agent()重新构建，保证无共享可变状态；
2. 模型配置声明式加载：ChatCompletionClient.load_component负责根据 YAML 的provider实例化模型客户端；
3. 会话记忆来自状态文件而非常驻对象：若agent_state.json已存在，则调用agent.load_state(state)把历史对话上下文恢复到 Agent 内部——这正是第 7 节状态持久化机制的关键。

4.3 历史记录读取接口

@app.get("/history") async def history() -> list[dict[str, Any]]: try: return await get_history() except Exception as e: raise HTTPException(status_code=500, detail=str(e)) from e

get_history()从agent_history.json读取浏览器端展示所需的全部消息记录；文件不存在时返回空列表，保证前端页面刷新后仍能看到历史。

4.4 对话接口 /chat

@app.post("/chat", response_model=TextMessage) async def chat(request: TextMessage) -> TextMessage: try: agent = await get_agent() response = await agent.on_messages(messages=[request], cancellation_token=CancellationToken()) # 每一轮对话后保存 Agent 状态到文件 state = await agent.save_state() async with aiofiles.open(state_path, "w") as file: await file.write(json.dumps(state)) # 追加本次请求与回复到历史记录 history = await get_history() history.append(request.model_dump()) history.append(response.chat_message.model_dump()) async with aiofiles.open(history_path, "w") as file: await file.write(json.dumps(history)) assert isinstance(response.chat_message, TextMessage) return response.chat_message except Exception as e: error_message = { "type": "error", "content": f"Error: {str(e)}", "source": "system", } raise HTTPException(status_code=500, detail=error_message) from e

请求与响应直接复用 AgentChat 的TextMessage作为 FastAPI 的 Pydantic 模型，是整个接口最巧妙的一点：

• 前端POST /chat发送{"content": "...", "source": "user"}，FastAPI 自动反序列化为TextMessage；
• agent.on_messages(messages=[request], cancellation_token=CancellationToken())完成一次推理，参数语义为"本次只传入新增消息，Agent 内部自行维护历史上下文"（这一点在AssistantAgent的类注释中有明确强调，见 python/packages/autogen-agentchat/src/autogen_agentchat/agents/_assistant_agent.py）；
• response.chat_message是 Agent 返回的最终回复消息，再以TextMessage形式直接序列化为 HTTP 响应；
• 返回HTTPException(500)时附带结构化的error_message，前端据此以"错误气泡"呈现。

4.5 服务入口

if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8001)

绑定0.0.0.0便于局域网访问，端口为8001。

五、单 Agent 前端页面：app_agent.html

app_agent.html 是一个无构建依赖的单文件页面，逻辑核心集中在 JS 部分：

1. 发送消息：点击 Send 或按回车触发sendMessage()，向http://localhost:8001/chat发起POST请求，消息体为{ content: message, source: 'user' }；发送期间禁用输入框与按钮；
2. 渲染回复：response.ok时以assistant样式展示data.content；失败时（后端返回的detail.type === 'error'）以红色错误气泡展示错误内容；
3. 历史回放：window.onload = loadHistory，页面加载时请求GET /history并把已有消息逐条渲染。

由此形成了"请求/响应 + 文件历史"的简洁单 Agent 交互闭环。注意：页面中的接口地址是写死的http://localhost:8001，若服务部署在其他主机需同步修改。

六、多 Agent 团队聊天服务：app_team.py

启动命令（在示例目录内执行）：

python app_team.py

浏览器访问 http://localhost:8002。团队聊天与单 Agent 模式最大的差异在于：

1. 通信方式升级为WebSocket（/ws/chat），因为多轮团队协作期间服务端需要随时"暂停"并向浏览器请求用户输入，HTTP 一问一答难以表达这种长连接交互；
2. 加入了第二个AssistantAgent与一个UserProxyAgent，组成RoundRobinGroupChat团队。

6.1 团队构建：两个 Assistant + 一个用户代理

app_team.py 中get_team构建了三位参与者：

agent = AssistantAgent( name="assistant", model_client=model_client, system_message="You are a helpful assistant.", ) yoda = AssistantAgent( name="yoda", model_client=model_client, system_message="Repeat the same message in the tone of Yoda.", ) user_proxy = UserProxyAgent( name="user", input_func=user_input_func, # 使用传入的自定义用户输入函数 ) team = RoundRobinGroupChat([agent, yoda, user_proxy])

yoda智能体模拟"尤达大师"的说话口吻来复述消息，示例因此直观展示多个大模型智能体之间的角色差异。在服务端加载状态的部分与单 Agent 相同：若team_state.json存在则await team.load_state(state)。

6.2 团队状态保存的内部结构

团队层状态的保存比单 Agent 更复杂。RoundRobinGroupChat的save_state最终由基类 python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_base_group_chat.py 实现为嵌套字典，其结构为：

{ "agent_states": { "assistant": {...}, // 每个 Agent 各自保存的状态 "yoda": {...}, "user": {...}, "RoundRobinGroupChatManager": {...} // 团队管理器的消息线程与轮次信息 } }

其中参与者各自的状态（即AssistantAgentState）由 AssistantAgent.save_state 负责：它把内部的model_context（模型对话上下文，即历史 LLM 消息）序列化后返回；各状态类型定义可参见 python/packages/autogen-agentchat/src/autogen_agentchat/state/_states.py。

6.3 WebSocket 端点与自定义用户输入函数

团队模式下用户消息通过 WebSocket 通道收发，其端点为/ws/chat。核心代码如下：

@app.websocket("/ws/chat") async def chat(websocket: WebSocket): await websocket.accept() # 团队使用的用户输入函数 async def _user_input(prompt: str, cancellation_token: CancellationToken | None) -> str: try: data = await websocket.receive_json() message = TextMessage.model_validate(data) return message.content except WebSocketDisconnect: logger.info("Client disconnected while waiting for user input") raise try: while True: data = await websocket.receive_json() request = TextMessage.model_validate(data) try: team = await get_team(_user_input) history = await get_history() stream = team.run_stream(task=request) async for message in stream: if isinstance(message, TaskResult): continue # 流结束标记，跳过 await websocket.send_json(message.model_dump()) if not isinstance(message, UserInputRequestedEvent): history.append(message.model_dump()) # 不保存“请求用户输入”事件 async with aiofiles.open(state_path, "w") as file: state = await team.save_state() await file.write(json.dumps(state)) async with aiofiles.open(history_path, "w") as file: await file.write(json.dumps(history)) except WebSocketDisconnect: break except Exception as e: error_message = {"type": "error", "content": f"Error: {str(e)}", "source": "system"} try: await websocket.send_json(error_message) # 出错后重新启用输入框 await websocket.send_json({ "type": "UserInputRequestedEvent", "content": "An error occurred. Please try again.", "source": "system", }) except WebSocketDisconnect: break ...

这一段隐藏了UserProxyAgent人机协作的全部精髓，可从源码层面拆解：

为什么需要自定义输入函数？查看 python/packages/autogen-agentchat/src/autogen_agentchat/agents/_user_proxy_agent.py 可知：UserProxyAgent通过input_func获取用户输入，未显式传入时使用默认的cancellable_input——它调用标准输入input()（见该文件cancellable_input定义），这只适合终端。示例为它注入的_user_input则改成从 WebSocket 接收 JSON 并返回其中的文本内容，从而把"等待用户在浏览器输入"接入团队执行流。UserProxyAgent的异步输入函数签名类型为Callable[[str, Optional[CancellationToken]], Awaitable[str]]。

团队何时"等待"用户？从UserProxyAgent.on_messages_stream的实现（见上文同一文件源码）可以看到：它在收到消息后先yield一个UserInputRequestedEvent事件，再调用input_func阻塞等待。UserInputRequestedEvent被team.run_stream作为流中消息发送出来，服务端将其send_json到浏览器——前端正是以收到该事件作为"轮到我输入"的信号去启用输入框。

为什么 run_stream 要跳过 TaskResult？team.run_stream(task=request)是异步生成器，会依次产出团队执行过程中产生的事件/消息，最后产出一个TaskResult作为收尾。TaskResult并非需要逐条展示的聊天消息，因此代码continue跳过它；同时为了避免把"请求用户输入"这类控制事件混入聊天历史，代码只把非UserInputRequestedEvent的消息追加到历史文件中。

状态保存时机：整个团队跑完一轮（包括用户输入）后，立即调用team.save_state()与历史写入，确保任何时刻异常中断都有最近一份完整状态可恢复。

6.4 团队前端页面：app_team.html

app_team.html 与单 Agent 页面相比多了 WebSocket 管理逻辑：

• 页面加载时建立new WebSocket('ws://localhost:8002/ws/chat')；
• ws.onmessage中依据message.type分支处理：
  • UserInputRequestedEvent→enableInput()启用输入框与发送按钮（即"轮到用户了"）；
  • error→ 展示错误并enableInput()；
  • 其他消息 → 按message.source展示气泡；
• 用户发送消息时disableInput()并ws.send(...)，此后直到再次收到UserInputRequestedEvent前输入框一直禁用；
• ws.onclose/ws.onerror时提示用户刷新页面。

前后端配合便形成了 README 描述的完整交互节奏：团队采用轮询（round-robin）策略，每个智能体轮流发言；轮到用户时输入框启用，用户发送后输入框立即禁用、智能体们继续轮流回复。

七、状态持久化：save_state / load_state 深入解析

7.1 文件约定

两个应用分别管理两组 JSON 文件：

两者的职责完全不同：

• 状态文件（*_state.json）：保存 Agent / Team 的完整运行上下文，供服务重启后恢复记忆与对话轮次；
• 历史文件（*_history.json）：保存纯消息列表，仅用于浏览器刷新后重绘聊天界面。

7.2 底层实现原理

Agent 层：AssistantAgent.save_state（见 python/packages/autogen-agentchat/src/autogen_agentchat/agents/_assistant_agent.py）把内部模型上下文序列化为AssistantAgentState；load_state则把序列化结果写回模型上下文。这意味着Agent 的"记忆"即它的消息上下文，持久化上下文即可在重启后无感续聊。

Team 层：RoundRobinGroupChat由RoundRobinGroupChatManager驱动（源码见 python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_round_robin_group_chat.py）。管理器在save_state中记录三样东西：团队共享的message_thread（消息线程）、current_turn（当前轮次）、next_speaker_index（下一位发言者在参与者列表中的下标）；其select_speaker方法正是通过(current + 1) % len(participants)实现轮询选择。而团队基类的save_state会把每个参与者与管理器各自的上述状态汇总为一个以名字为键的TeamState（对应状态类定义见 python/packages/autogen-agentchat/src/autogen_agentchat/state/_states.py）。

因此，当服务端重启后执行：

if not os.path.exists(state_path): return team # 或 return agent：首次运行 async with aiofiles.open(state_path, "r") as file: state = json.loads(await file.read()) await team.load_state(state) # 恢复消息线程、轮次、下一位发言者及各 Agent 上下文

团队即可从中断处继续——既保留了所有 Agent 的上下文记忆，也保留了"该轮到谁发言"的执行位置。

7.3 如何在运行时观察状态

README 明确建议：与服务对话几轮后，直接查看上述 JSON 文件即可直观理解内部状态。例如agent_state.json中可看到模型上下文中累积的 user/assistant 消息；team_state.json中则是以agent_states为键、各参与者状态与RoundRobinGroupChatManager消息线程为值的嵌套结构。它们是理解 AgentChat 状态机的最佳"教学标本"。

八、运行与验证清单

按顺序操作即可跑通两个应用：

1. 在示例目录 python/samples/agentchat_fastapi 内安装依赖并创建model_config.yaml（参照 model_config_template.yaml，填入有效的 API Key）；
2. 终端 A 执行python app_agent.py，打开 http://localhost:8001，体验单 Agent 对话；
3. 终端 B 执行python app_team.py，打开 http://localhost:8002，体验"assistant → yoda → 用户"轮询团队对话（注意观察：轮到用户时输入框自动启用，消息发出后禁用直到再次轮到用户）；
4. 对话几轮后，检查同目录生成的agent_state.json/agent_history.json（单 Agent）与team_state.json/team_history.json（团队）；
5. 重启服务后再次访问页面，历史消息会自动回放（来自历史文件），并且 Agent/团队仍记得重启前的上下文（来自状态文件），可让智能体引用"之前提到过"的信息来验证记忆恢复。

九、小结

通过agentchat_fastapi示例，可以看到 AutoGen AgentChat 与 Web 框架结合的三条通用范式：

1. HTTP 简单问答：复用TextMessage作为请求/响应模型，on_messages处理单轮推理，适合无状态、单智能体场景；
2. WebSocket 长连接协作：UserProxyAgent的input_func是连接"大模型执行流"与"浏览器用户"的桥；run_stream把团队内部的各类事件实时推送前端，UserInputRequestedEvent驱动人机轮询的 UI 状态机；
3. 文件化状态持久化：save_state/load_state在每次交互后落盘、在每次请求前恢复，配合历史文件实现重启无感续聊。

若要继续深入，可研读以下仓库文件：Agent 与用户代理的实现见 python/packages/autogen-agentchat/src/autogen_agentchat/agents/_assistant_agent.py 与 python/packages/autogen-agentchat/src/autogen_agentchat/agents/_user_proxy_agent.py；团队编排见 python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_round_robin_group_chat.py；状态结构见 python/packages/autogen-agentchat/src/autogen_agentchat/state/_states.py。以本示例为起点，你可以继续为其补充工具调用（tools）、终止条件（termination_condition）等 AgentChat 能力，把它扩展成更完整的 Agent 应用。

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