某公司 AI 后端平台是一个处理社交信息的 Python AI 微服务系统。它需要把一条输入拆成多个可独立演进的步骤：内容解析、候选识别、风险过滤、Agent 分析，再把结果回传给上层应用。单体函数很难同时满足异步处理、可扩展、可追踪和故障隔离，因此系统采用 Monorepo 管理多个 FastAPI API 与 Worker，并用 DAG 和消息队列连接各个节点。
| 层次 | 技术与职责 |
|---|---|
| API | FastAPI，提供任务、配置和调试接口 |
| 编排 | DAG 状态机，保存节点依赖与上下文 |
| 消息 | Kafka 为默认通道，同时兼容 RabbitMQ |
| Worker | 独立消费节点消息，执行 Handler 并发送回调 |
| 数据 | PostgreSQL、SQLAlchemy 2.x 异步会话、Pydantic 模型 |
| 缓存与状态 | Redis，用于幂等租约、任务状态和短期上下文 |
| AI 能力 | Agno Agent、工具调用、结构化输出、模型降级 |
| 观测 | Langfuse Trace、Prometheus 指标、阿里云 ARMS 告警、结构化日志 |
情报分析通常不是一次模型调用，而是多步处理：解析原文、识别显式和隐含标的、查询市场数据、过滤风险候选、执行最终优化。步骤之间存在依赖，但并非每条情报都经过同一条路径。DAG 将流程拆成节点并保存依赖关系，节点只通过上下文和消息通信，从而实现解耦、并行和故障隔离。
原始情报
├─▶ 显式提取 ─┐
├─▶ 账号找币 ──┼─▶ 结果合并 ─▶ 最终优化
└─▶ 图片分析 ─┘环境：Docker
redis:7-alpine+redis-py8.0.1 · Python 3.13
Redis 在 AI Task Backend 里不是主存储，而是性能层。它的定位是：高频读写的临时状态放 Redis，需要持久化、复杂查询的数据落 PostgreSQL。本节把这条边界讲清楚。
用户请求
│
┌────────▼────────┐
│ FastAPI API │
└────────┬────────┘
│
┌──────────────┼──────────────┐
▼ ▼ ▼
PostgreSQL RabbitMQ Redis
(真相来源) (异步任务) (性能层)