Implementation Guide: Monitor Global S&T Publications & Patent Filings for Near-Peer Advances — Generate Competitive Technology Assessments

# Autonomous S&T monitoring agent that monitors publications, preprints, and
# patents

# st_monitoring_agent.py
# Autonomous S&T monitoring agent — monitors publications, preprints, and patents

import requests
import os, json, datetime, time
from openai import AzureOpenAI
from azure.search.documents import SearchClient
from azure.core.credentials import AzureKeyCredential

aoai_client = AzureOpenAI(
    azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
    api_key=os.environ["AZURE_OPENAI_KEY"],
    api_version="2024-08-01-preview"
)

SEARCH_CLIENT = SearchClient(
    endpoint=os.environ["AZURE_SEARCH_ENDPOINT"],
    index_name="st-publications",
    credential=AzureKeyCredential(os.environ["AZURE_SEARCH_KEY"])
)

# Technology watch list — customized per client's research focus
TECHNOLOGY_WATCH_LIST = [
    {
        "id": "TW-001",
        "area": "Hypersonic Guidance and Control",
        "keywords": ["hypersonic glide vehicle", "HGV", "boost-glide", "maneuvering reentry",
                     "terminal guidance hypersonic", "HTV", "aerothermal guidance"],
        "priority_institutions": ["NUDT", "CASC", "CASIC", "China Aerospace", "Harbin Institute",
                                  "Beijing Institute of Technology", "MIPT", "TsAGI"],
        "current_us_trl": 7,
        "assessment_frequency": "weekly"
    },
    {
        "id": "TW-002",
        "area": "High-Power Microwave Directed Energy",
        "keywords": ["high power microwave", "HPM", "directed energy weapon", "DEW",
                     "active denial", "gyrotron", "vircator", "relativistic magnetron"],
        "priority_institutions": ["NUDT", "NINT", "Novosibirsk Institute", "RFNC-VNIIEF"],
        "current_us_trl": 6,
        "assessment_frequency": "weekly"
    },
    {
        "id": "TW-003",
        "area": "Quantum Sensing and Navigation",
        "keywords": ["quantum sensing", "quantum navigation", "atom interferometry",
                     "quantum inertial sensor", "quantum gravimeter", "quantum magnetometer",
                     "GPS-denied navigation quantum"],
        "priority_institutions": ["CAS", "Peking University", "Tsinghua University",
                                  "SITP", "Russian Quantum Center"],
        "current_us_trl": 5,
        "assessment_frequency": "biweekly"
    },
    {
        "id": "TW-004",
        "area": "AI/ML for Autonomous Weapons Systems",
        "keywords": ["autonomous target recognition", "ATR", "deep learning targeting",
                     "autonomous weapons AI", "machine learning fire control",
                     "neural network guidance", "lethal autonomous"],
        "priority_institutions": ["NUDT", "IACAS", "Baidu Research", "SenseTime Defense",
                                  "MIPT", "Kronstadt Technologies"],
        "current_us_trl": 6,
        "assessment_frequency": "weekly"
    }
]


# ── Semantic Scholar ──────────────────────────────────────────────────────────

def search_semantic_scholar(keywords: list, institutions: list,
                            days_back: int = 7) -> list:
    """Search Semantic Scholar for recent publications matching keywords."""
    S2_API_KEY = os.environ.get("SEMANTIC_SCHOLAR_API_KEY", "")
    headers = {"x-api-key": S2_API_KEY} if S2_API_KEY else {}

    query = " OR ".join([f'"{kw}"' for kw in keywords[:5]])  # Top 5 keywords
    cutoff = (datetime.date.today() - datetime.timedelta(days=days_back)).strftime("%Y-%m-%d")

    resp = requests.get(
        "https://api.semanticscholar.org/graph/v1/paper/search",
        headers=headers,
        params={
            "query": query,
            "fields": "title,abstract,authors,year,publicationDate,venue,externalIds,citationCount",
            "publicationDateOrYear": f"{cutoff}:",
            "limit": 50
        }
    )

    if resp.status_code != 200:
        return []

    papers = []
    for paper in resp.json().get("data", []):
        # Check for priority institution affiliation
        author_affiliations = [
            author.get("affiliations", [])
            for author in paper.get("authors", [])
        ]
        flat_affiliations = [aff for sublist in author_affiliations for aff in sublist]
        affiliation_text = " ".join([str(a) for a in flat_affiliations]).lower()

        priority_match = any(
            inst.lower() in affiliation_text
            for inst in institutions
        )

        papers.append({
            "source": "Semantic Scholar",
            "title": paper.get("title", ""),
            "abstract": paper.get("abstract", "")[:1000] if paper.get("abstract") else "",
            "authors": [a.get("name", "") for a in paper.get("authors", [])[:5]],
            "year": paper.get("year"),
            "publication_date": paper.get("publicationDate", ""),
            "venue": paper.get("venue", ""),
            "citations": paper.get("citationCount", 0),
            "doi": paper.get("externalIds", {}).get("DOI", ""),
            "priority_institution_match": priority_match,
            "affiliation_text": affiliation_text[:200]
        })

    return papers


# ── arXiv ─────────────────────────────────────────────────────────────────────

def search_arxiv(keywords: list, days_back: int = 7) -> list:
    """Search arXiv for recent preprints matching keywords."""
    import xml.etree.ElementTree as ET

    query = "+AND+".join([f'all:"{kw.replace(" ", "+")}"' for kw in keywords[:3]])
    cutoff = (datetime.datetime.now() - datetime.timedelta(days=days_back)).strftime("%Y%m%d")

    resp = requests.get(
        "http://export.arxiv.org/api/query",
        params={
            "search_query": f"({query})",
            "start": 0,
            "max_results": 50,
            "sortBy": "submittedDate",
            "sortOrder": "descending"
        }
    )

    if resp.status_code != 200:
        return []

    papers = []
    root = ET.fromstring(resp.content)
    ns = {"atom": "http://www.w3.org/2005/Atom"}

    for entry in root.findall("atom:entry", ns):
        published = entry.find("atom:published", ns)
        if not published or published.text[:8] < cutoff:
            continue

        authors = [a.find("atom:name", ns).text
                   for a in entry.findall("atom:author", ns)
                   if a.find("atom:name", ns) is not None]

        papers.append({
            "source": "arXiv (preprint)",
            "title": entry.find("atom:title", ns).text.strip() if entry.find("atom:title", ns) is not None else "",
            "abstract": entry.find("atom:summary", ns).text.strip()[:1000] if entry.find("atom:summary", ns) is not None else "",
            "authors": authors[:5],
            "publication_date": published.text[:10] if published is not None else "",
            "arxiv_id": entry.find("atom:id", ns).text if entry.find("atom:id", ns) is not None else "",
            "venue": "arXiv preprint",
            "citations": 0,  # Preprints have no citations yet
            "priority_institution_match": False  # Affiliation not in arXiv feed
        })

    return papers


# ── WIPO Patent Monitor ───────────────────────────────────────────────────────

def search_wipo_patents(keywords: list, applicant_countries: list = None,
                        days_back: int = 30) -> list:
    """Search WIPO PatentScope for international patent applications."""
    WIPO_API_KEY = os.environ.get("WIPO_API_KEY", "")
    headers = {"Authorization": f"Bearer {WIPO_API_KEY}"}

    applicant_countries = applicant_countries or ["CN", "RU"]  # Default: China, Russia
    country_filter = " OR ".join([f"AC:{c}" for c in applicant_countries])
    keyword_query = " OR ".join([f'"{kw}"' for kw in keywords[:3]])

    resp = requests.get(
        "https://patentscope.wipo.int/search/en/result.jsf",
        headers=headers,
        params={
            "query": f"({keyword_query}) AND ({country_filter})",
            "office": "PCT",
            "dateRangeField": "FP",
            "dateRangeFrom": (datetime.date.today() - datetime.timedelta(days=days_back)).strftime("%Y-%m-%d"),
            "dateRangeTo": datetime.date.today().strftime("%Y-%m-%d"),
            "resultsPerPage": 25
        }
    )

    if resp.status_code != 200:
        return []

    patents = []
    for patent in resp.json().get("results", []):
        patents.append({
            "source": "WIPO PatentScope",
            "title": patent.get("title", ""),
            "abstract": patent.get("abstract", "")[:1000],
            "applicant": patent.get("applicantName", ""),
            "applicant_country": patent.get("applicantCountry", ""),
            "filing_date": patent.get("filingDate", ""),
            "publication_number": patent.get("publicationNumber", ""),
            "ipc_classifications": patent.get("ipcClassifications", []),
            "priority_institution_match": True  # Already filtered by applicant country
        })

    return patents

# competitive_assessment_generator.py

RELEVANCY_SCORING_PROMPT = """You are a defense S&T intelligence analyst assessing
the strategic significance of a newly identified publication or patent.

TECHNOLOGY WATCH AREA: {tech_area}
CURRENT U.S. TRL FOR THIS AREA: {us_trl}/9
WATCH KEYWORDS: {keywords}

PUBLICATION/PATENT:
Title: {title}
Source: {source}
Date: {publication_date}
Authors/Applicants: {authors}
Affiliation: {affiliation}
Abstract: {abstract}

ASSESS:
1. RELEVANCE SCORE (1-10): How relevant is this to the watch area?
   10 = directly advances the technology being monitored
   5 = tangentially related
   1 = false positive

2. TECHNICAL SIGNIFICANCE: What specific technical advance does this represent?
   (Be specific — what new capability, material, method, or result is claimed?)

3. TRL ASSESSMENT: What TRL level does this work appear to represent?
   (1=Basic research, 9=Operational system proven in mission environment)

4. TRL DELTA: If TRL {us_trl} is current U.S. state, does this work:
   [AHEAD] of U.S. capability / [BEHIND] U.S. capability / [COMPARABLE] / [INSUFFICIENT DATA]

5. STRATEGIC IMPLICATIONS: What does this advance mean for U.S. military capability advantage?
   (1-2 sentences — be direct about military relevance)

6. COLLECTION PRIORITY: Should this publication receive analyst follow-up?
   [PRIORITY — analyst review recommended] / [WATCH — add to database, no immediate action] / [PASS — below threshold]

7. RELATED SOURCES: Does this cite or build on other works that should be monitored?

Return as JSON:
{{
  "relevance_score": N,
  "technical_significance": "...",
  "trl_assessment": N,
  "trl_delta": "AHEAD|BEHIND|COMPARABLE|INSUFFICIENT DATA",
  "strategic_implications": "...",
  "collection_priority": "PRIORITY|WATCH|PASS",
  "follow_up_sources": ["title or author of related works"],
  "analyst_flag": true/false
}}"""

def score_publication(pub: dict, tech_watch: dict) -> dict:
    """Score a publication for relevance to a technology watch area."""
    response = aoai_client.chat.completions.create(
        model=os.environ["AZURE_OPENAI_DEPLOYMENT"],
        messages=[{
            "role": "user",
            "content": RELEVANCE_SCORING_PROMPT.format(
                tech_area=tech_watch["area"],
                us_trl=tech_watch["current_us_trl"],
                keywords=", ".join(tech_watch["keywords"][:8]),
                title=pub.get("title", "Unknown"),
                source=pub.get("source", "Unknown"),
                publication_date=pub.get("publication_date", "Unknown"),
                authors=", ".join(pub.get("authors", [pub.get("applicant", "Unknown")])),
                affiliation=pub.get("affiliation_text", pub.get("applicant_country", "Unknown")),
                abstract=pub.get("abstract", "No abstract available")
            )
        }],
        temperature=0.0,
        max_tokens=800,
        response_format={"type": "json_object"}
    )
    score = json.loads(response.choices[0].message.content)
    return {**pub, **score, "watch_area_id": tech_watch["id"], "watch_area": tech_watch["area"]}


def generate_weekly_digest(scored_pubs: list, tech_watch: dict, period: str) -> str:
    """Generate a weekly S&T intelligence digest for a technology watch area."""

    priority_pubs = [p for p in scored_pubs if p.get("collection_priority") == "PRIORITY"]
    watch_pubs = [p for p in scored_pubs if p.get("collection_priority") == "WATCH"]

    digest_prompt = f"""Generate a weekly S&T intelligence digest for defense research staff.

TECHNOLOGY WATCH AREA: {tech_watch['area']}
MONITORING PERIOD: {period}
CURRENT U.S. TRL: {tech_watch['current_us_trl']}/9

PRIORITY PUBLICATIONS (analyst review recommended):
{json.dumps(priority_pubs[:10], indent=2, default=str)[:4000]}

WATCH PUBLICATIONS (added to database):
{len(watch_pubs)} additional publications identified — see database for details.

Generate:
## S&T WATCH DIGEST — {tech_watch['area']}
**Period:** {period} | **Classification:** UNCLASSIFIED

### KEY FINDINGS THIS PERIOD
[3-5 bullet points summarizing the most significant advances identified]

### PRIORITY PUBLICATIONS

For each priority publication:
**[Author/Applicant] — "[Title]" ([Source], [Date])**
- Technical advance: [What specifically was demonstrated or claimed]
- TRL assessment: [N]/9
- U.S. TRL delta: [AHEAD/BEHIND/COMPARABLE by N years/levels]
- Strategic significance: [1-2 sentences on military relevance]
- Recommended action: [Analyst review / Deeper technical analysis / Share with sponsor]

### TREND ANALYSIS
[Based on all publications this period: is the competitor technology trajectory
 accelerating, decelerating, or stable? Any new institutional players appearing?
 Any evidence of shifting research focus?]

### COLLECTION GAPS
[What aspects of this technology area are NOT being reported in open sources?
 What would be valuable to look for in next period?]

### RECOMMENDED SPONSOR ACTIONS
[1-3 specific recommendations for the research program based on this week's findings]

[UNCLASSIFIED — Based entirely on open-source publications and patent filings]
[Distribution: Program Director, Chief Scientist, Research Sponsor — per need-to-know]
[DRAFT — Requires Chief Scientist Review Before Distribution]"""

    response = aoai_client.chat.completions.create(
        model=os.environ["AZURE_OPENAI_DEPLOYMENT"],
        messages=[
            {"role": "system", "content": "You are a defense S&T intelligence analyst. Generate precise, technically substantive competitive technology assessments based only on the open-source information provided. Do not speculate beyond what the source material supports."},
            {"role": "user", "content": digest_prompt}
        ],
        temperature=0.1,
        max_tokens=3000
    )

    return response.choices[0].message.content


def generate_quarterly_competitive_assessment(
    tech_watch: dict,
    quarterly_pubs: list,
    prior_assessment: str = None
) -> str:
    """Generate a comprehensive quarterly competitive technology assessment."""

    top_pubs = sorted(quarterly_pubs, key=lambda x: -x.get("relevance_score", 0))[:20]
    ahead_count = sum(1 for p in quarterly_pubs if p.get("trl_delta") == "AHEAD")
    behind_count = sum(1 for p in quarterly_pubs if p.get("trl_delta") == "BEHIND")

    assessment_prompt = f"""Generate a comprehensive quarterly competitive technology assessment.

TECHNOLOGY AREA: {tech_watch['area']}
QUARTER: {datetime.date.today().strftime('%Y Q') + str((datetime.date.today().month-1)//3+1)}
TOTAL PUBLICATIONS MONITORED: {len(quarterly_pubs)}
PUBLICATIONS AHEAD OF U.S.: {ahead_count}
PUBLICATIONS BEHIND U.S.: {behind_count}
CURRENT U.S. TRL: {tech_watch['current_us_trl']}/9

TOP PUBLICATIONS THIS QUARTER:
{json.dumps(top_pubs[:10], indent=2, default=str)[:5000]}

PRIOR QUARTER ASSESSMENT SUMMARY (for trend comparison):
{prior_assessment[:500] if prior_assessment else 'First assessment — no prior baseline.'}

Generate a formal competitive technology assessment:

## COMPETITIVE TECHNOLOGY ASSESSMENT
**Technology Area:** {tech_watch['area']}
**Classification:** UNCLASSIFIED // OPEN SOURCE
**Assessment Period:** [Quarter and Year]
**Prepared For:** Research Sponsor / Program Director

### KEY JUDGMENTS
[3-5 declarative intelligence-style key judgments on the state of competitor
 technology development. Use IC analytic standards: "We assess...", "We judge...".
 Express confidence levels: High / Medium / Low confidence.]

### EXECUTIVE SUMMARY
[3-4 paragraph summary of competitor progress, U.S.-competitor TRL comparison,
 trajectory assessment, and implications for U.S. program strategy.]

### COMPETITOR PROGRAM ANALYSIS

#### Primary Competitors
For each leading competitor nation/institution:
- Institution name and affiliation
- Research focus and technical approach
- TRL estimate for key sub-components
- Publication volume trend (increasing/stable/decreasing)
- Evidence of maturation or transition from research to development

#### Emerging Players
[New institutions or actors entering this technology space this quarter]

### U.S.-COMPETITOR TRL COMPARISON
| Sub-Technology | U.S. TRL | Competitor TRL | Delta | Trend |
|----------------|----------|----------------|-------|-------|
| [Sub-component 1] | {tech_watch['current_us_trl']} | [Estimated] | [+/-N] | [↑↓→] |

### TECHNOLOGY TRAJECTORY ASSESSMENT
[Based on publication volume, TRL progression, and institutional investment signals:
 What is the projected competitor TRL in 2 years? 5 years?
 At what point might the competitor achieve operational capability?]

### GAPS AND UNCERTAINTIES
[What do we NOT know? What would significantly change this assessment if known?
 What are the key intelligence gaps that cannot be resolved from open sources?]

### IMPLICATIONS FOR U.S. PROGRAM
[Direct recommendations for the U.S. research program based on this assessment:
 - Areas where increased U.S. investment is warranted
 - Areas where the U.S. maintains sufficient lead time
 - Specific technical approaches from competitor publications worth evaluating]

### COLLECTION PRIORITIES (NEXT QUARTER)
[What sources, institutions, or publication venues should be prioritized
 for monitoring next quarter to fill identified gaps?]

### SOURCE SUMMARY
Total open-source publications analyzed: {len(quarterly_pubs)}
Period covered: [Quarter dates]
Primary sources: Semantic Scholar, arXiv, WIPO PatentScope, DTIC

[UNCLASSIFIED — All sources open-source. Cleared for unclassified distribution
 per program distribution list. Export-controlled technology areas require ECO
 review before sharing with foreign persons.]
[DRAFT — REQUIRES CHIEF SCIENTIST REVIEW AND PROGRAM DIRECTOR APPROVAL]"""

    response = aoai_client.chat.completions.create(
        model=os.environ["AZURE_OPENAI_DEPLOYMENT"],
        messages=[
            {"role": "system", "content": "You are a senior defense S&T intelligence analyst with expertise in competitive technology assessment. Apply IC analytic standards. Distinguish clearly between what is known from open sources, what is assessed, and what is uncertain."},
            {"role": "user", "content": assessment_prompt}
        ],
        temperature=0.1,
        max_tokens=4000
    )

    return response.choices[0].message.content

# Configure Azure Cognitive Search index for S&T publication database

from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
    SearchIndex, SearchField, SearchFieldDataType,
    SimpleField, SearchableField, ComplexField,
    SemanticConfiguration, SemanticSearch, SemanticPrioritizedFields,
    SemanticField
)
from azure.core.credentials import AzureKeyCredential

index_client = SearchIndexClient(
    endpoint=os.environ["AZURE_SEARCH_ENDPOINT"],
    credential=AzureKeyCredential(os.environ["AZURE_SEARCH_KEY"])
)

def create_publication_index():
    fields = [
        SimpleField(name="id", type=SearchFieldDataType.String, key=True),
        SearchableField(name="title", type=SearchFieldDataType.String, analyzer_name="en.microsoft"),
        SearchableField(name="abstract", type=SearchFieldDataType.String, analyzer_name="en.microsoft"),
        SearchableField(name="authors", type=SearchFieldDataType.Collection(SearchFieldDataType.String)),
        SimpleField(name="source", type=SearchFieldDataType.String, filterable=True, facetable=True),
        SimpleField(name="publication_date", type=SearchFieldDataType.DateTimeOffset,
                    filterable=True, sortable=True),
        SimpleField(name="watch_area_id", type=SearchFieldDataType.String,
                    filterable=True, facetable=True),
        SimpleField(name="watch_area", type=SearchFieldDataType.String, filterable=True),
        SimpleField(name="relevance_score", type=SearchFieldDataType.Double,
                    filterable=True, sortable=True),
        SimpleField(name="trl_assessment", type=SearchFieldDataType.Int32,
                    filterable=True, sortable=True),
        SimpleField(name="trl_delta", type=SearchFieldDataType.String,
                    filterable=True, facetable=True),
        SimpleField(name="collection_priority", type=SearchFieldDataType.String,
                    filterable=True, facetable=True),
        SearchableField(name="strategic_implications", type=SearchFieldDataType.String),
        SimpleField(name="priority_institution_match", type=SearchFieldDataType.Boolean,
                    filterable=True),
        SimpleField(name="applicant_country", type=SearchFieldDataType.String,
                    filterable=True, facetable=True),
        SimpleField(name="doi", type=SearchFieldDataType.String),
        SimpleField(name="arxiv_id", type=SearchFieldDataType.String),
        SimpleField(name="indexed_date", type=SearchFieldDataType.DateTimeOffset,
                    filterable=True, sortable=True)
    ]

    # Semantic search configuration for conceptual similarity matching
    semantic_config = SemanticConfiguration(
        name="publication-semantic",
        prioritized_fields=SemanticPrioritizedFields(
            title_field=SemanticField(field_name="title"),
            content_fields=[SemanticField(field_name="abstract"),
                           SemanticField(field_name="strategic_implications")]
        )
    )

    index = SearchIndex(
        name="st-publications",
        fields=fields,
        semantic_search=SemanticSearch(configurations=[semantic_config])
    )

    index_client.create_or_update_index(index)
    print("S&T publication search index created.")

create_publication_index()

SYSTEM PROMPT:
You are a defense S&T program manager preparing a technology gap briefing
for a research sponsor audience (program managers, contracting officers,
technology directors). The briefing should be 10-15 slides in structure.

Generate a slide-by-slide outline (title + 5 bullet points per slide) covering:

Slide 1: Executive Summary — Top 3 findings
Slide 2: Monitoring Methodology — Sources, scope, process
Slides 3-N: One slide per technology watch area:
  - U.S. TRL vs. competitor TRL
  - Key competitor publications/patents from the period
  - Trend (accelerating/stable/decelerating)
  - Risk to U.S. advantage
  - Recommended U.S. program response
Slide N+1: Cross-cutting themes across all watch areas
Slide N+2: Recommended investment priorities
Slide N+3: Collection priorities and gaps

TONE: Direct, analytic, briefing-style (bullets, not prose)
CLASSIFICATION: UNCLASSIFIED — suitable for unclassified sponsor briefing

ASSESSMENT INPUTS:
{quarterly_assessments}

SYSTEM PROMPT:
You are a defense S&T intelligence analyst tracking research institutions.
Review the following new institutions appearing in recent monitored publications
and assess whether they warrant addition to the priority monitoring list.

For each new institution:
1. IDENTIFICATION: What is this institution? (Government/military/civilian university/private)
2. GOVERNMENT AFFILIATION: Is it affiliated with a defense ministry or military? (Yes/No/Unclear)
3. PUBLICATION VOLUME: How many publications appeared in the monitored period?
4. TECHNICAL FOCUS: What specific sub-areas of the watch topic is this institution researching?
5. RECOMMENDATION: Add to priority list / Continue passive monitoring / No action
6. RATIONALE: Why is this institution significant (or not)?

NEW INSTITUTIONS TO ASSESS:
{new_institutions}

CURRENT PRIORITY INSTITUTION LIST:
{current_list}