改用Grok联网搜索并保留每轮信号

2026-07-09 00:14:20 +08:00
parent 4f50de8def
commit e02d5db495
14 changed files with 465 additions and 110 deletions
@@ -13,7 +13,7 @@ from app.db.repository import AgentRepository
 from app.integrations.github_client import GitHubHotProjectClient
 from app.integrations.news_search_client import NewsSearchClient
 from app.llm.grok_client import GrokIntelligenceClient
-from app.schemas.agent import RunResponse
+from app.schemas.agent import RunResponse, SignalItem

 logger = logging.getLogger(__name__)

@@ -49,7 +49,12 @@ class AgentRunner:
                github_projects=github_projects,
                run_date=current_date,
            )
-            signals = self.repo.save_signals(run=run, items=result.signals)
+            signal_items = self._compose_run_signals(
+                news_candidates=news_candidates,
+                github_projects=github_projects,
+                model_signals=result.signals,
+            )
+            signals = self.repo.save_signals(run=run, items=signal_items)
            report_path = self.writer.write_daily_report(current_date, result.report_markdown)
            self.repo.save_report(
                run=run,
@@ -64,6 +69,8 @@ class AgentRunner:
                    "topic_count": len(topics),
                    "news_candidate_count": len(news_candidates),
                    "github_project_count": len(github_projects),
+                    "model_signal_count": len(result.signals),
+                    "saved_signal_count": len(signal_items),
                    "model": self.grok.settings.llm_model,
                    "model_response": result.raw_response,
                },
@@ -81,3 +88,29 @@ class AgentRunner:
            self.db.commit()
            logger.exception("agent_run_failed run_id=%s error=%s", run.id, exc)
            raise
+
+    def _compose_run_signals(
+        self,
+        news_candidates: list[SignalItem],
+        github_projects: list[SignalItem],
+        model_signals: list[SignalItem],
+    ) -> list[SignalItem]:
+        # 入库清单以采集候选为准，模型同链接结果只增强摘要和重要度，避免日报模型漏选导致信号消失。
+        model_by_url = {item.source_url: item for item in model_signals}
+        composed = []
+        for candidate in [*news_candidates, *github_projects]:
+            model_item = model_by_url.get(candidate.source_url)
+            if model_item is None:
+                composed.append(candidate)
+                continue
+            composed.append(
+                candidate.model_copy(
+                    update={
+                        "topic": model_item.topic,
+                        "summary": model_item.summary,
+                        "importance": model_item.importance,
+                        "entities": model_item.entities or candidate.entities,
+                    }
+                )
+            )
+        return composed
@@ -2,11 +2,12 @@ from __future__ import annotations

 from typing import Literal, Optional

-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, BackgroundTasks, Depends
 from fastapi.responses import HTMLResponse
 from sqlalchemy import select
 from sqlalchemy.orm import Session

+from app.agent.scheduler import run_scheduled_job
 from app.core.config import Settings, get_settings
 from app.db.models import AgentRun, Report, Signal, Topic
 from app.db.session import get_db
@@ -14,6 +15,7 @@ from app.schemas.agent import (
    DashboardRead,
    ReportRead,
    RunRead,
+    RunTriggerRead,
    SignalRead,
    TopicCreate,
    TopicRead,
@@ -41,6 +43,22 @@ def list_runs(db: Session = Depends(get_db)) -> list[AgentRun]:
    return list(db.scalars(select(AgentRun).order_by(AgentRun.id.desc()).limit(20)).all())


+@router.post("/agent/runs/daily", response_model=RunTriggerRead)
+def trigger_daily_run(
+    background_tasks: BackgroundTasks,
+    settings: Settings = Depends(get_settings),
+    db: Session = Depends(get_db),
+) -> RunTriggerRead:
+    # 手动搜索复用定时任务入口，避免测试按钮和自动调度走出两套采集逻辑。
+    running = db.scalar(
+        select(AgentRun).where(AgentRun.status == "running").order_by(AgentRun.id.desc()).limit(1)
+    )
+    if running is not None:
+        return RunTriggerRead(status="running", run_id=running.id)
+    background_tasks.add_task(run_scheduled_job, settings)
+    return RunTriggerRead(status="started")
+
+
@router.get("/signals", response_model=list[SignalRead])
 def list_signals(
    limit: int = 100,
@@ -15,7 +15,7 @@ class Settings(BaseSettings):
    llm_base_url: str = "https://api.zayuapi.com/v1"
    llm_api_key: str = ""
    llm_model: str = "grok-4.3"
-    llm_timeout_seconds: int = 90
+    llm_timeout_seconds: int = 180
    llm_max_tokens: int = 8000
    news_recent_days: int = 3
    news_max_items: int = 60
@@ -3,7 +3,7 @@ from __future__ import annotations
 from datetime import datetime
 from typing import Optional

-from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, JSON, String, Text, func
+from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, JSON, String, Text, UniqueConstraint, func
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


@@ -41,6 +41,9 @@ class AgentRun(Base):

 class Signal(Base):
    __tablename__ = "signals"
+    __table_args__ = (
+        UniqueConstraint("run_id", "source_url_hash", name="uq_signals_run_source_url_hash"),
+    )

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    run_id: Mapped[int] = mapped_column(ForeignKey("agent_runs.id"), nullable=False, index=True)
@@ -49,7 +52,7 @@ class Signal(Base):
    title: Mapped[str] = mapped_column(String(300), nullable=False)
    summary: Mapped[str] = mapped_column(Text, nullable=False)
    source_url: Mapped[str] = mapped_column(Text, nullable=False)
-    source_url_hash: Mapped[str] = mapped_column(String(64), nullable=False, unique=True)
+    source_url_hash: Mapped[str] = mapped_column(String(64), nullable=False)
    source_name: Mapped[str] = mapped_column(String(200), nullable=False)
    published_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True)
    importance: Mapped[int] = mapped_column(Integer, nullable=False)
@@ -1,8 +1,6 @@
 from __future__ import annotations

 from hashlib import sha256
-from typing import Optional
-
 from sqlalchemy import select
 from sqlalchemy.orm import Session

@@ -41,21 +39,22 @@ class AgentRepository:
        self.db.flush()

    def save_signals(self, run: AgentRun, items: list[SignalItem]) -> list[Signal]:
-        # 相同来源链接只保存一次，避免日报和仪表盘重复出现同一条情报。
+        # 去重范围限定在同一次运行内，保证每轮情报结果都能完整留档。
        saved: list[Signal] = []
        for item in items:
            signal = self.save_or_update_signal(run=run, item=item)
-            if signal is None:
-                continue
            saved.append(signal)
        return saved

-    def save_or_update_signal(self, run: AgentRun, item: SignalItem) -> Optional[Signal]:
-        # 同一来源链接只保留一条情报，流式接口和普通接口共用这条去重规则。
+    def save_or_update_signal(self, run: AgentRun, item: SignalItem) -> Signal:
+        # 同一次运行里相同链接更新为一条，跨运行保留各自结果用于历史对比。
        source_url_hash = self._source_url_hash(item.source_url)
-        existing = self.db.scalar(select(Signal).where(Signal.source_url_hash == source_url_hash))
-        if existing and existing.run_id != run.id:
-            return None
+        existing = self.db.scalar(
+            select(Signal).where(
+                Signal.run_id == run.id,
+                Signal.source_url_hash == source_url_hash,
+            )
+        )
        if existing:
            existing.source_type = item.source_type
            existing.topic = item.topic
@@ -2,114 +2,214 @@ from __future__ import annotations

 import json
 import logging
-from datetime import date, datetime, time, timedelta
+from datetime import date, datetime, timedelta
 from typing import Any, Optional
 from urllib.error import HTTPError, URLError
-from urllib.parse import urlencode
 from urllib.request import Request, urlopen

 from app.core.config import Settings
-from app.core.timezone import BEIJING_TZ, to_beijing_naive
+from app.core.timezone import to_beijing_naive
 from app.schemas.agent import SignalItem

 logger = logging.getLogger(__name__)


 class NewsSearchClient:
-    """通过公开技术新闻检索接口采集动态AI新闻候选。"""
-
-    endpoint = "https://hn.algolia.com/api/v1/search_by_date"
+    """通过Grok Web Search采集中英文AI新闻候选。"""

    def __init__(self, settings: Settings) -> None:
        self.settings = settings

    def collect_ai_news(self, run_date: date) -> list[SignalItem]:
-        # HN Algolia按时间倒序返回技术新闻候选，模型后续只从这些证据里挑选日报条目。
-        since = run_date - timedelta(days=self.settings.news_recent_days)
-        stories: dict[str, dict[str, Any]] = {}
-        for query in self._queries():
-            payload = self._search(query=query, since=since, until=run_date)
-            for story in payload.get("hits", []):
-                url = self._story_url(story)
-                object_id = story.get("objectID")
-                key = str(object_id or url)
-                if isinstance(url, str) and url.startswith("http"):
-                    stories[key] = story
-
-        candidates = [self._story_to_signal(story) for story in stories.values()]
+        # Grok负责联网搜索新闻候选，入库前仍统一校验来源链接和结构字段。
+        items: list[dict[str, Any]] = []
+        for query_group in self._query_groups():
+            payload = self._search_with_grok(run_date=run_date, query_group=query_group)
+            items.extend(payload.get("items", []))
+        candidates = [self._article_to_signal(item) for item in items]
        candidates = [candidate for candidate in candidates if candidate is not None]
-        candidates.sort(key=lambda item: item.published_at or datetime.min, reverse=True)
-        logger.info("news_candidates_collected days=%s items=%s", self.settings.news_recent_days, len(candidates))
-        return candidates[: self.settings.news_max_items]
+        unique: dict[str, SignalItem] = {}
+        for candidate in candidates:
+            unique[candidate.source_url] = candidate
+        ranked = sorted(
+            unique.values(),
+            key=lambda item: (item.importance, item.published_at or datetime.min),
+            reverse=True,
+        )
+        logger.info(
+            "news_candidates_collected source=grok_web_search days=%s items=%s",
+            self.settings.news_recent_days,
+            len(ranked),
+        )
+        return ranked[: self.settings.news_max_items]

-    def _queries(self) -> list[str]:
-        # 查询词覆盖模型、Agent、开源AI和融资产品动态，避免单一关键词漏掉日报候选。
+    def _query_groups(self) -> list[dict[str, Any]]:
+        # 多组搜索覆盖中文、英文、融资、模型和Agent工具，避免单次搜索只返回少量相似新闻。
        return [
-            "AI OR LLM OR agent",
-            "large language model",
-            "OpenAI Anthropic",
-            "AI startup funding",
+            {
+                "name": "中文AI新闻",
+                "language": "zh",
+                "queries": ["人工智能 大模型 智能体 AI 新闻", "AI 大模型 创业 融资 产品 发布"],
+                "target_count": 15,
+            },
+            {
+                "name": "英文AI新闻",
+                "language": "en",
+                "queries": ["latest AI news LLM agents OpenAI Anthropic", "AI startup funding model release agent tools"],
+                "target_count": 15,
+            },
+            {
+                "name": "模型与Agent发布",
+                "language": "mixed",
+                "queries": ["大模型 发布 Agent 工具 开源", "LLM model release AI agent framework open source"],
+                "target_count": 15,
+            },
+            {
+                "name": "AI商业与融资",
+                "language": "mixed",
+                "queries": ["AI 融资 创业 公司 产品", "AI funding startup acquisition product launch"],
+                "target_count": 15,
+            },
        ]

-    def _search(self, query: str, since: date, until: date) -> dict[str, Any]:
-        # search_by_date接口稳定返回最新故事，numericFilters把候选限制在本次日报窗口内。
-        start_at = int(datetime.combine(since, time.min, tzinfo=BEIJING_TZ).timestamp())
-        end_at = int(datetime.combine(until, time.max, tzinfo=BEIJING_TZ).timestamp())
-        params = {
-            "query": query,
-            "tags": "story",
-            "hitsPerPage": str(self.settings.news_search_max_records),
-            "numericFilters": f"created_at_i>={start_at},created_at_i<={end_at}",
-        }
-        url = f"{self.endpoint}?{urlencode(params)}"
-        request = Request(url, headers={"User-Agent": "SignalScout"})
+    def _search_with_grok(self, run_date: date, query_group: dict[str, Any]) -> dict[str, Any]:
+        # Responses API的web_search工具是当前新闻发现入口，确保候选来自真实网页搜索。
+        if not self.settings.llm_api_key:
+            raise RuntimeError("SIGNALSCOUT_LLM_API_KEY is required")
+        since = run_date - timedelta(days=self.settings.news_recent_days)
+        body = json.dumps(
+            {
+                "model": self.settings.llm_model,
+                "input": [
+                    {
+                        "role": "system",
+                        "content": (
+                            "你是 SignalScout 的新闻搜索器。必须使用 web_search 找到真实网页来源，"
+                            "只输出 JSON object，不输出 Markdown 或额外解释。"
+                        ),
+                    },
+                    {
+                        "role": "user",
+                        "content": json.dumps(
+                            {
+                                "task": "联网搜索最近AI新闻，覆盖中文和英文来源。",
+                                "date_window": {
+                                    "from": since.isoformat(),
+                                    "to": run_date.isoformat(),
+                                },
+                                "search_group": query_group["name"],
+                                "preferred_language": query_group["language"],
+                                "target_count": query_group["target_count"],
+                                "queries": query_group["queries"],
+                                "output_schema": {
+                                    "items": [
+                                        {
+                                            "title": "新闻标题",
+                                            "summary": "中文摘要，说明为什么值得关注",
+                                            "url": "真实可访问来源链接",
+                                            "source": "媒体或网站名称",
+                                            "published_at": "ISO时间；不知道可为null",
+                                            "language": "zh或en",
+                                            "importance": "1到5整数",
+                                            "entities": ["公司、模型、项目、人名等"],
+                                        }
+                                    ]
+                                },
+                                "rules": [
+                                    "优先返回中文和英文来源各一部分，不要只返回英文来源。",
+                                    "只使用最近日期窗口内或明显接近日内的新闻。",
+                                    "每条必须有真实 url，不要使用搜索结果页、YouTube集合页或空链接。",
+                                    "如果中文来源足够，至少保留三分之一中文新闻。",
+                                    "返回尽可能接近 target_count 的 items。",
+                                ],
+                            },
+                            ensure_ascii=False,
+                        ),
+                    },
+                ],
+                "tools": [{"type": "web_search"}],
+                "max_output_tokens": self.settings.llm_max_tokens,
+            },
+            ensure_ascii=False,
+        ).encode("utf-8")
+        request = Request(
+            f"{self.settings.llm_base_url.rstrip('/')}/responses",
+            data=body,
+            headers={
+                "Authorization": f"Bearer {self.settings.llm_api_key}",
+                "Content-Type": "application/json",
+                "Accept": "application/json",
+            },
+            method="POST",
+        )
        try:
-            with urlopen(request, timeout=20) as response:
-                body = response.read().decode("utf-8")
+            with urlopen(request, timeout=self.settings.llm_timeout_seconds) as response:
+                response_body = response.read().decode("utf-8")
        except HTTPError as exc:
            message = exc.read().decode("utf-8", errors="replace")
-            raise RuntimeError(f"News search failed: HTTP {exc.code} {message}") from exc
+            raise RuntimeError(f"Grok news search failed: HTTP {exc.code} {message}") from exc
        except URLError as exc:
-            raise RuntimeError(f"News search failed: {exc.reason}") from exc
-        return json.loads(body)
+            raise RuntimeError(f"Grok news search failed: {exc.reason}") from exc
+        return self._parse_response_json(json.loads(response_body))

-    def _story_to_signal(self, story: dict[str, Any]) -> Optional[SignalItem]:
-        # HN故事先保留标题、链接和互动指标，重要性与摘要由模型最终确定。
-        title = story.get("title")
-        url = self._story_url(story)
-        if not isinstance(title, str) or not isinstance(url, str):
+    def _parse_response_json(self, response_payload: dict[str, Any]) -> dict[str, Any]:
+        # Responses返回多段output，最终message里的output_text才是模型给业务层的JSON。
+        text_parts: list[str] = []
+        for output in response_payload.get("output", []):
+            if not isinstance(output, dict):
+                continue
+            for content in output.get("content", []):
+                if isinstance(content, dict) and content.get("type") == "output_text":
+                    text = content.get("text")
+                    if isinstance(text, str):
+                        text_parts.append(text)
+        content_text = "\n".join(text_parts).strip()
+        if not content_text:
+            raise RuntimeError("Grok news search returned no output_text")
+        try:
+            payload = json.loads(content_text)
+        except json.JSONDecodeError as exc:
+            raise RuntimeError(f"Grok news search returned invalid JSON: {content_text[:1200]}") from exc
+        if not isinstance(payload, dict):
+            raise RuntimeError("Grok news search JSON root is not an object")
+        return payload
+
+    def _article_to_signal(self, item: dict[str, Any]) -> Optional[SignalItem]:
+        # 搜索结果只接受带真实网页链接的条目，避免把模型说明文本写入情报库。
+        title = item.get("title")
+        url = item.get("url")
+        if not isinstance(title, str) or not isinstance(url, str) or not url.startswith("http"):
            return None
-        author = str(story.get("author") or "unknown")
-        points = int(story.get("points") or 0)
-        comments = int(story.get("num_comments") or 0)
-        published_at = self._parse_hn_date(story.get("created_at"))
+        source = item.get("source")
+        language = item.get("language")
+        entities = item.get("entities")
        return SignalItem(
            source_type="news",
-            topic="AI 新闻候选",
+            topic="中文 AI 新闻" if language == "zh" else "AI 新闻",
            title=title[:300],
-            summary=f"Hacker News 最新技术讨论，作者 {author}，约 {points} points、{comments} 条评论。",
+            summary=str(item.get("summary") or "Grok Web Search 搜索到的AI新闻候选。"),
            source_url=url,
-            source_name="Hacker News",
-            published_at=published_at,
-            importance=3,
-            entities=["Hacker News", author],
+            source_name=str(source or "Grok Web Search"),
+            published_at=self._parse_datetime(item.get("published_at")),
+            importance=self._parse_importance(item.get("importance")),
+            entities=[str(entity) for entity in entities] if isinstance(entities, list) else [],
        )

-    def _story_url(self, story: dict[str, Any]) -> Optional[str]:
-        # 外链优先；没有外链的Ask/Show HN故事使用HN讨论页作为证据链接。
-        url = story.get("url") or story.get("story_url")
-        if isinstance(url, str) and url.startswith("http"):
-            return url
-        object_id = story.get("objectID")
-        if object_id:
-            return f"https://news.ycombinator.com/item?id={object_id}"
-        return None
-
-    def _parse_hn_date(self, value: Any) -> Optional[datetime]:
-        # HN Algolia返回UTC ISO时间，入库前先换算为北京时间。
-        if not isinstance(value, str):
+    def _parse_datetime(self, value: Any) -> Optional[datetime]:
+        # 搜索结果可能只有日期或ISO时间，能解析则统一换算为北京时间。
+        if not isinstance(value, str) or not value:
            return None
        try:
+            if len(value) == 10:
+                value = f"{value}T00:00:00"
            return to_beijing_naive(datetime.fromisoformat(value.replace("Z", "+00:00")))
        except ValueError:
            return None
+
+    def _parse_importance(self, value: Any) -> int:
+        # 重要度来自搜索模型，但入库前仍限制在业务契约允许范围内。
+        try:
+            importance = int(value)
+        except (TypeError, ValueError):
+            return 3
+        return max(1, min(importance, 5))
@@ -39,6 +39,11 @@ class RunResponse(BaseModel):
    report_path: Optional[str] = None


+class RunTriggerRead(BaseModel):
+    status: str
+    run_id: Optional[int] = None
+
+
 class RunRead(BaseModel):
    id: int
    status: str
@@ -162,6 +162,30 @@ HOME_PAGE_HTML = """
      margin-top: 5px;
    }

+    .actions {
+      display: flex;
+      justify-content: flex-end;
+      margin-top: 12px;
+    }
+
+    .action-button {
+      border: 1px solid oklch(25% 0.02 210);
+      border-radius: 8px;
+      padding: 9px 13px;
+      background: var(--ink);
+      color: white;
+      font: inherit;
+      font-size: 13px;
+      font-weight: 700;
+      cursor: pointer;
+      min-width: 96px;
+    }
+
+    .action-button:disabled {
+      cursor: wait;
+      opacity: 0.66;
+    }
+
    .layout {
      display: grid;
      grid-template-columns: minmax(0, 1.2fr) minmax(340px, 0.8fr);
@@ -410,6 +434,9 @@ HOME_PAGE_HTML = """
        <div class="pulse">
          <strong id="todayCountText">读取中</strong>
          <span id="todayCountDetail">正在整理新闻与项目。</span>
+          <div class="actions">
+            <button class="action-button" id="manualRunButton" type="button">手动搜索</button>
+          </div>
        </div>
      </section>

@@ -484,6 +511,7 @@ HOME_PAGE_HTML = """
    const runCountFull = document.querySelector("#runCountFull");
    const todayCountText = document.querySelector("#todayCountText");
    const todayCountDetail = document.querySelector("#todayCountDetail");
+    const manualRunButton = document.querySelector("#manualRunButton");

    const stateText = {
      completed: "完成",
@@ -667,9 +695,34 @@ HOME_PAGE_HTML = """
      renderRuns(data.runs || []);
    }

+    async function triggerManualRun() {
+      manualRunButton.disabled = true;
+      manualRunButton.textContent = "搜索中";
+      try {
+        const response = await fetch("/agent/runs/daily", {
+          method: "POST",
+          headers: { Accept: "application/json" }
+        });
+        if (!response.ok) throw new Error("手动搜索失败");
+        await loadDashboard();
+      } finally {
+        window.setTimeout(() => {
+          manualRunButton.disabled = false;
+          manualRunButton.textContent = "手动搜索";
+          loadDashboard().catch(() => {});
+        }, 3000);
+      }
+    }
+
    navButtons.forEach((button) => {
      button.addEventListener("click", () => activateView(button.dataset.viewTarget));
    });
+    manualRunButton.addEventListener("click", () => {
+      triggerManualRun().catch(() => {
+        manualRunButton.disabled = false;
+        manualRunButton.textContent = "手动搜索";
+      });
+    });

    const initialView = window.location.hash.replace("#", "");
    if (["signals", "report", "runs"].includes(initialView)) {