Skip to content

03 蓝图拆分大型Agent项目

上一篇文章最后,我们把所有接口都写在一个app.py里:健康检查、对话、会话管理、错误处理……随着功能增加,这个文件会越来越长,越来越难维护。

你可能会想:把不同功能的路由分到不同文件不就行了?问题是,Flask的@app.route()装饰器需要一个app对象,如果每个文件都import app,就会产生循环引用问题。

Blueprint(蓝图)就是Flask提供的解决方案——先定义路由,后注册到应用。你可以在独立的文件中定义路由,最后在创建应用时把它们组装起来。

打个比方:如果Flask应用是一栋大楼,Blueprint就是每层楼的施工图纸。施工队(路由)在各自的图纸上画好方案,最后交给你统一施工(注册到应用)。

一、创建Blueprint

一个Blueprint就是一个独立的路由模块。创建方式和Flask应用很像:

python
# routes/chat.py
from flask import Blueprint, request

# 创建Blueprint实例
# 第一个参数是蓝图名称,第二个参数通常是__name__
chat_bp = Blueprint("chat", __name__)


@chat_bp.route("/chat", methods=["POST"])
def chat():
    """对话接口"""
    data = request.get_json()
    message = data.get("message", "")
    return {"reply": f"收到: {message}"}


@chat_bp.route("/chat/models")
def list_models():
    """列出可用模型"""
    return {"models": ["deepseek-v4-flash", "deepseek-v3"]}

注意看:这里用的是@chat_bp.route()而不是@app.route()。Blueprint只是记录了"我要注册这些路由",但还没真正生效。

1.1 Blueprint构造参数

python
Blueprint(name, import_name, **options)
参数说明示例
name蓝图名称,用于端点前缀"chat"
import_name通常传__name__,用于定位资源文件__name__
url_prefixURL前缀,所有路由都会加上这个前缀"/api/v1"
template_folder模板文件夹"templates"
static_folder静态文件夹"static"

二、注册Blueprint

定义好Blueprint之后,需要在应用中注册它:

python
# app.py
from flask import Flask
from routes.chat import chat_bp

app = Flask(__name__)

# 注册蓝图
app.register_blueprint(chat_bp)

if __name__ == "__main__":
    app.run(debug=True)

注册之后,chat_bp中定义的所有路由才会真正生效。

2.1 URL前缀

注册时可以指定URL前缀,让所有路由都加上统一的路径:

python
# 所有路由前面自动加上 /api/v1
app.register_blueprint(chat_bp, url_prefix="/api/v1")

这样原来定义的/chat就变成了/api/v1/chat/chat/models变成了/api/v1/chat/models

前缀也可以在创建Blueprint时就指定:

python
chat_bp = Blueprint("chat", __name__, url_prefix="/api/v1")

两种方式效果一样,选哪种看你的习惯。

三、拆分Agent项目

来看一个实际的Agent项目拆分方案:

my-agent-api/
├── app.py                  # 应用入口,组装所有蓝图
├── routes/
│   ├── __init__.py
│   ├── chat.py             # 对话相关接口
│   ├── session.py          # 会话管理接口
│   └── health.py           # 健康检查接口
└── errors.py               # 错误处理

3.1 chat.py — 对话接口

python
# routes/chat.py
from flask import Blueprint, request, abort

chat_bp = Blueprint("chat", __name__)


@chat_bp.route("/chat", methods=["POST"])
def chat():
    """发送消息,获取Agent回复"""
    data = request.get_json()
    if not data or "message" not in data:
        abort(400, description="缺少message字段")

    message = data["message"]
    session_id = data.get("session_id", "default")

    # 后面会替换成真正的Agent调用
    return {
        "reply": f"收到: {message}",
        "session_id": session_id,
    }


@chat_bp.route("/chat/models")
def list_models():
    """列出可用模型"""
    return {
        "models": [
            {"id": "deepseek-v4-flash", "name": "DeepSeek V4 Flash"},
            {"id": "deepseek-v3", "name": "DeepSeek V3"},
        ]
    }

3.2 session.py — 会话管理

python
# routes/session.py
from flask import Blueprint, jsonify

session_bp = Blueprint("session", __name__)


@session_bp.route("/session/<session_id>")
def get_session(session_id):
    """获取会话信息"""
    return {
        "session_id": session_id,
        "messages": [],
        "created_at": "2026-01-01T00:00:00Z",
    }


@session_bp.route("/session/<session_id>", methods=["DELETE"])
def delete_session(session_id):
    """删除会话"""
    return {"deleted": session_id}


@session_bp.route("/sessions")
def list_sessions():
    """列出所有会话"""
    return {"sessions": []}

3.3 health.py — 健康检查

python
# routes/health.py
from flask import Blueprint

health_bp = Blueprint("health", __name__)


@health_bp.route("/health")
def health():
    """健康检查"""
    return {"status": "ok"}


@health_bp.route("/version")
def version():
    """版本信息"""
    return {"version": "1.0.0"}

3.4 errors.py — 错误处理

python
# errors.py
from flask import Flask, jsonify


def register_error_handlers(app: Flask):
    """注册全局错误处理器"""

    @app.errorhandler(400)
    def bad_request(error):
        return jsonify({"error": str(error.description)}), 400

    @app.errorhandler(404)
    def not_found(error):
        return jsonify({"error": "接口不存在"}), 404

    @app.errorhandler(500)
    def internal_error(error):
        return jsonify({"error": "服务器内部错误"}), 500

3.5 app.py — 组装一切

python
# app.py
from flask import Flask
from routes.chat import chat_bp
from routes.session import session_bp
from routes.health import health_bp
from errors import register_error_handlers


def create_app():
    """创建Flask应用"""
    app = Flask(__name__)

    # 注册蓝图
    app.register_blueprint(chat_bp, url_prefix="/api")
    app.register_blueprint(session_bp, url_prefix="/api")
    app.register_blueprint(health_bp)

    # 注册错误处理
    register_error_handlers(app)

    return app


if __name__ == "__main__":
    app = create_app()
    app.run(debug=True)

现在app.py只有十几行,清清楚楚。每个模块各管各的,互不干扰。

最终的API列表:

方法路径来源
GET/healthhealth_bp
GET/versionhealth_bp
POST/api/chatchat_bp
GET/api/chat/modelschat_bp
GET/api/session/<id>session_bp
DELETE/api/session/<id>session_bp
GET/api/sessionssession_bp

四、Blueprint的错误处理

Blueprint可以有自己的错误处理器:

python
chat_bp = Blueprint("chat", __name__)


@chat_bp.errorhandler(429)
def rate_limit_exceeded(error):
    """对话接口的限流错误"""
    return {"error": "请求太频繁,请稍后再试"}, 429

Blueprint的错误处理器只在该蓝图的视图函数中触发。如果蓝图内部没有匹配的处理器,会向上查找应用级别的处理器。

一个实用的做法:按前缀区分错误格式:

python
@app.errorhandler(404)
@app.errorhandler(405)
def handle_api_error(ex):
    if request.path.startswith("/api/"):
        # API接口返回JSON
        return jsonify(error=str(ex)), ex.code
    else:
        # 其他页面返回默认HTML
        return ex

五、Blueprint的URL生成

url_for生成Blueprint中的URL时,需要加上蓝图名前缀:

python
from flask import url_for

# 格式:蓝图名.函数名
url_for("chat.chat")           # → /api/chat
url_for("session.get_session", session_id="abc")  # → /api/session/abc
url_for("health.health")       # → /health

在Blueprint内部可以省略蓝图名,用.函数名

python
@chat_bp.route("/chat")
def chat():
    # 在chat_bp内部跳转到其他端点
    models_url = url_for(".list_models")  # → /api/chat/models
    return {"models_url": models_url}

六、嵌套Blueprint

Blueprint可以注册到另一个Blueprint上,实现更细粒度的模块划分:

python
# 父蓝图
api_bp = Blueprint("api", __name__, url_prefix="/api")

# 子蓝图
chat_bp = Blueprint("chat", __name__)
session_bp = Blueprint("session", __name__)

# 子蓝图注册到父蓝图
api_bp.register_blueprint(chat_bp, url_prefix="/chat")
api_bp.register_blueprint(session_bp, url_prefix="/session")

# 父蓝图注册到应用
app.register_blueprint(api_bp)

最终的URL:

路径说明
/api/chat/chat_bp的/路由
/api/chat/modelschat_bp的/models路由
/api/session/<id>session_bp的/<id>路由

url_for时需要完整的嵌套路径:

python
url_for("api.chat.chat")          # → /api/chat/
url_for("api.session.get_session", session_id="abc")  # → /api/session/abc

这种方式适合API版本管理——v1v2各自是一个父蓝图,下面挂不同的子蓝图。

七、总结

Blueprint是Flask组织大型项目的核心机制:

  • 先定义后注册:在独立文件中定义路由,最后组装到应用
  • URL前缀:统一管理API路径,如/api/v1
  • 模块化:每个Blueprint负责一个功能领域
  • 错误处理:Blueprint可以有自己的错误处理器
  • 嵌套:Blueprint可以嵌套,适合API版本管理

有了Blueprint,你的Agent项目就可以从一个app.py演进成一个结构清晰的多模块项目。

在下一篇文章中,我们将学习流式响应——这是Agent API最重要的特性之一,让用户能看到Agent逐字输出的效果。