From 3ece2327c895ea888db42ea17c1615ff6e4764f4 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 8 Jul 2026 22:43:57 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=B9=E4=B8=BA=E8=87=AA=E5=8A=A8=E9=87=87?= =?UTF-8?q?=E9=9B=86=E5=B9=B6=E7=A7=BB=E9=99=A4=E6=89=8B=E5=8A=A8=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E5=85=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 4 +- app/agent/scheduler.py | 5 +- app/agent/streaming_runner.py | 136 ---------------------------------- app/api/routes.py | 28 +------ app/ui/home.py | 16 ++-- tests/test_api_health.py | 14 ++++ 6 files changed, 25 insertions(+), 178 deletions(-) delete mode 100644 app/agent/streaming_runner.py diff --git a/README.md b/README.md index 3c945d4..c2cb63e 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ SignalScout 是一个自动运行的 AI 情报 Agent。它先用动态新闻搜 ## 主流程 ```text -Scheduler/API +Scheduler -> AgentRunner -> News Search API 采集近期 AI 新闻候选 -> GitHub Search API 采集近期热门 AI 项目候选 @@ -65,8 +65,6 @@ http://127.0.0.1:8010/docs ## API - `GET /health`：服务健康检查 -- `POST /agent/runs/daily`：运维触发一次 Agent -- `POST /agent/runs/daily/stream`：运维以 SSE 方式触发一次 Agent - `GET /agent/runs`：查看最近运行记录 - `GET /signals`：查看结构化情报 - `GET /reports/latest`：查看最新日报 diff --git a/app/agent/scheduler.py b/app/agent/scheduler.py index 7f5f50b..b952bd4 100644 --- a/app/agent/scheduler.py +++ b/app/agent/scheduler.py @@ -1,5 +1,3 @@ -from datetime import datetime - from apscheduler.schedulers.background import BackgroundScheduler from sqlalchemy.orm import Session @@ -18,7 +16,7 @@ def run_scheduled_job(settings: Settings) -> None: def build_scheduler(settings: Settings) -> BackgroundScheduler: - # APScheduler负责自动循环运行，启动后先跑一次，再按固定间隔持续采集。 + # APScheduler负责按固定间隔自动运行，避免服务重启时连续触发外部采集限流。 scheduler = BackgroundScheduler(timezone=settings.agent_timezone) scheduler.add_job( run_scheduled_job, @@ -26,7 +24,6 @@ def build_scheduler(settings: Settings) -> BackgroundScheduler: minutes=settings.agent_interval_minutes, args=[settings], id="continuous-intelligence", - next_run_time=datetime.now(scheduler.timezone), replace_existing=True, max_instances=1, coalesce=True, diff --git a/app/agent/streaming_runner.py b/app/agent/streaming_runner.py deleted file mode 100644 index 73ff913..0000000 --- a/app/agent/streaming_runner.py +++ /dev/null @@ -1,136 +0,0 @@ -from __future__ import annotations - -import json -import logging -from datetime import date -from typing import Iterator, Optional - -from app.agent.report_writer import ReportWriter -from app.core.config import Settings -from app.db.models import AgentRun, Signal -from app.db.repository import AgentRepository -from app.db.session import SessionLocal -from app.integrations.github_client import GitHubHotProjectClient -from app.integrations.news_search_client import NewsSearchClient -from app.llm.grok_client import GrokIntelligenceClient - -logger = logging.getLogger(__name__) - - -def stream_daily_agent_events(settings: Settings, run_date: Optional[date] = None) -> Iterator[str]: - """以SSE方式运行每日Agent，并在模型完成后逐条发送结构化情报。""" - db = SessionLocal() - repo = AgentRepository(db) - current_date = run_date or date.today() - run: Optional[AgentRun] = None - - try: - request_payload = { - "run_date": current_date.isoformat(), - "mode": "grok_agent_stream", - } - run = repo.create_run(request_payload=request_payload) - db.commit() - logger.info("agent_stream_started run_id=%s run_date=%s mode=grok_agent", run.id, current_date) - yield _sse("started", {"run_id": run.id, "status": run.status}) - - news = NewsSearchClient(settings) - github = GitHubHotProjectClient(settings) - grok = GrokIntelligenceClient(settings) - topics = repo.list_enabled_topics() - yield _sse("stage", {"run_id": run.id, "stage": "news"}) - news_candidates = news.collect_ai_news(run_date=current_date) - yield _sse("stage", {"run_id": run.id, "stage": "github"}) - github_projects = github.collect_hot_projects(run_date=current_date) - yield _sse("stage", {"run_id": run.id, "stage": "grok"}) - result = grok.collect_daily_intelligence( - topics=topics, - news_candidates=news_candidates, - github_projects=github_projects, - run_date=current_date, - ) - - saved_count = 0 - final_signals = [] - for item in result.signals: - # 模型一次性完成筛选和摘要，SSE只负责把最终情报逐条落库并推给前端。 - signal = repo.save_or_update_signal(run=run, item=item) - if signal is None: - continue - db.commit() - db.refresh(signal) - final_signals.append(signal) - saved_count += 1 - yield _sse("candidate", {"signal": _signal_payload(signal), "saved_count": saved_count}) - - writer = ReportWriter(settings.report_dir) - report_path = writer.write_daily_report(current_date, result.report_markdown) - repo.save_report( - run=run, - title=result.title, - content_md=result.report_markdown, - file_path=str(report_path), - ) - repo.mark_run_complete( - run=run, - raw_response={ - "mode": "grok_agent_stream", - "topic_count": len(topics), - "news_candidate_count": len(news_candidates), - "github_project_count": len(github_projects), - "model": settings.llm_model, - "model_response": result.raw_response, - }, - ) - db.commit() - logger.info( - "agent_stream_completed run_id=%s candidates=%s final_signals=%s report_path=%s", - run.id, - saved_count, - len(final_signals), - report_path, - ) - yield _sse( - "completed", - { - "run_id": run.id, - "status": "completed", - "candidate_count": saved_count, - "final_signal_count": len(final_signals), - "report_path": str(report_path), - }, - ) - except Exception as exc: - if run is not None: - repo.mark_run_failed(run=run, error=str(exc)) - db.commit() - logger.exception("agent_stream_failed run_id=%s error=%s", run.id, exc) - yield _sse("failed", {"run_id": run.id, "status": "failed", "error": str(exc)}) - else: - logger.exception("agent_stream_failed_before_run error=%s", exc) - yield _sse("failed", {"run_id": None, "status": "failed", "error": str(exc)}) - finally: - db.close() - - -def _signal_payload(signal: Signal) -> dict: - # SSE载荷只使用JSON安全字段，方便前端直接追加到列表。 - return { - "id": signal.id, - "run_id": signal.run_id, - "topic": signal.topic, - "title": signal.title, - "summary": signal.summary, - "source_url": signal.source_url, - "source_name": signal.source_name, - "published_at": signal.published_at.isoformat() if signal.published_at else None, - "importance": signal.importance, - "entities": signal.entities, - "created_at": signal.created_at.isoformat() if signal.created_at else None, - } - - -def _sse(event: str, data: dict) -> str: - # 标准SSE帧让前端可以通过流式请求消费事件。 - payload = json.dumps(data, ensure_ascii=False) - return f"event: {event}\ndata: {payload}\n\n" diff --git a/app/api/routes.py b/app/api/routes.py index 0e187d1..cb62e56 100644 --- a/app/api/routes.py +++ b/app/api/routes.py @@ -1,15 +1,12 @@ from __future__ import annotations -from datetime import date from typing import Optional from fastapi import APIRouter, Depends -from fastapi.responses import HTMLResponse, StreamingResponse +from fastapi.responses import HTMLResponse from sqlalchemy import select from sqlalchemy.orm import Session -from app.agent.runner import AgentRunner -from app.agent.streaming_runner import stream_daily_agent_events from app.core.config import Settings, get_settings from app.db.models import AgentRun, Report, Signal, Topic from app.db.session import get_db @@ -17,7 +14,6 @@ from app.schemas.agent import ( DashboardRead, ReportRead, RunRead, - RunResponse, SignalRead, TopicCreate, TopicRead, @@ -39,28 +35,6 @@ def health(settings: Settings = Depends(get_settings)) -> dict: return {"status": "ok", "app": settings.app_name, "env": settings.env} -@router.post("/agent/runs/daily", response_model=RunResponse) -def run_daily_agent( - run_date: Optional[date] = None, - db: Session = Depends(get_db), - settings: Settings = Depends(get_settings), -) -> RunResponse: - # 手动触发和定时任务共用同一个Runner，保证行为一致。 - return AgentRunner(db=db, settings=settings).run_daily(run_date=run_date) - - -@router.post("/agent/runs/daily/stream") -def stream_daily_agent( - run_date: Optional[date] = None, - settings: Settings = Depends(get_settings), -) -> StreamingResponse: - # 流式接口立即启动Agent，并把最终结构化情报逐条推送给前端。 - return StreamingResponse( - stream_daily_agent_events(settings=settings, run_date=run_date), - media_type="text/event-stream", - ) - - @router.get("/agent/runs", response_model=list[RunRead]) def list_runs(db: Session = Depends(get_db)) -> list[AgentRun]: # 运行列表保持紧凑，方便仪表盘和命令行查看最近状态。 diff --git a/app/ui/home.py b/app/ui/home.py index ec02dcf..73a5fc6 100644 --- a/app/ui/home.py +++ b/app/ui/home.py @@ -344,11 +344,11 @@ HOME_PAGE_HTML = """
自动整理 - 启动后自动运行，之后约每 30 分钟整理一次。 + 约每 30 分钟自动整理一次。

聚合 AI 新闻、模型发布、Agent 工具和 GitHub 热门项目。

@@ -387,7 +387,7 @@ HOME_PAGE_HTML = """
-

最近运行

+

自动记录

0 次
@@ -410,7 +410,7 @@ HOME_PAGE_HTML = """
-

最近运行

+

自动记录

0 次
@@ -522,10 +522,10 @@ HOME_PAGE_HTML = """ runCount.textContent = `${runs.length} 次`; runCountFull.textContent = `${runs.length} 次`; if (!runs.length) { - runPreview.innerHTML = '
Agent 会在启动后自动开始整理。
'; - runList.innerHTML = '
Agent 会在启动后自动开始整理。
'; - lastRunText.textContent = "等待运行"; - lastRunDetail.textContent = "启动后自动运行，之后约每 30 分钟整理一次。"; + runPreview.innerHTML = '
Agent 会按固定节奏自动整理。
'; + runList.innerHTML = '
Agent 会按固定节奏自动整理。
'; + lastRunText.textContent = "等待记录"; + lastRunDetail.textContent = "约每 30 分钟自动整理一次。"; return; } runPreview.innerHTML = runRows(runs.slice(0, 4)); diff --git a/tests/test_api_health.py b/tests/test_api_health.py index 2a72407..67bf999 100644 --- a/tests/test_api_health.py +++ b/tests/test_api_health.py @@ -28,3 +28,17 @@ def test_home_page_returns_dashboard_shell(monkeypatch) -> None: assert response.status_code == 200 assert "SignalScout" in response.text assert "今日 AI 情报" in response.text + assert "采集今日情报" not in response.text + + +def test_manual_agent_run_endpoint_is_not_exposed(monkeypatch) -> None: + # Agent只通过后台调度器自动运行，HTTP API不提供手动启动入口。 + monkeypatch.setenv("SIGNALSCOUT_ENV", "test") + get_settings.cache_clear() + with TestClient(create_app()) as client: + response = client.post("/agent/runs/daily") + stream_response = client.post("/agent/runs/daily/stream") + get_settings.cache_clear() + + assert response.status_code == 404 + assert stream_response.status_code == 404