# 工作器系统

<cite>
**本文档引用的文件**
- [backend/app/workers/scheduler.py](file://backend/app/workers/scheduler.py)
- [backend/app/workers/citation_engine.py](file://backend/app/workers/citation_engine.py)
- [backend/app/workers/platforms/base.py](file://backend/app/workers/platforms/base.py)
- [backend/app/workers/platforms/kimi.py](file://backend/app/workers/platforms/kimi.py)
- [backend/app/workers/platforms/wenxin.py](file://backend/app/workers/platforms/wenxin.py)
- [backend/app/workers/__init__.py](file://backend/app/workers/__init__.py)
- [backend/app/models/query.py](file://backend/app/models/query.py)
- [backend/app/models/query_task.py](file://backend/app/models/query_task.py)
- [backend/app/models/citation_record.py](file://backend/app/models/citation_record.py)
- [backend/app/main.py](file://backend/app/main.py)
- [backend/app/database.py](file://backend/app/database.py)
- [backend/app/config.py](file://backend/app/config.py)
- [backend/requirements.txt](file://backend/requirements.txt)
- [backend/app/api/queries.py](file://backend/app/api/queries.py)
- [backend/app/api/citations.py](file://backend/app/api/citations.py)
</cite>

## 目录
1. [引言](#引言)
2. [项目结构](#项目结构)
3. [核心组件](#核心组件)
4. [架构总览](#架构总览)
5. [详细组件分析](#详细组件分析)
6. [依赖分析](#依赖分析)
7. [性能考虑](#性能考虑)
8. [故障排查指南](#故障排查指南)
9. [结论](#结论)
10. [附录](#附录)

## 引言
本文件为 GEO 工作器系统的全面架构文档，聚焦以下主题：
- APScheduler 任务调度器的配置与使用：调度策略、并发控制与错误恢复
- 工作器抽象基类（BaseWorker）设计：通用接口、生命周期与状态跟踪
- 平台适配器架构：BasePlatformAdapter 抽象、接口规范与扩展机制
- Kimi 平台适配器实现：Playwright 自动化、页面交互与重试策略
- 文心平台适配器设计：API 封装、响应解析与配置管理
- 工作器注册、启动与停止流程
- 性能监控、资源管理与故障诊断

## 项目结构
后端采用 FastAPI + SQLAlchemy Async + APScheduler 架构，工作器模块位于 backend/app/workers，包含调度器、引用检测引擎与平台适配器；数据库模型位于 backend/app/models，API 路由位于 backend/app/api。

```mermaid
graph TB
subgraph "应用层"
API["FastAPI 应用<br/>lifespan 启停"]
QUERIES["查询 API 路由"]
CITATIONS["引用 API 路由"]
end
subgraph "工作器层"
SCHED["调度器 QueryScheduler"]
ENGINE["引用检测引擎 CitationEngine"]
KIMI["Kimi 适配器"]
WENXIN["文心适配器"]
end
subgraph "数据层"
MODELS["SQLAlchemy 模型<br/>Query/QueryTask/CitationRecord"]
DB["AsyncSessionLocal"]
end
API --> SCHED
API --> QUERIES
API --> CITATIONS
SCHED --> ENGINE
ENGINE --> KIMI
ENGINE --> WENXIN
ENGINE --> MODELS
SCHED --> MODELS
MODELS --> DB
```

图表来源
- [backend/app/main.py:13-22](file://backend/app/main.py#L13-L22)
- [backend/app/workers/scheduler.py:25-40](file://backend/app/workers/scheduler.py#L25-L40)
- [backend/app/workers/citation_engine.py:148-158](file://backend/app/workers/citation_engine.py#L148-L158)
- [backend/app/models/query.py:11-55](file://backend/app/models/query.py#L11-L55)
- [backend/app/models/query_task.py:11-39](file://backend/app/models/query_task.py#L11-L39)
- [backend/app/models/citation_record.py:11-42](file://backend/app/models/citation_record.py#L11-L42)
- [backend/app/database.py:12-18](file://backend/app/database.py#L12-L18)

章节来源
- [backend/app/main.py:13-22](file://backend/app/main.py#L13-L22)
- [backend/app/workers/__init__.py:1-13](file://backend/app/workers/__init__.py#L1-L13)

## 核心组件
- 调度器 QueryScheduler：基于 APScheduler AsyncIOScheduler，每小时扫描并执行到期的查询任务，负责事件循环与异步任务派发。
- 引用检测引擎 CitationEngine：编排平台适配器、品牌匹配、竞争品牌检测与结果持久化，维护 QueryTask 状态与 Query 下次执行时间。
- 平台适配器：BasePlatformAdapter 定义统一接口，KimiAdapter/WenxinAdapter 实现浏览器自动化与响应稳定检测。
- 数据模型：Query（查询）、QueryTask（任务）、CitationRecord（引用记录）三者通过外键关联，配合索引优化查询性能。

章节来源
- [backend/app/workers/scheduler.py:25-95](file://backend/app/workers/scheduler.py#L25-L95)
- [backend/app/workers/citation_engine.py:148-309](file://backend/app/workers/citation_engine.py#L148-L309)
- [backend/app/workers/platforms/base.py:4-18](file://backend/app/workers/platforms/base.py#L4-L18)
- [backend/app/models/query.py:11-55](file://backend/app/models/query.py#L11-L55)
- [backend/app/models/query_task.py:11-39](file://backend/app/models/query_task.py#L11-L39)
- [backend/app/models/citation_record.py:11-42](file://backend/app/models/citation_record.py#L11-L42)

## 架构总览
系统通过 FastAPI 应用在 lifespan 中启动调度器，调度器周期性扫描数据库中到期的查询，调用 CitationEngine 执行跨平台检索与分析，并将结果写入数据库。平台适配器通过 Playwright 控制 Chromium 浏览器进行页面交互，具备指数退避重试与响应稳定性检测。

```mermaid
sequenceDiagram
participant App as "FastAPI 应用"
participant Sched as "QueryScheduler"
participant Engine as "CitationEngine"
participant DB as "数据库"
participant Plat as "平台适配器"
App->>Sched : "启动调度器"
loop 每小时
Sched->>DB : "查询 active 且 next_query_at <= now()"
DB-->>Sched : "查询列表"
loop 遍历查询
Sched->>Engine : "execute_query(query, db)"
Engine->>DB : "创建/更新 QueryTask"
Engine->>Plat : "adapter.query(keyword)"
Plat-->>Engine : "原始响应文本"
Engine->>Engine : "品牌匹配/竞争品牌检测"
Engine->>DB : "写入 CitationRecord"
Engine->>DB : "更新 Query.next_query_at"
end
end
```

图表来源
- [backend/app/main.py:17-21](file://backend/app/main.py#L17-L21)
- [backend/app/workers/scheduler.py:30-90](file://backend/app/workers/scheduler.py#L30-L90)
- [backend/app/workers/citation_engine.py:159-234](file://backend/app/workers/citation_engine.py#L159-L234)
- [backend/app/models/query.py:27-31](file://backend/app/models/query.py#L27-L31)
- [backend/app/models/query_task.py:24-32](file://backend/app/models/query_task.py#L24-L32)
- [backend/app/models/citation_record.py:24-29](file://backend/app/models/citation_record.py#L24-L29)

## 详细组件分析

### 调度器（APScheduler）配置与使用
- 调度策略
  - 使用 AsyncIOScheduler，作业类型为间隔触发（每小时一次），作业 ID 为“check_queries”，名称为“检查并执行到期的查询任务”。
  - 通过 replace_existing=True 确保重复启动时替换旧作业。
- 并发控制
  - 同步包装函数 _run_check 在没有运行中事件循环时使用新事件循环执行；否则在当前事件循环创建任务，避免阻塞。
  - 每次检查独立创建任务，避免阻塞后续调度。
- 错误恢复
  - 单个查询执行失败会记录错误并继续处理下一个查询，不影响整体调度。
  - 关闭时优雅停止调度器并关闭引擎资源。

```mermaid
flowchart TD
Start(["启动调度器"]) --> AddJob["添加间隔作业<br/>每小时触发"]
AddJob --> StartOK["启动成功"]
StartOK --> Loop["每小时执行一次"]
Loop --> CheckDB["查询到期的 Query"]
CheckDB --> ForEach{"是否有待执行查询？"}
ForEach --> |是| ExecOne["_execute_single_query()"]
ForEach --> |否| Sleep["等待下一小时"]
ExecOne --> TryExec["捕获异常并记录"]
TryExec --> Next["继续下一个查询"]
Sleep --> Loop
Next --> Loop
```

图表来源
- [backend/app/workers/scheduler.py:30-90](file://backend/app/workers/scheduler.py#L30-L90)

章节来源
- [backend/app/workers/scheduler.py:25-95](file://backend/app/workers/scheduler.py#L25-L95)

### 引擎（CitationEngine）设计与流程
- 组件职责
  - 维护平台适配器映射（wenxin/kimi）
  - 品牌匹配器（精确/别名/模糊）与竞争品牌检测
  - 任务状态管理（QueryTask）与下次查询时间计算
- 生命周期与状态
  - 每个查询对应多个平台任务，逐个执行并更新 QueryTask 状态（pending → running → success/failed）
  - 成功时写入 CitationRecord，失败时仍写入一条 cited=False 的记录用于占位
- 数据持久化
  - 使用 AsyncSessionLocal 进行事务性读写，提交后刷新对象状态
- 平台扩展
  - 新增平台只需在 platforms 字典中注册适配器实例，无需修改引擎主流程

```mermaid
classDiagram
class CitationEngine {
+platforms : dict
+matcher : BrandMatcher
+competitor_detector : CompetitorDetector
+execute_query(query, db) CitationRecord[]
+execute_single_platform(keyword, platform, target_brand, aliases) dict
+close() void
}
class BrandMatcher {
+target_brand : str
+brand_aliases : str[]
+match(text) dict
}
class CompetitorDetector {
+KNOWN_BRANDS : dict
+detect(text, target_brand) str[]
}
class KimiAdapter {
+platform_name : str
+platform_url : str
+query(keyword) str
+close() void
}
class WenxinAdapter {
+platform_name : str
+platform_url : str
+query(keyword) str
+close() void
}
CitationEngine --> KimiAdapter : "使用"
CitationEngine --> WenxinAdapter : "使用"
CitationEngine --> BrandMatcher : "使用"
CitationEngine --> CompetitorDetector : "使用"
```

图表来源
- [backend/app/workers/citation_engine.py:148-309](file://backend/app/workers/citation_engine.py#L148-L309)
- [backend/app/workers/platforms/kimi.py:11-206](file://backend/app/workers/platforms/kimi.py#L11-L206)
- [backend/app/workers/platforms/wenxin.py:11-205](file://backend/app/workers/platforms/wenxin.py#L11-L205)

章节来源
- [backend/app/workers/citation_engine.py:148-309](file://backend/app/workers/citation_engine.py#L148-L309)

### 平台适配器架构（BasePlatformAdapter）
- 接口规范
  - 必须实现异步 query(keyword) -> str，返回平台原始响应文本
  - 可选实现 close() 清理资源
- 扩展机制
  - 新平台继承 BasePlatformAdapter，设置 platform_name 与 platform_url，实现 query 与可选 close
  - 在 CitationEngine.platforms 中注册实例即可启用

```mermaid
classDiagram
class BasePlatformAdapter {
<<abstract>>
+platform_name : str
+platform_url : str
+query(keyword) str*
+close() void*
}
```

图表来源
- [backend/app/workers/platforms/base.py:4-18](file://backend/app/workers/platforms/base.py#L4-L18)

章节来源
- [backend/app/workers/platforms/base.py:4-18](file://backend/app/workers/platforms/base.py#L4-L18)

### Kimi 平台适配器实现
- 浏览器自动化
  - 使用 Playwright 启动 headless Chromium，设置视口与 UA
  - 导航至平台首页，动态查找输入框与发送按钮，支持多种选择器与回退策略
- 页面交互逻辑
  - 输入关键词后尝试点击发送按钮或按下 Enter 键
  - 等待回复稳定：连续多次检测文本不再变化，超时则返回当前文本
- 错误重试策略
  - 最多重试 3 次，指数退避（2^attempt 秒延迟）
  - 超时与异常转换为 RuntimeError 并记录日志
- 资源管理
  - 每次查询结束后关闭 page/context，保证资源释放

```mermaid
sequenceDiagram
participant Adapter as "KimiAdapter"
participant PW as "Playwright"
participant Page as "Page"
Adapter->>Adapter : "_ensure_browser()"
Adapter->>PW : "启动 Chromium"
Adapter->>Page : "new_context() + new_page()"
Adapter->>Page : "goto(platform_url)"
Adapter->>Page : "查找输入框/发送按钮"
Adapter->>Page : "fill(keyword) / press Enter / click Send"
Adapter->>Adapter : "_wait_for_response_stable()"
Adapter-->>Adapter : "返回稳定文本"
Adapter->>Page : "close()"
Adapter->>PW : "stop()"
```

图表来源
- [backend/app/workers/platforms/kimi.py:21-125](file://backend/app/workers/platforms/kimi.py#L21-L125)
- [backend/app/workers/platforms/kimi.py:126-197](file://backend/app/workers/platforms/kimi.py#L126-L197)

章节来源
- [backend/app/workers/platforms/kimi.py:11-206](file://backend/app/workers/platforms/kimi.py#L11-L206)

### 文心平台适配器设计
- 设计要点
  - 结构与 Kimi 类似，使用 Playwright 控制浏览器，支持多选择器定位输入框与发送按钮
  - 等待回复稳定算法一致，超时返回当前文本
  - 指数退避重试，异常转为 RuntimeError
- 配置管理
  - 通过 Config 设置 Playwright 浏览器路径等参数，适配容器环境

章节来源
- [backend/app/workers/platforms/wenxin.py:11-205](file://backend/app/workers/platforms/wenxin.py#L11-L205)
- [backend/app/config.py:11-13](file://backend/app/config.py#L11-L13)

### 工作器注册、启动与停止流程
- 注册
  - 通过 workers/__init__.py 暴露 CitationEngine、KimiAdapter、WenxinAdapter、QueryScheduler、query_scheduler
- 启动
  - FastAPI lifespan 中调用 query_scheduler.start()，内部注册 APScheduler 作业并启动
- 停止
  - 应用退出时调用 query_scheduler.shutdown()，优雅关闭调度器与引擎资源

```mermaid
sequenceDiagram
participant Main as "main.py"
participant Lifespan as "lifespan"
participant Sched as "QueryScheduler"
participant Engine as "CitationEngine"
Main->>Lifespan : "应用启动"
Lifespan->>Sched : "start()"
Sched->>Sched : "add_job(check_queries)"
Sched->>Engine : "初始化引擎"
Main-->>Lifespan : "yield"
Lifespan->>Sched : "shutdown()"
Sched->>Engine : "close()"
```

图表来源
- [backend/app/main.py:13-22](file://backend/app/main.py#L13-L22)
- [backend/app/workers/scheduler.py:30-90](file://backend/app/workers/scheduler.py#L30-L90)

章节来源
- [backend/app/main.py:13-22](file://backend/app/main.py#L13-L22)
- [backend/app/workers/__init__.py:1-13](file://backend/app/workers/__init__.py#L1-L13)

## 依赖分析
- 外部依赖
  - Web/ASGI：FastAPI、Uvicorn
  - ORM/数据库：SQLAlchemy 2.x、asyncpg、Alembic
  - 任务调度：APScheduler
  - 浏览器自动化：Playwright
  - 配置：Pydantic Settings、python-dotenv
- 内部模块耦合
  - scheduler 依赖 database 与 models，调用 CitationEngine
  - citation_engine 依赖 platforms 子模块与 models
  - platforms 依赖 base 抽象类
  - main 依赖 scheduler 并在 lifespan 中启停

```mermaid
graph LR
REQ["requirements.txt"] --> FAST["FastAPI"]
REQ --> SQL["SQLAlchemy/asyncpg/Alembic"]
REQ --> APS["APScheduler"]
REQ --> PW["Playwright"]
REQ --> PYD["Pydantic Settings"]
MAIN["main.py"] --> SCHED["scheduler.py"]
SCHED --> ENGINE["citation_engine.py"]
ENGINE --> KIMI["platforms/kimi.py"]
ENGINE --> WENXIN["platforms/wenxin.py"]
ENGINE --> MODELS["models/*.py"]
SCHED --> MODELS
MODELS --> DB["database.py"]
```

图表来源
- [backend/requirements.txt:1-35](file://backend/requirements.txt#L1-L35)
- [backend/app/main.py:10-21](file://backend/app/main.py#L10-L21)
- [backend/app/workers/scheduler.py:18-20](file://backend/app/workers/scheduler.py#L18-L20)
- [backend/app/workers/citation_engine.py:13-14](file://backend/app/workers/citation_engine.py#L13-L14)
- [backend/app/models/query.py:11-55](file://backend/app/models/query.py#L11-L55)
- [backend/app/database.py:6-18](file://backend/app/database.py#L6-L18)

章节来源
- [backend/requirements.txt:1-35](file://backend/requirements.txt#L1-L35)

## 性能考虑
- 调度频率与并发
  - 每小时一次的调度频率适合周期性任务；如需更频繁，可调整 APScheduler 触发器
  - 检查过程为异步批处理，单个查询失败不会阻塞其他查询
- 数据库访问
  - 使用 AsyncSessionLocal，开启索引（如 queries.status、queries.next_query_at）提升查询效率
  - 引擎在事务内批量写入，减少往返开销
- 浏览器资源
  - 每次查询后及时关闭 page/context，避免内存泄漏
  - 重试采用指数退避，降低平台限流风险
- 平台扩展
  - 新增平台仅需实现 query/close，通过引擎映射注册，不影响现有流程

## 故障排查指南
- 调度器未启动
  - 确认 lifespan 正常执行，检查日志中“查询调度器已启动”
  - 如无输出，检查 main.py 中 lifespan 注册与 FastAPI 版本兼容性
- 查询未执行
  - 检查 Query.status 是否为 active，next_query_at 是否小于等于当前时间
  - 确认数据库连接字符串与表结构正确
- 平台适配器异常
  - Playwright 未安装：根据日志提示运行安装命令
  - 页面选择器失效：平台 UI 变更导致，需更新选择器策略
  - 超时与不稳定：适当增加等待时间或放宽稳定性阈值
- 引擎写入失败
  - 检查数据库事务提交与异常捕获，确认 CitationRecord 字段完整性
- 资源泄露
  - 确认每次查询后 page/context 已关闭，必要时调用 adapter.close()

章节来源
- [backend/app/workers/scheduler.py:42-90](file://backend/app/workers/scheduler.py#L42-L90)
- [backend/app/workers/platforms/kimi.py:21-32](file://backend/app/workers/platforms/kimi.py#L21-L32)
- [backend/app/workers/platforms/wenxin.py:21-32](file://backend/app/workers/platforms/wenxin.py#L21-L32)
- [backend/app/workers/citation_engine.py:209-228](file://backend/app/workers/citation_engine.py#L209-L228)

## 结论
本系统以 APScheduler 为核心调度器，结合 CitationEngine 的平台编排能力与 Playwright 的浏览器自动化，实现了跨平台的引用检测与分析。通过清晰的抽象与模块化设计，系统具备良好的可扩展性与可维护性。建议在生产环境中进一步完善监控指标、日志分级与平台 UI 变更的自适应策略。

## 附录
- API 路由概览
  - 查询管理：GET/POST/PUT/DELETE /api/v1/queries
  - 引用数据：GET /api/v1/citations 与 GET /api/v1/citations/stats
  - 立即执行：POST /api/v1/queries/{query_id}/run-now
- 数据模型 ER 关系

```mermaid
erDiagram
QUERY {
uuid id PK
uuid user_id FK
string keyword
string target_brand
json brand_aliases
json platforms
string frequency
string status
timestamp last_queried_at
timestamp next_query_at
}
QUERY_TASK {
uuid id PK
uuid query_id FK
string platform
string status
text error_message
timestamp scheduled_at
timestamp started_at
timestamp completed_at
}
CITATION_RECORD {
uuid id PK
uuid query_id FK
string platform
boolean cited
int citation_position
text citation_text
json competitor_brands
text raw_response
timestamp queried_at
}
QUERY ||--o{ QUERY_TASK : "包含"
QUERY ||--o{ CITATION_RECORD : "包含"
```

图表来源
- [backend/app/models/query.py:11-55](file://backend/app/models/query.py#L11-L55)
- [backend/app/models/query_task.py:11-39](file://backend/app/models/query_task.py#L11-L39)
- [backend/app/models/citation_record.py:11-42](file://backend/app/models/citation_record.py#L11-L42)