先别急着看代码,先把分层关系在脑子里建立起来。下面这张图把三层结构拆得很细:最上层 Flask Route 只处理 HTTP 请求与响应,中间 AgentService 负责业务协调、会话、权限和任务编排,最下层 AgentEngine 则专注提示词、工具调用、模型执行和 Agent 核心逻辑。
![图片[1]-小节二:Flask Route、AgentService、AgentEngine 为什么必须分层?-大赛人网](https://www.dsrw.com/wp-content/uploads/2026/11/image-1592.png)
你顺着箭头看一遍就会发现,这种分层不是为了“看起来更规范”,而是为了让职责边界足够清楚。Route 不要写复杂业务,Engine 不要背 HTTP 细节,Service 负责承上启下。这样后面不管是接 Web、飞书还是定时任务,都能稳很多。
第一次写 Flask Agent,很多人会自然写成:
@app.post("/chat")
def chat():
# 读取请求
# 拼 Context
# 调模型
# Tool Calling
# 执行 Tool
# 更新 Working Memory
# 保存 Session
# 返回结果
...
Demo 没问题。
但 Route 很快会变成几百行。
真正难维护的原因,不是代码行数,而是职责混在一起。
Flask Route:只处理 HTTP
它最适合负责:
读取 JSON
检查必要参数
调用业务 Service
转换 Response
映射 HTTP Status
例如:
message 缺失 → 400
session 不存在 → 404
内部异常 → 500
Route 不应该理解 Main Loop。
AgentService:负责业务协调
Service 可以负责:
获取 / 创建 Session
后面创建 Run
调用 AgentEngine
保存结果
返回统一业务对象
它是网络层和 Runtime 之间的桥。
Service 最好也不要接收 Flask 的 request 对象。否则 CLI 和飞书 Adapter 想复用它时,又要伪造一份 Web 请求。我们可以先在入口处把不同协议转换成普通业务参数:
from dataclasses import dataclass, field
from typing import Any, Dict, Optional
@dataclass
class RunCommand:
message: str
actor_id: str
session_id: Optional[str] = None
metadata: Dict[str, Any] = field(
default_factory=dict
)
class AgentService:
def execute(
self,
command: RunCommand,
) -> dict:
...
Flask、CLI 和机器人只负责构造 RunCommand。身份由可信的鉴权层解析成 actor_id,不能直接相信请求 JSON 里随手传来的用户 ID。这种依赖方向很重要:Web 层依赖 Service,Service 依赖 Engine;Engine 不反过来 import Flask。
AgentEngine:继续负责 Agent 核心
这里仍然是我们已经实现的:
ContextBuilder
LLM Provider
Main Loop
Tool Registry
Tool Executor
State
Working Memory
也就是说,引入 Flask 后,AgentEngine 最好几乎不需要改。
如果为了接一个 Web API,必须把 Agent 全部重写,说明之前边界没有设计好。
一个完整最小 Demo
先安装:
python3 -m pip install "Flask>=3.0,<3.2"
运行:
python3 app.py
然后请求:
curl -X POST \
http://127.0.0.1:5000/api/v1/demo \
-H "Content-Type: application/json" \
-d '{"message":"分析这个项目"}'
响应:
{
"answer": "Agent 已处理任务:分析这个项目",
"status": "COMPLETED"
}
这一节先不正式实现完整:
POST /api/v1/chat
下一节会专门完成。
这里真正要掌握的是:
HTTP
↓
Service
↓
Engine
这条边界。
这里使用 127.0.0.1,表示只允许本机访问,很适合课程实验。换成 0.0.0.0 会监听所有网络接口,不等于服务已经具备公网暴露条件。鉴权、TLS、反向代理、请求大小限制和生产服务器都还没有补齐,别在测试服务刚跑通时就顺手把端口暴露出去。
为什么 AgentService 很重要?
现在看起来:
service.execute()
只是多包了一层。
但后面它会自然承担:
SessionManager
RunManager
Persistence
Authorization Context
Route 不需要知道这些细节。
这就是提前加 Service 层的价值。
课程项目里的最终代码已经继续演进为 RunService,并通过 create_app() 创建 Flask 应用。它比本节最小 Demo 多了异步 Run、SQLite 和 Streaming,是后续章节完成后的形态;阅读时先抓住 Route → Service → Engine 这条依赖方向,不必在这一节一次吃完所有实现。
在课程项目根目录可以直接运行:
python3 projects/06_flask_run_api/app.py
若 5000 端口已被占用,使用 --port 5007 即可切换。端口冲突不属于 Agent 故障,先换端口验证服务,比一上来结束未知系统进程更稳。
















请登录后查看评论内容