面向切面编程：指导 AI 抽离横切逻辑，日志、鉴权、埋点不再重复编写

logger.info("开始处理请求")

if not token:
    raise UnauthorizedError()

start = time.time()
result = await service.do_something()
logger.info(f"执行耗时: {time.time() - start}")
return result

用户接口
  ├─ 鉴权
  ├─ 日志
  ├─ 参数检查
  ├─ 业务逻辑
  └─ 埋点

智能体接口
  ├─ 鉴权
  ├─ 日志
  ├─ 参数检查
  ├─ 业务逻辑
  └─ 埋点

知识库接口
  ├─ 鉴权
  ├─ 日志
  ├─ 参数检查
  ├─ 业务逻辑
  └─ 埋点

  ...

安检
↓
查证件
↓
登记信息
↓
登机

安检
↓
查证件
↓
登记信息
↓
登机

Router A
 ├─ 日志
 ├─ Token 校验
 ├─ 权限校验
 ├─ 业务代码
 └─ 性能统计

Router B
 ├─ 日志
 ├─ Token 校验
 ├─ 权限校验
 ├─ 业务代码
 └─ 性能统计

@Before
@After
@Around

Middleware
依赖注入
Decorator
ContextVar
统一异常处理

core/
├── audit_context.py
├── deps.py
├── logger_config.py
├── security/
│   ├── auth_permission.py
│   ├── jwt_auth.py
│   └── ...
└── service_container.py

请求日志
Request ID
接口耗时
全局审计
链路追踪
统一 Header

@app.middleware("http")
async def middleware(request, call_next):

    # 请求之前
    ...

    response = await call_next(request)

    # 请求之后
    ...

    return response

           Before
              ↓
Request → Middleware → Router
              ↑
            After

request_id
method
path
ip_address
user_id
username
change_count

Middleware
    +
Dependency / Permission
    +
ContextVar

@app.middleware("http")
async def log_requests(request: Request, call_next):
    """Record all HTTP requests and inject request_id into every log line."""

    request_id = str(uuid4())
    request.state.request_id = request_id

    start_time = time.time()

    with logger.contextualize(request_id=request_id):

        logger.info(
            f"📥 {request.method} {request.url.path}"
        )

        response = await call_next(request)
        process_time = time.time() - start_time

        logger.info(
            f"📤 {request.method} {request.url.path} "
            f"- {response.status_code} ({process_time:.3f}s)"
        )

        response.headers["X-Process-Time"] = str(process_time)
        response.headers["X-Request-ID"] = request_id

        return response

@router.get("/users")
async def list_users():

    logger.info("GET /users start")
    start = time.time()

    result = await user_service.list_users()

    logger.info(
        f"GET /users finished: {time.time() - start}"
    )

    return result

@router.get("/users")
async def list_users():
    return await user_service.list_users()

谁访问了？
访问什么接口？
什么时候开始？
什么时候结束？
状态码多少？
耗时多少？
Request ID 是什么？

/user/profile

/admin/users

/admin/users/{id}

Depends(...)

JWT 验证
   ↓
取得用户
   ↓
检查用户状态
   ↓
取得权限集合
   ↓
权限校验

token = request.headers.get("Authorization")
username = verify_token(token)
user = await find_user(username)
permissions = await load_permissions(user.id)

if permission not in permissions:
    raise HTTPException(...)

async def check(
    self,
    user_id: int,
    required: str,
) -> None:

    perms = await self.get_permissions(user_id)

    if SUPER_PERMISSION in perms or required in perms:
        return

    raise_forbidden(
        "auth.permission_denied",
        required=required,
    )

@router.delete("/users/{user_id}")
async def delete_user(...):
    return await user_service.delete(user_id)

需要登录
需要 system:user:delete 权限

@router.delete("/users/{user_id}")
async def delete_user(...):

    # 解析 Authorization
    ...

    # 校验 JWT
    ...

    # 查询用户
    ...

    # 查询权限
    ...

    # 判断 super admin
    ...

    # 写访问日志
    ...

    # 记录开始时间
    ...

    # ===== 终于开始真正业务 =====

    await user_service.delete(user_id)

    # ===== 业务结束 =====

    # 写审计日志
    ...

    # 写耗时日志
    ...

    return ...

class Permission:

    def __init__(
        self,
        permission_code: str,
    ) -> None:
        self._code = permission_code

    async def __call__(
        self,
        request: Request,
        credentials = Depends(_bearer_scheme),
    ) -> int:

        auth = request.app.state.container.auth

        user_id = await auth.resolve_user_id(
            credentials.credentials
        )

        await auth.check(
            user_id,
            self._code
        )

        return user_id

Router
  │
  │ 声明权限
  ↓
Permission("system:user:delete")
  │
  ↓
AuthPermission
  │
  ├── Verify JWT
  ├── Resolve User
  ├── Load Permission
  └── Check Permission

JWT 怎么解析？
用户怎么查？
权限缓存在哪里？
超级管理员怎么判断？

我要 system:user:delete

HTTP Request
     ↓
Router
     ↓
Service
     ↓
Repository
     ↓
Database

request_id
user_id
username

service.run(
    request_id=request_id,
    user_id=user_id,
    ...
)

repository.save(
    request_id=request_id,
    user_id=user_id,
    ...
)

_audit_context: ContextVar[
    Optional[AuditRequestContext]
] = ContextVar(
    "audit_request_context",
    default=None
)

begin_audit_context(...)

reset_audit_context(...)

set_audit_user(
    user.id,
    user.username
)

            HTTP Request
                 ↓
           Middleware
                 │
                 ├── request_id
                 ├── method
                 ├── path
                 └── ip
                 ↓
           Audit Context
                 ↓
            AuthPermission
                 │
                 ├── user_id
                 └── username
                 ↓
           Audit Context
                 ↓
        Service / Repository

100 个接口

100 份日志代码
100 份鉴权代码
100 份耗时代码
100 份审计代码

request_id
client_ip
user_id
latency

修改成本 ≈ 200 次

修改成本 ≈ 1 次

接收什么请求
调用什么 Service
返回什么结果

token
logger
permission
cache
metrics
audit

if permission not in permissions:

if required_permission not in permissions:

if required not in permissions:

## 横切关注点

日志、认证、鉴权、审计、指标统计、链路追踪和异常处理属于横切关注点。

禁止在 Router 或业务 Service 中重复实现这些逻辑。

统一规则：

- 全局 HTTP 请求逻辑使用 Middleware；
- 身份认证和权限校验优先使用 FastAPI Depends；
- 只有依赖注入不合适时才使用 Decorator；
- 请求级上下文使用 ContextVar；
- 异常和响应转换使用全局异常处理；

Router 只负责请求编排；
Service 只负责业务逻辑；
Repository 只负责数据访问。

请写高质量代码

请为该接口增加权限控制。

要求：

1. 不允许在 Router 中解析 JWT；
2. 不允许重新实现权限判断；
3. 使用项目已有的 AuthPermission；
4. 权限检查通过 FastAPI Depends 或现有 Permission 机制完成；
5. Router 只声明 required permission；
6. 复用 ServiceContainer 中现有 auth 单例；
7. 不改变现有统一异常体系。

给每个接口增加访问日志。

logger.info(...)

需要增加 HTTP 请求访问日志。

请先判断该功能是否属于横切关注点。

要求：

1. 不允许逐个修改 Router；
2. 使用 FastAPI Middleware 统一实现；
3. 每个请求生成 request_id；
4. 记录 method、path、status_code 和 latency；
5. response 返回 X-Request-ID；
6. 日志上下文应自动传播到当前请求中的其他模块；
7. 不污染 Service 和 Repository 的业务参数。

用户点击
Agent 调用次数
LLM Token 消耗
接口耗时
知识库命中率
工具成功率

metrics.inc("agent_call")
logger.info(...)
tracer.start(...)
audit.record(...)
result = await agent.run()
metrics.observe(...)
logger.info(...)
tracer.end(...)

result = await agent.run()

Metrics
Logging
Tracing
Audit
   │
   │ 横向切入
   ↓
────────────────────
 Router → Service
────────────────────

Router
 ↓
Metrics
 ↓
Logging
 ↓
Tracing
 ↓
Auth
 ↓
Audit
 ↓
Business

Middleware
    +
Dependency Injection
    +
Decorator
    +
ContextVar
    +
Global Exception Handler

每个文件单独看都挺漂亮。

整个项目放在一起，仍然像一个人设计出来的。