Index

zocrates.agents.online_search

Online scholarly literature discovery and import agent package.

__all__ module-attribute

__all__ = ['build_online_coordinator_view', 'create_online_search_agent', 'create_online_synthesizer_agent', 'search_online_literature', 'search_online_pipeline_node']

build_online_coordinator_view

build_online_coordinator_view(text, candidates)

Build a clean Markdown coordinator view with a trailing candidate papers mapping.

Appends a # Discovered Candidate Papers bibliography mapping 1-based candidate indices ([1], [2], ...) to bibliographic details and import rationales while omitting full abstracts to bound coordinator context token usage.

Source code in zocrates/agents/online_search/search_pipeline.py

def build_online_coordinator_view(
    text: str,
    candidates: list[CandidatePaper] | None,
) -> str:
    """Build a clean Markdown coordinator view with a trailing candidate papers mapping.

    Appends a `# Discovered Candidate Papers` bibliography mapping 1-based candidate
    indices (`[1]`, `[2]`, ...) to bibliographic details and import rationales while
    omitting full abstracts to bound coordinator context token usage.
    """
    # Strip trailing candidate paper / reference headings for idempotency
    clean_text = re.sub(
        r"(?:\n{1,3}|^\s*)#{1,3}\s*(?:References|Cited Sources|Sources Consulted|Recommended Papers for Import|Discovered Candidate Papers)[\s\S]*$",
        "",
        text or "",
        flags=re.IGNORECASE,
    ).strip()
    if not candidates:
        return clean_text

    lines = ["# Discovered Candidate Papers"]
    for i, raw_c in enumerate(candidates, 1):
        extra_pdf_url = raw_c.get("pdf_url") if isinstance(raw_c, dict) else None
        c = CandidatePaper.from_dict(raw_c) if isinstance(raw_c, dict) else raw_c

        authors_str = _format_candidate_authors(getattr(c, "authors", None))
        year_str = str(c.year).strip() if getattr(c, "year", None) else "n.d."
        raw_title = str(getattr(c, "title", None) or "").strip()
        title_str = raw_title.rstrip(".") if raw_title else "Untitled"

        raw_venue = str(getattr(c, "venue", None) or "").strip().rstrip(".")
        venue_str = f" In {raw_venue}." if raw_venue else ""

        raw_doi = str(getattr(c, "doi", None) or "").strip()
        raw_url = str(
            getattr(c, "oa_pdf_url", None)
            or getattr(c, "pdf_url", None)
            or extra_pdf_url
            or ""
        ).strip()
        if raw_doi:
            doi_str = f" DOI: {raw_doi}"
        elif raw_url:
            doi_str = f" URL: {raw_url}"
        else:
            doi_str = ""

        entry = f"- [{i}] {authors_str} ({year_str}). {title_str}.{venue_str}{doi_str}"
        raw_tldr = str(getattr(c, "tldr", None) or "").strip()
        if raw_tldr:
            entry += f"\n  - Rationale: {raw_tldr}"

        lines.append(entry)

    candidate_section = "\n".join(lines)
    if clean_text:
        return f"{clean_text}\n\n{candidate_section}"
    return candidate_section

create_online_search_agent

create_online_search_agent(model=None, thinking_level='low', generate_content_config=None)

Create the stateless online literature search subagent.

Equipped with OpenAlex retrieval, candidate conclusion, and extension request tools.

Source code in zocrates/agents/online_search/agent.py

def create_online_search_agent(
    model: Any = None,
    thinking_level: str | None = "low",
    generate_content_config: types.GenerateContentConfig | None = None,
) -> Agent:
    """Create the stateless online literature search subagent.

    Equipped with OpenAlex retrieval, candidate conclusion, and extension request tools.
    """
    if model is None:
        from zocrates.llm import get_llm_model

        model = get_llm_model()

    if generate_content_config is None and thinking_level is not None:
        generate_content_config = get_agent_generate_content_config(thinking_level)

    return Agent(
        name="online_search_agent",
        model=model,
        instruction=ONLINE_SEARCH_AGENT_INSTRUCTION,
        tools=[
            FunctionTool(search_openalex),
            FunctionTool(conclude_online_search),
            FunctionTool(request_search_extension),
        ],
        generate_content_config=generate_content_config,
        mode="single_turn",
        include_contents="default",
    )

create_online_synthesizer_agent

create_online_synthesizer_agent(model=None, thinking_level='high', generate_content_config=None)

Create the online synthesizer subagent.

Produces an inquiry overview and structured candidate recommendations with numbered references.

Source code in zocrates/agents/online_search/agent.py

def create_online_synthesizer_agent(
    model: Any = None,
    thinking_level: str | None = "high",
    generate_content_config: types.GenerateContentConfig | None = None,
) -> Agent:
    """Create the online synthesizer subagent.

    Produces an inquiry overview and structured candidate recommendations with numbered references.
    """
    if model is None:
        from zocrates.llm import get_llm_model

        model = get_llm_model()

    if generate_content_config is None and thinking_level is not None:
        generate_content_config = get_agent_generate_content_config(thinking_level)

    return Agent(
        name="online_synthesizer_agent",
        model=model,
        instruction=ONLINE_SYNTHESIZER_AGENT_INSTRUCTION,
        tools=[],
        generate_content_config=generate_content_config,
        mode="single_turn",
        include_contents="none",
    )

search_online_literature async

search_online_literature(request, tool_context)

Autonomous discovery tool executing the online scholarly literature search pipeline.

Use this tool when the user wants to discover new papers, preprints, external literature, or resolve external DOIs/arXiv links not in their library.

Source code in zocrates/agents/online_search/search_pipeline.py

async def search_online_literature(request: str, tool_context: ToolContext) -> str:
    """Autonomous discovery tool executing the online scholarly literature search pipeline.

    Use this tool when the user wants to discover new papers, preprints, external literature,
    or resolve external DOIs/arXiv links not in their library.
    """
    tool_context.actions.skip_summarization = True
    fc_id = (
        getattr(tool_context, "function_call_id", None)
        or getattr(tool_context, "invocation_id", None)
        or getattr(getattr(tool_context, "_invocation_context", None), "invocation_id", None)
        or "default"
    )
    result = await tool_context.run_node(
        search_online_pipeline_node,
        node_input=request,
        override_isolation_scope=f"search_online_pipeline:{fc_id}",
    )
    return coerce_to_text(result)

search_online_pipeline_node async

search_online_pipeline_node(ctx, node_input='')

Dynamic workflow function orchestrating online literature search and synthesis.

Wraps _run_online_search_pipeline in a try/finally so the node always closes its isolation scope (search_online_pipeline:{fc_id}) before returning, whether it succeeds ("Task completed.") or raises ("Task failed.").

Source code in zocrates/agents/online_search/search_pipeline.py

@node(name="search_online_pipeline_node", rerun_on_resume=True)
async def search_online_pipeline_node(
    ctx: Context,
    node_input: str = "",
) -> str:
    """Dynamic workflow function orchestrating online literature search and synthesis.

    Wraps ``_run_online_search_pipeline`` in a try/finally so the node always closes
    its isolation scope (``search_online_pipeline:{fc_id}``) before returning, whether
    it succeeds ("Task completed.") or raises ("Task failed.").
    """
    _ensure_isolation_scope(ctx)

    completed = False
    interrupted = False
    try:
        response_text = await _run_online_search_pipeline(ctx, node_input)
        completed = True
        return response_text
    except NodeInterruptedError:
        interrupted = True
        raise
    finally:
        if not interrupted:
            await _enqueue_isolation_scope_closing_event(
                ctx,
                result="Task completed." if completed else "Task failed.",
            )