Utils and functionality

Author: msingh-openai

examples/agents_sdk/AI_Research_Assistant_Cookbook.ipynb

@@ -0,0 +1,100 @@
+# query_expansion_agent.py
+from agents import Agent, Runner
+from pydantic import BaseModel
+from typing import Literal
+from agents.run import RunConfig
+from typing import Optional
+from ..guardrails.topic_content_guardrail import ai_topic_guardrail
+
+class QueryExpansion(BaseModel):
+    """Structured output model for the query-expansion agent."""
+
+    expanded_query: str
+    questions: str
+    is_task_clear: Literal["yes", "no"]
+
+
+class QueryExpansionAgent:
+    """A wrapper class that internally creates an `agents.Agent` instance on construction.
+
+    Example
+    -------
+    >>> query_agent = QueryExpansionAgent()
+    >>> prompt = "Draft a research report on the latest trends in AI in the last 5 years."
+    >>> expanded_prompt = query_agent.task(prompt)
+    >>> print(expanded_prompt)
+    """
+
+    _DEFAULT_NAME = "Query Expansion Agent"
+
+    _DEFAULT_INSTRUCTIONS = """You are a helpful agent who receives a raw research prompt from the user.
+
+    1. Determine whether the task is sufficiently clear to proceed (“task clarity”).
+    2. If the task is missing timeframe of the research or the industry/domain, ask the user for the missing information.
+    3. If the task *is* clear, expand the prompt into a single, well-structured paragraph that makes the research request specific, actionable, and self-contained • Do **not** add any qualifiers or meta-commentary; write only the improved prompt itself.
+    4. If the task *is not* clear, generate concise clarifying questions that will make the request actionable.  Prioritize questions about:
+    • Domain / industry focus  
+    • Timeframe (e.g., current year 2025, last 5 years, last 10 years, or all time)  
+    • Any other missing constraints or desired perspectives
+
+    Return your response **exclusively** as a JSON object with these exact fields:
+
+    ```json
+    {
+    "expanded_query": "",
+    "questions": "",
+    "is_task_clear": "yes" | "no"
+    }
+    ```
+
+    When "is_task_clear" is "yes", populate "expanded_query" with the enhanced one-paragraph prompt and leave "questions" empty.
+
+    When "is_task_clear" is "no", populate "questions" with your list of clarifying questions and leave "expanded_query" empty.
+
+    """
+
+    def __init__(self, *, model: str = "o3-mini", tools: list | None = None, name: str | None = None,
+                 instructions: str | None = None, input_guardrails: list | None = None):
+
+        # Initialise the underlying `agents.Agent` with a structured `output_type` so it
+        # returns a validated `QueryExpansion` instance instead of a raw string.
+        self._agent = Agent(
+            name=name or self._DEFAULT_NAME,
+            instructions=instructions or self._DEFAULT_INSTRUCTIONS,
+            tools=tools or [],
+            model=model,
+            output_type=QueryExpansion,
+            input_guardrails=input_guardrails or [ai_topic_guardrail],
+        )
+        self._last_result = None
+
+    # ------------------------------------------------------------------
+    # Public helpers
+    # ------------------------------------------------------------------
+    
+    @property
+    def agent(self) -> Agent:  # type: ignore[name-defined]
+        """Return the underlying ``agents.Agent`` instance."""
+        return self._agent
+
+    async def task(self, prompt: str) -> QueryExpansion:
+        """Fire-and-forget API that auto-threads each call."""
+
+        cfg = RunConfig(tracing_disabled=True)
+
+        # ── First turn ──────────────────────────────────────────────
+        if self._last_result is None:
+            self._last_result = await Runner.run(
+                self._agent, prompt, run_config=cfg
+            )
+        # ── Later turns ─────────────────────────────────────────────
+        else:
+            new_input = (
+                self._last_result.to_input_list() +
+                [{"role": "user", "content": prompt}]
+            )
+            self._last_result = await Runner.run(
+                self._agent, new_input, run_config=cfg
+            )
+
+        return self._last_result.final_output  

examples/agents_sdk/ai_research_assistant_resources/agents_tools_registry/query_expansion_agent.py

@@ -0,0 +1,52 @@
+# web_page_summary_agent.py
+
+from agents import Agent, Runner, RunConfig
+
+
+class WebPageSummaryAgent:
+    _DEFAULT_NAME = "Web Page Summary Agent"
+
+    # NOTE: Placeholders will be filled at construction time so that the agent carries
+    # all contextual instructions. The task method will then only need the raw content.
+    _DEFAULT_INSTRUCTIONS = """You are an AI assistant tasked with summarising provided web page content. 
+        "Return a concise summary in {character_limit} characters or less focused on '{search_term}'. "
+        "Only output the summary text; do not include any qualifiers or metadata."""
+
+    def __init__(
+        self,
+        search_term: str,
+        character_limit: int = 1000,
+        *,
+        model: str = "gpt-4o",
+        tools: list | None = None,
+        name: str | None = None,
+        instructions: str | None = None,
+    ) -> None:
+        # Format instructions with the dynamic values unless explicitly overridden.
+        formatted_instructions = instructions or self._DEFAULT_INSTRUCTIONS.format(
+            search_term=search_term, character_limit=character_limit
+        )
+
+        self._agent = Agent(
+            name=name or self._DEFAULT_NAME,
+            instructions=formatted_instructions,
+            tools=tools or [],
+            model=model,
+        )
+
+    # ------------------------------------------------------------------
+    # Public helpers
+    # ------------------------------------------------------------------
+
+    @property
+    def agent(self) -> Agent:  # type: ignore[name-defined]
+        """Return the underlying :class:`agents.Agent` instance."""
+        
+        return self._agent
+
+    async def task(self, content: str) -> str:
+        
+        cfg = RunConfig(tracing_disabled=True)
+        
+        result = await Runner.run(self._agent, content, run_config=cfg)
+        return result.final_output

examples/agents_sdk/ai_research_assistant_resources/agents_tools_registry/web_page_summary_agent.py

@@ -0,0 +1,77 @@
+from pydantic import BaseModel
+from agents import Agent, Runner, RunConfig
+
+_NUM_SEARCH_TERMS = 5
+
+class SearchTerms(BaseModel):
+    """Structured output model for search-terms suggestions."""
+
+    Search_Queries: list[str]
+
+
+class WebSearchTermsGenerationAgent:
+    _DEFAULT_NAME = "Search Terms Agent"
+
+    # Template for building default instructions at runtime.
+    # NOTE: Double braces are used so that ``str.format`` leaves the JSON braces intact.
+    _DEFAULT_INSTRUCTIONS_TEMPLATE = """You are a helpful agent assigned a research task. Your job is to provide the top 
+        {num_search_terms} Search Queries relevant to the given topic in this year (2025). The output should be in JSON format.
+
+        Example format provided below:
+        <START OF EXAMPLE>
+        {{
+        "Search_Queries": [
+            "Top ranked auto insurance companies US 2025 by market capitalization",
+            "Geico rates and comparison with other auto insurance companies",
+            "Insurance premiums of top ranked companies in the US in 2025", 
+            "Total cost of insuring autos in US 2025", 
+            "Top customer service feedback for auto insurance in 2025"
+        ]
+        }}
+        </END OF EXAMPLE>
+        """
+    
+
+    def __init__(
+        self,
+        num_search_terms: int = _NUM_SEARCH_TERMS,
+        *, 
+        model: str = "gpt-4o",
+        tools: list | None = None,
+        name: str | None = None,
+        instructions: str | None = None,
+    ):
+        # Store number of search terms for potential future use
+        self._num_search_terms = num_search_terms
+
+        # Build default instructions if none were provided.
+        final_instructions = (
+            instructions
+            or self._DEFAULT_INSTRUCTIONS_TEMPLATE.format(
+                num_search_terms=num_search_terms
+            )
+        )
+
+        self._agent = Agent(
+            name=name or self._DEFAULT_NAME,
+            instructions=final_instructions,
+            tools=tools or [],
+            model=model,
+            output_type=SearchTerms,
+        )
+
+    # ------------------------------------------------------------------
+    # Public helpers
+    # ------------------------------------------------------------------
+
+    @property
+    def agent(self) -> Agent:  # type: ignore[name-defined]
+        """Return the underlying ``agents.Agent`` instance."""
+
+        return self._agent
+
+    async def task(self, prompt: str) -> SearchTerms:  # type: ignore[name-defined]
+        """Execute the agent synchronously and return the structured ``SearchTerms`` output."""
+        cfg = RunConfig(tracing_disabled=True)
+        result = await Runner.run(self._agent, prompt, run_config=cfg)
+        return result.final_output

examples/agents_sdk/ai_research_assistant_resources/agents_tools_registry/web_search_terms_generation_agent.py

@@ -0,0 +1,57 @@
+# topic_guardrail.py
+from typing import List
+
+from pydantic import BaseModel
+
+from agents import (  # type: ignore  (SDK imports)
+    Agent,
+    GuardrailFunctionOutput,
+    RunContextWrapper,
+    Runner,
+    TResponseInputItem,
+    input_guardrail,
+)
+
+# ---------------------------------------------------------------------------
+# 1. Tiny classifier agent → “Is this prompt about AI?”
+# ---------------------------------------------------------------------------
+
+class TopicCheckOutput(BaseModel):
+    """Structured result returned by the classifier."""
+    is_about_ai: bool          # True → prompt is AI-related
+    reasoning: str             # short rationale (useful for logs)
+
+topic_guardrail_agent = Agent(
+    name="Topic guardrail (AI)",
+    instructions=(
+        "You are a binary classifier. "
+        "Reply with is_about_ai = true **only** when the user's request is mainly "
+        "about artificial-intelligence research, applications, tooling, ethics, "
+        "policy, or market trends. "
+        "Return is_about_ai = false for all other domains (finance, biology, history, etc.)."
+    ),
+    model="gpt-4o-mini",           # lightweight, fast
+    output_type=TopicCheckOutput,
+)
+
+# ---------------------------------------------------------------------------
+# 2. Guardrail function (decorated) that wraps the classifier
+# ---------------------------------------------------------------------------
+
+@input_guardrail
+async def ai_topic_guardrail(           
+    ctx: RunContextWrapper[None],
+    agent: Agent,
+    input: str | List[TResponseInputItem],
+) -> GuardrailFunctionOutput:
+    result = await Runner.run(topic_guardrail_agent, input, context=ctx.context)
+
+    output = GuardrailFunctionOutput(
+        output_info=result.final_output,
+        tripwire_triggered=not result.final_output.is_about_ai,
+    )
+
+    return output
+
+# Optional: tidy public surface
+__all__ = ["ai_topic_guardrail", "TopicCheckOutput"]