改为自动采集并移除手动运行入口
This commit is contained in:
Codex
@@ -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,
|
||||
|
||||
@@ -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"
|
||||
+1
-27
@@ -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]:
|
||||
# 运行列表保持紧凑，方便仪表盘和命令行查看最近状态。
|
||||
|
||||
+8
-8
@@ -344,11 +344,11 @@ HOME_PAGE_HTML = """
|
||||
<nav class="nav" aria-label="主导航">
|
||||
<button class="active" type="button" data-view-target="signals">情报</button>
|
||||
<button type="button" data-view-target="report">日报</button>
|
||||
<button type="button" data-view-target="runs">运行</button>
|
||||
<button type="button" data-view-target="runs">记录</button>
|
||||
</nav>
|
||||
<div class="auto-card">
|
||||
<strong>自动整理</strong>
|
||||
<span>启动后自动运行，之后约每 30 分钟整理一次。</span>
|
||||
<span>约每 30 分钟自动整理一次。</span>
|
||||
</div>
|
||||
<p class="nav-note">聚合 AI 新闻、模型发布、Agent 工具和 GitHub 热门项目。</p>
|
||||
</aside>
|
||||
@@ -387,7 +387,7 @@ HOME_PAGE_HTML = """
|
||||
|
||||
<div class="panel">
|
||||
<div class="panel-head">
|
||||
<h2>最近运行</h2>
|
||||
<h2>自动记录</h2>
|
||||
<span class="count" id="runCount">0 次</span>
|
||||
</div>
|
||||
<div class="runs" id="runPreview">
|
||||
@@ -410,7 +410,7 @@ HOME_PAGE_HTML = """
|
||||
<section class="view" data-view="runs">
|
||||
<div class="panel">
|
||||
<div class="panel-head">
|
||||
<h2>最近运行</h2>
|
||||
<h2>自动记录</h2>
|
||||
<span class="count" id="runCountFull">0 次</span>
|
||||
</div>
|
||||
<div class="runs" id="runList">
|
||||
@@ -522,10 +522,10 @@ HOME_PAGE_HTML = """
|
||||
runCount.textContent = `${runs.length} 次`;
|
||||
runCountFull.textContent = `${runs.length} 次`;
|
||||
if (!runs.length) {
|
||||
runPreview.innerHTML = '<div class="empty">Agent 会在启动后自动开始整理。</div>';
|
||||
runList.innerHTML = '<div class="empty">Agent 会在启动后自动开始整理。</div>';
|
||||
lastRunText.textContent = "等待运行";
|
||||
lastRunDetail.textContent = "启动后自动运行，之后约每 30 分钟整理一次。";
|
||||
runPreview.innerHTML = '<div class="empty">Agent 会按固定节奏自动整理。</div>';
|
||||
runList.innerHTML = '<div class="empty">Agent 会按固定节奏自动整理。</div>';
|
||||
lastRunText.textContent = "等待记录";
|
||||
lastRunDetail.textContent = "约每 30 分钟自动整理一次。";
|
||||
return;
|
||||
}
|
||||
runPreview.innerHTML = runRows(runs.slice(0, 4));
|
||||
|
||||
Reference in New Issue
Block a user