小节二:Flask Route、AgentService、AgentEngine 为什么必须分层?

先别急着看代码,先把分层关系在脑子里建立起来。下面这张图把三层结构拆得很细:最上层 Flask Route 只处理 HTTP 请求与响应,中间 AgentService 负责业务协调、会话、权限和任务编排,最下层 AgentEngine 则专注提示词、工具调用、模型执行和 Agent 核心逻辑。

图片[1]-小节二:Flask Route、AgentService、AgentEngine 为什么必须分层?-大赛人网

你顺着箭头看一遍就会发现,这种分层不是为了“看起来更规范”,而是为了让职责边界足够清楚。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 故障,先换端口验证服务,比一上来结束未知系统进程更稳。

© 版权声明
THE END
喜欢就支持一下吧
点赞8 分享
评论 抢沙发
头像-大赛人网
欢迎您留下宝贵的见解!
提交
头像-大赛人网

昵称

取消
昵称

    请登录后查看评论内容