research-agent

A bounded research agent: loop … until, budgets, monitors, and tool calling.

Run it from sema/:

sema check examples/research-agent
SEMA_STRICT=1 sema run examples/research-agent
sema assure examples/research-agent --grade silver

Source

src/main.sema

"""research-agent — a small end-to-end pipeline built on the Sema standard library.

A compact replica of a search→write research flow, showing the neurosymbolic
components composing in real Sema app code — all imported from `std.*`, compiled
and run:

  • provenance   — assign global citation ids + rewrite markers (std.provenance)
  • semantic     — de-duplicate candidate facts (semantic.dedup verb)
  • belief       — bounded evidence iteration with an explicit threshold (std.belief)
  • metering     — track model spend ambiently (`with meter`)
  • document     — render a typed report to markdown (std.document)

Run:  sema run examples/research-agent
"""

assure silver

from std.provenance import Cit, Doc, build_url_to_id, rewrite
from std.belief import Belief
from std.document import Report, render
from std.collections import join_str


struct BeliefRun:
    confidence: f64
    iterations: int
    reached: bool


struct Synthesis:
    text: str
    model_calls: int
    cost: f64


def source_docs() -> list[Doc] !{}:
    return [
        Doc(text="Solar capacity grew [1]. Costs fell [2].",
            citations=[Cit(url="iea.org", start=20, end=23), Cit(url="irena.org", start=36, end=39)]),
        Doc(text="Costs fell sharply [1].",
            citations=[Cit(url="irena.org", start=19, end=22)]),
    ]


def rewritten_sections(docs: list[Doc]) -> list[str] !{}:
    ids = build_url_to_id(docs)
    mut sections: list[str] = []
    for d in docs:
        sections.append(rewrite(d.text, d.citations, ids))
    return sections


def canonical_facts() -> list[str] !{model.embed}:
    ensure result == ["costs fell", "capacity grew"]
    return semantic.dedup(["costs fell", "costs fell", "capacity grew"], 0.99)


def clamped_threshold(threshold: f64) -> f64 !{}:
    ensure result >= 0.0 and result <= 1.0
    if threshold < 0.0:
        return 0.0
    if threshold > 1.0:
        return 1.0
    return threshold


def evaluate_belief(threshold: f64) -> BeliefRun !{}:
    ensure result.iterations >= 0 and result.iterations <= 3
    ensure result.reached == (result.confidence >= clamped_threshold(threshold))
    ensure result.reached or result.iterations == 3
    target = clamped_threshold(threshold)
    decisions = [0.7, 0.85, 0.95]
    mut belief = Belief(alpha=1.0, beta=1.0, history=[0.5])
    mut iters = 0
    while iters < len(decisions) and iters < 3 and belief.confidence() < target:
        belief.update(decisions[iters])
        iters = iters + 1
    confidence = belief.confidence()
    return BeliefRun(confidence=confidence, iterations=iters, reached=confidence >= target)


@provides("generate")
def deterministic_generate(_prompt: str, _max_tokens: int) -> str !{}:
    ensure result == "Renewable capacity is growing while costs decline."
    return "Renewable capacity is growing while costs decline."


def synthesize() -> Synthesis !{model.invoke}:
    ensure result.text == "Renewable capacity is growing while costs decline."
    ensure result.model_calls == 1
    ensure result.cost >= 0.0
    mut text = ""
    mut calls = 0
    mut cost = 0.0
    with meter as usage:
        text = generate("Summarize renewable energy findings", 64)
        calls = usage.total_calls
        cost = usage.cost
    return Synthesis(text=text, model_calls=calls, cost=cost)


def research_report(sections: list[str], takeaways: list[str], confidence: f64) -> Report !{}:
    return Report(
        title="Renewable Energy Findings",
        context="Auto-synthesized from 2 sources.",
        confidence=confidence,
        rationale="Confidence is a Beta-Bernoulli posterior over iteration evidence.",
        takeaways=takeaways,
        section_titles=["Findings"],
        sections_text=join_str(sections, "\n\n"),
        conclusion="Costs continue to decline as capacity scales.",
    )


def main() -> None !{model.invoke, model.embed, observe.record}:
    # 1. Sources with local citations → stable global ids + rewritten text.
    docs = source_docs()
    sections = rewritten_sections(docs)

    # 2. Candidate facts, de-duplicated semantically.
    unique_facts = canonical_facts()

    # 3. Belief-driven loop: iterate until confidence crosses the threshold,
    #    bounded by the three available evidence values.
    belief = evaluate_belief(0.7)

    # 4. Draft a synthesis with ambient usage metering — no usage tuples.
    synthesis = synthesize()

    # 5. Render a typed report to markdown.
    r = research_report(sections, unique_facts, belief.confidence)
    print(render(r, "\n"))
    log.info("run", stopped_iter=belief.iterations, confidence=belief.confidence, model_calls=synthesis.model_calls, cost=synthesis.cost)


test "citation ids are first-seen global ids and repeated URLs share one id":
    docs = source_docs()
    ids = build_url_to_id(docs)
    ensure len(ids) == 2
    ensure ids["iea.org"] == 1
    ensure ids["irena.org"] == 2


test "citation markers rewrite left to right without changing surrounding prose":
    sections = rewritten_sections(source_docs())
    ensure len(sections) == 2
    ensure sections == ["Solar capacity grew [1]. Costs fell [2].", "Costs fell sharply [2]."]


test "semantic dedup preserves the first representative in canonical order":
    ensure canonical_facts() == ["costs fell", "capacity grew"]


test "belief stopping honors the threshold and the three-evidence bound":
    initial = evaluate_belief(0.5)
    boundary = evaluate_belief(0.7)
    above = evaluate_belief(0.71)
    ensure initial.iterations == 0 and initial.reached
    ensure boundary.iterations == 3 and boundary.reached
    ensure boundary.confidence >= 0.7
    ensure above.iterations == 3 and not above.reached
    ensure above.confidence < 0.71


test "rendered report contains its title citation markers and takeaways":
    belief = evaluate_belief(0.7)
    markdown = render(research_report(rewritten_sections(source_docs()), canonical_facts(), belief.confidence), "\n")
    ensure markdown.startswith("# Renewable Energy Findings\n")
    ensure "[1]" in markdown and "[2]" in markdown
    ensure "## Key Takeaways" in markdown
    ensure "* costs fell" in markdown and "* capacity grew" in markdown


test "fixture generation and ambient metering are deterministic":
    first = synthesize()
    second = synthesize()
    ensure first.text == "Renewable capacity is growing while costs decline."
    ensure first.text == second.text
    ensure first.model_calls == 1 and second.model_calls == 1
    ensure first.cost == second.cost

Reflected API

main

research-agent — a small end-to-end pipeline built on the Sema standard library.

A compact replica of a search→write research flow, showing the neurosymbolic components composing in real Sema app code — all imported from std.*, compiled and run:

• provenance — assign global citation ids + rewrite markers (std.provenance) • semantic — de-duplicate candidate facts (semantic.dedup verb) • belief — bounded evidence iteration with an explicit threshold (std.belief) • metering — track model spend ambiently (with meter) • document — render a typed report to markdown (std.document)

Run: sema run examples/research-agent

struct BeliefRun

Fields

field	type	descriptor
confidence	f64
iterations	int
reached	bool

struct Synthesis

Fields

field	type	descriptor
text	str
model_calls	int
cost	f64

def source_docs

def source_docs() -> list[Doc] !{}

Returns list[Doc]

Effects !{}

def rewritten_sections

def rewritten_sections(docs: list[Doc]) -> list[str] !{}

Parameters

name	type
docs	list[Doc]

Returns list[str]

Effects !{}

def canonical_facts

def canonical_facts() -> list[str] !{model.embed}

Returns list[str]

Effects !{model.embed}

def clamped_threshold

def clamped_threshold(threshold: f64) -> f64 !{}

Parameters

name	type
threshold	f64

Returns f64

Effects !{}

def evaluate_belief

def evaluate_belief(threshold: f64) -> BeliefRun !{}

Parameters

name	type
threshold	f64

Returns BeliefRun

Effects !{}

def deterministic_generate

def deterministic_generate(_prompt: str, _max_tokens: int) -> str !{}

Parameters

name	type
_prompt	str
_max_tokens	int

Returns str

Effects !{}

def synthesize

def synthesize() -> Synthesis !{model.invoke}

Returns Synthesis

Effects !{model.invoke}

def research_report

def research_report(sections: list[str], takeaways: list[str], confidence: f64) -> Report !{}

Parameters

name	type
sections	list[str]
takeaways	list[str]
confidence	f64

Returns Report

Effects !{}

def main

def main() -> None !{model.invoke, model.embed, observe.record}

Returns None

Effects !{model.invoke, model.embed, observe.record}