张胜宇
e239eb778b
1) 客服 Agent 四份交付文档 + 构建脚手架：品牌由包装占位 XX科技 / 旧名 南方财富 统一为南方基金（热线 400-889-8899 / 官网 nffund.com），系统名改为「智能服务系统」； 同步追加 §0.4 修订记录行，工程记录行保留原占位字面以支撑硬编码扫描验收。 2) 开发文档：清理 28 份已作废/残留文档（14 份移出归档 + 14 份仓库副本）， 新增《文档规整方案与开发前待决事项-2026-09-17》。 3) 客服agent 四份交付文档首次纳入本分支。
114 lines
4.1 KiB
Python
114 lines
4.1 KiB
Python
"""只读探查：Worker 的运行状态（队列有没有在消费）。
|
||
|
||
只做 SELECT，结果写 `docs/evidence/worker-state.json`：
|
||
|
||
python tools/probe_worker_state.py
|
||
|
||
三个问题：
|
||
1. `domain_event_outbox` 里各状态各有多少条、最老的一条积压了多久
|
||
—— Worker 没在跑，事件就永远停在 pending；
|
||
2. `agent_run` 里 queued/running 有多少
|
||
—— Worker 没在跑，用户发起的对话就一直"排队中"；
|
||
3. `memory_episode` 有没有待提取的片段（同样是 Worker 负责消费）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from sqlalchemy import func, select
|
||
|
||
from app.infrastructure.db import SessionFactory
|
||
from app.model.platform import AgentRun, DomainEventOutbox
|
||
|
||
OUTPUT = Path("docs/evidence/worker-state.json")
|
||
|
||
|
||
async def _grouped(session: Any, model: Any, column_name: str) -> dict[str, int]:
|
||
column = getattr(model, column_name, None)
|
||
if column is None:
|
||
return {}
|
||
rows = await session.execute(select(column, func.count()).group_by(column))
|
||
return {str(key): int(count) for key, count in rows.all()}
|
||
|
||
|
||
async def collect() -> dict[str, Any]:
|
||
report: dict[str, Any] = {}
|
||
async with SessionFactory() as session:
|
||
report["outbox_columns"] = list(DomainEventOutbox.__table__.columns.keys())
|
||
report["agent_run_columns"] = list(AgentRun.__table__.columns.keys())
|
||
|
||
report["outbox_by_status"] = await _grouped(
|
||
session, DomainEventOutbox, "status"
|
||
)
|
||
report["agent_run_by_status"] = await _grouped(session, AgentRun, "status")
|
||
|
||
oldest = await session.scalar(
|
||
select(func.min(DomainEventOutbox.created_at))
|
||
)
|
||
newest = await session.scalar(
|
||
select(func.max(DomainEventOutbox.created_at))
|
||
)
|
||
report["outbox_created_at_range"] = [str(oldest), str(newest)]
|
||
report["outbox_total"] = await session.scalar(
|
||
select(func.count()).select_from(DomainEventOutbox)
|
||
)
|
||
report["agent_run_total"] = await session.scalar(
|
||
select(func.count()).select_from(AgentRun)
|
||
)
|
||
|
||
# 哪一类事件在堆、哪一类已经进了死信 —— 死信意味着那些事件的副作用永远不会发生。
|
||
report["outbox_by_event_type_status"] = [
|
||
{"event_type": str(row[0]), "status": str(row[1]), "count": int(row[2])}
|
||
for row in (
|
||
await session.execute(
|
||
select(
|
||
DomainEventOutbox.event_type,
|
||
DomainEventOutbox.status,
|
||
func.count(),
|
||
)
|
||
.group_by(DomainEventOutbox.event_type, DomainEventOutbox.status)
|
||
.order_by(func.count().desc())
|
||
)
|
||
).all()
|
||
]
|
||
report["dead_last_errors"] = [
|
||
{"last_error": str(row[0])[:400], "count": int(row[1])}
|
||
for row in (
|
||
await session.execute(
|
||
select(DomainEventOutbox.last_error, func.count())
|
||
.where(DomainEventOutbox.status == "dead")
|
||
.group_by(DomainEventOutbox.last_error)
|
||
.order_by(func.count().desc())
|
||
)
|
||
).all()
|
||
]
|
||
|
||
try:
|
||
from app.model.memory import MemoryEpisode
|
||
|
||
report["episode_by_status"] = await _grouped(session, MemoryEpisode, "status")
|
||
report["episode_total"] = await session.scalar(
|
||
select(func.count()).select_from(MemoryEpisode)
|
||
)
|
||
except Exception as exc: # 模型名/字段与预期不符时只记录，不影响其余结论
|
||
report["episode_probe_error"] = f"{type(exc).__name__}: {exc}"
|
||
return report
|
||
|
||
|
||
async def main() -> None:
|
||
report = await collect()
|
||
OUTPUT.parent.mkdir(parents=True, exist_ok=True)
|
||
OUTPUT.write_text(
|
||
json.dumps(report, ensure_ascii=False, indent=2, default=str),
|
||
encoding="utf-8",
|
||
)
|
||
print(f"wrote {OUTPUT}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|