arXiv Research Radar with AI Relevance Scoring and a Slack Digest

id: arxiv-research-radar-ai-digest
namespace: company.team
description: |
  Every weekday, pull the newest arXiv papers for a search query, skip the ones already
  seen, let Gemini score each new paper against a plain-English description of what your
  team cares about, and post only the relevant ones to Slack with a one-line TL;DR.

inputs:
  - id: radar_name
    type: STRING
    displayName: Radar name
    description: Short name for this radar. It keys the "already seen" list in the
      KV store, so several radars can run side by side.
    defaults: llm-agents
    validator: "^[a-zA-Z0-9_-]+$"

  - id: search_query
    type: STRING
    displayName: arXiv search query
    description: arXiv API query syntax, for example cat:cs.CL AND abs:agent. See
      the arXiv API user manual for fields (ti, abs, au, cat) and boolean
      operators.
    defaults: "(cat:cs.CL OR cat:cs.AI) AND abs:agent"

  - id: research_interest
    type: STRING
    displayName: What your team cares about
    description: Plain-English description used by the LLM to judge relevance. Be
      specific about what is and is not interesting.
    defaults: |
      We build LLM agents that call tools and run multi-step workflows in production.
      Relevant: tool use, planning, agent evaluation and benchmarks, reliability, memory, cost reduction.
      Not relevant: pure theory without experiments, robotics, speech, non-English-only datasets.

  - id: max_results
    type: INT
    displayName: Papers to fetch
    description: Newest papers to fetch per run (1-50). arXiv publishes once per
      weekday, so 25 to 50 covers a busy category.
    defaults: 25
    min: 1
    max: 50

  - id: min_relevance
    type: INT
    displayName: Minimum relevance (1-5)
    description: Only papers scored at or above this value are posted.
    defaults: 4
    min: 1
    max: 5

  - id: model_name
    type: STRING
    displayName: Gemini model
    defaults: gemini-2.5-flash

tasks:
  - id: fetch_papers
    type: io.kestra.plugin.core.http.Download
    description: Query the public arXiv API (Atom XML, no key), newest submissions first.
    uri: "https://export.arxiv.org/api/query?search_query={{ inputs.search_query |
      urlencode }}&sortBy=submittedDate&sortOrder=descending&max_results={{
      inputs.max_results }}"
    headers:
      User-Agent: Kestra-arXiv-Research-Radar

  - id: load_seen
    type: io.kestra.plugin.core.kv.Get
    description: Load the arXiv ids already processed by this radar. Missing on the
      first run.
    key: "arxiv_radar_{{ inputs.radar_name }}"
    errorOnMissing: false

  - id: parse_new_papers
    type: io.kestra.plugin.scripts.python.Script
    description: Parse the Atom feed with the standard library and keep only papers
      this radar has not seen before.
    containerImage: python:3.12-slim
    inputFiles:
      feed.xml: "{{ outputs.fetch_papers.uri }}"
      seen.json: "{{ (outputs.load_seen.value ?? []) | toJson }}"
    script: |
      import json
      import re
      import xml.etree.ElementTree as ET

      NS = {"a": "http://www.w3.org/2005/Atom"}

      def clean(text):
          return re.sub(r"\s+", " ", text or "").strip()

      with open("seen.json") as f:
          seen = json.load(f) or []
      if isinstance(seen, str):
          seen = json.loads(seen)
      seen_set = set(seen)

      root = ET.parse("feed.xml").getroot()
      papers = []
      for entry in root.findall("a:entry", NS):
          abs_url = clean(entry.findtext("a:id", namespaces=NS))
          # http://arxiv.org/abs/2610.10444v1 -> 2610.10444 (versions count as the same paper)
          arxiv_id = re.sub(r"v\d+$", "", abs_url.rsplit("/abs/", 1)[-1])
          if not arxiv_id or arxiv_id in seen_set:
              continue
          authors = [clean(a.findtext("a:name", namespaces=NS)) for a in entry.findall("a:author", NS)]
          papers.append({
              "id": arxiv_id,
              "title": clean(entry.findtext("a:title", namespaces=NS)),
              "authors": ", ".join(authors[:3]) + (" et al." if len(authors) > 3 else ""),
              "published": clean(entry.findtext("a:published", namespaces=NS))[:10],
              "url": f"https://arxiv.org/abs/{arxiv_id}",
              "abstract": clean(entry.findtext("a:summary", namespaces=NS))[:1200],
          })

      # Newest ids first, capped so the KV value stays small.
      updated_seen = ([p["id"] for p in papers] + seen)[:1000]
      print(f"Feed parsed: {len(papers)} new papers, {len(seen_set)} previously seen.")
      print("::" + json.dumps({"outputs": {
          "new_count": len(papers),
          "papers": papers,
          "updated_seen": updated_seen,
      }}) + "::")

  - id: if_new_papers
    type: io.kestra.plugin.core.flow.If
    description: Only call the LLM and Slack when there is something new.
    condition: "{{ outputs.parse_new_papers.vars.new_count > 0 }}"
    then:
      - id: score_papers
        type: io.kestra.plugin.ai.completion.ChatCompletion
        description: Score every new paper from 1 to 5 against the research interest and
          write a one-line TL;DR. Structured JSON output keeps the result
          machine-readable.
        provider:
          type: io.kestra.plugin.ai.provider.GoogleGemini
          apiKey: "{{ secret('GEMINI_API_KEY') }}"
          modelName: "{{ inputs.model_name }}"
        configuration:
          temperature: 0.1
          responseFormat:
            type: JSON
            jsonSchema:
              type: object
              properties:
                papers:
                  type: array
                  items:
                    type: object
                    properties:
                      id:
                        type: string
                      title:
                        type: string
                      relevance:
                        type: integer
                      reason:
                        type: string
                      tldr:
                        type: string
                    required:
                      - id
                      - title
                      - relevance
                      - reason
                      - tldr
              required:
                - papers
        messages:
          - type: SYSTEM
            content: |
              You are a research assistant who screens new arXiv papers for a busy engineering team.
              For every paper you receive, return its exact id and title, a relevance score from 1 (ignore) to
              5 (read today) based only on the team's stated interest, a reason of at most 20 words,
              and a TL;DR of at most 25 words that states the paper's main result in plain language.
              Judge from the title and abstract only. Do not invent results that are not in the abstract.
          - type: USER
            content: |
              Team interest:
              {{ inputs.research_interest }}

              Papers (JSON):
              {{ outputs.parse_new_papers.vars.papers | toJson }}

      - id: build_digest
        type: io.kestra.plugin.scripts.python.Script
        description: Join the LLM scores back to the real paper metadata (titles and
          links never come from the LLM), filter by min_relevance and format the
          Slack digest.
        containerImage: python:3.12-slim
        inputFiles:
          papers.json: "{{ outputs.parse_new_papers.vars.papers | toJson }}"
          scores.json: "{{ outputs.score_papers.jsonOutput | toJson }}"
        script: |
          import difflib
          import json

          MIN_RELEVANCE = int("{{ inputs.min_relevance }}")

          def same_title(a, b):
              # Guards against a score or TL;DR being attached to the wrong paper.
              return difflib.SequenceMatcher(None, a.lower(), b.lower()).ratio() >= 0.8

          with open("papers.json") as f:
              papers = {p["id"]: p for p in json.load(f)}
          with open("scores.json") as f:
              scores = json.load(f)
          if isinstance(scores, str):
              scores = json.loads(scores)

          picked, seen_ids, dropped = [], set(), 0
          for s in scores.get("papers", []):
              paper = papers.get(str(s.get("id", "")).strip())
              if paper is None or paper["id"] in seen_ids or not same_title(paper["title"], str(s.get("title", ""))):
                  dropped += 1  # unknown id, duplicate, or text attached to the wrong paper
                  continue
              seen_ids.add(paper["id"])
              try:
                  relevance = int(s.get("relevance", 0))
              except (TypeError, ValueError):
                  continue
              if relevance >= MIN_RELEVANCE:
                  picked.append({**paper, "relevance": relevance, "reason": s.get("reason", ""), "tldr": s.get("tldr", "")})

          picked.sort(key=lambda p: (-p["relevance"], p["id"]))
          lines = []
          for p in picked[:10]:
              lines.append(
                  f"*<{p['url']}|{p['title']}>* ({p['relevance']}/5)\n"
                  f"{p['authors']}, {p['published']}\n"
                  f"TL;DR: {p['tldr']}\n"
                  f"_Why it matters: {p['reason']}_"
              )
          scored = len(scores.get("papers", []))
          print(f"{len(picked)} of {len(papers)} new papers scored {MIN_RELEVANCE}+ ({scored} scores returned, {dropped} dropped as unknown, duplicate or mismatched).")
          print("::" + json.dumps({"outputs": {
              "relevant_count": len(picked),
              "relevant": picked,
              "digest_markdown": "\n\n".join(lines),
          }}) + "::")

      - id: if_relevant
        type: io.kestra.plugin.core.flow.If
        description: Stay quiet in Slack when nothing new is relevant enough.
        condition: "{{ outputs.build_digest.vars.relevant_count > 0 }}"
        then:
          - id: post_digest
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Post the relevant papers, highest score first (top 10).
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            messageText: |
              :newspaper: *arXiv radar `{{ inputs.radar_name }}`*: {{ outputs.build_digest.vars.relevant_count }} of {{ outputs.parse_new_papers.vars.new_count }} new papers are worth a look

              {{ outputs.build_digest.vars.digest_markdown }}
        else:
          - id: log_nothing_relevant
            type: io.kestra.plugin.core.log.Log
            message: "{{ outputs.parse_new_papers.vars.new_count }} new papers, none scored
              {{ inputs.min_relevance }} or higher. Nothing posted."
    else:
      - id: log_no_new_papers
        type: io.kestra.plugin.core.log.Log
        message: "No new papers for '{{ inputs.search_query }}' since the last run."

  - id: remember_seen
    type: io.kestra.plugin.core.kv.Set
    description: Mark this run's papers as seen. Runs last, so if scoring or Slack
      fails the same papers are retried next time.
    key: "arxiv_radar_{{ inputs.radar_name }}"
    kvType: JSON
    overwrite: true
    value: "{{ outputs.parse_new_papers.vars.updated_seen | toJson }}"

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Report failures so a broken radar is not mistaken for a quiet day on arXiv.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: ":warning: arXiv research radar failed in execution {{ execution.id
      }} ({{ flow.namespace }}.{{ flow.id }}). Check the search query,
      GEMINI_API_KEY and SLACK_WEBHOOK_URL."

triggers:
  - id: weekday_morning
    type: io.kestra.plugin.core.trigger.Schedule
    description: arXiv announces new papers Sunday to Thursday evenings (US
      Eastern), so a weekday morning run catches each batch. Disabled until the
      secrets are set.
    cron: "0 8 * * 1-5"
    disabled: true

  - id: on_demand
    type: io.kestra.plugin.core.trigger.Webhook
    description: Run the radar on demand, for example from a Slack slash command.
    key: arxiv-radar-change-me

outputs:
  - id: new_papers
    type: INT
    value: "{{ outputs.parse_new_papers.vars.new_count }}"
  - id: relevant_papers
    type: JSON
    value: "{{ (outputs.build_digest.vars.relevant ?? []) | toJson }}"