autogen-arithmetic-agent

The autogen-arithmetic-agent worked example.

Run it from sema/:

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

Source

src/main.sema

"""Bounded AutoGen arithmetic-team parity with persisted typed state."""

assure silver

enum OutcomeKind:
    success | failure

struct Request:
    lhs: int
    rhs: int

struct Outcome:
    kind: OutcomeKind
    value: Option[int]
    error: Option[str]
    invariant (kind == OutcomeKind.success and value != None and error == None) or (kind == OutcomeKind.failure and value == None and error != None)

struct Candidate:
    outcome: Outcome
    content: str

struct Verdict:
    outcome: Outcome
    content: str

struct TeamState:
    request: Request
    candidate: Candidate
    messages: list[dict[str, any]]
    next_speaker: str
    generation: int
    invariant len(messages) == 2
    invariant next_speaker == "verifier"
    invariant generation >= 0

struct ResetTeam:
    request: Request
    messages: list[dict[str, any]]
    next_speaker: str
    generation: int
    invariant len(messages) == 0
    invariant next_speaker == "solver"
    invariant generation >= 0

struct UsedTeam:
    request: Request
    messages: list[dict[str, any]]
    generation: int
    invariant len(messages) == 3
    invariant generation >= 0

struct Execution:
    view: dict[str, any]
    used: UsedTeam

struct ResumeExecution:
    execution: Execution
    midpoint_message_count: int
    duplicate_count: int
    invariant midpoint_message_count == 2
    invariant duplicate_count >= 0

def success(value: int) -> Outcome !{}:
    return Outcome(kind=OutcomeKind.success, value=Some(value), error=None)

def failure(name: str) -> Outcome !{}:
    require len(name) > 0
    return Outcome(kind=OutcomeKind.failure, value=None, error=Some(name))

def outcome_value(outcome: Outcome) -> int !{}:
    match outcome.value:
        case Some(value):
            return value
        case None:
            return 0

def outcome_error(outcome: Outcome) -> str !{}:
    match outcome.error:
        case Some(error):
            return error
        case None:
            return ""

def outcome_tag(outcome: Outcome) -> str !{}:
    return "ok" if outcome.kind == OutcomeKind.success else "error"

def candidate_for(request: Request) -> Candidate !{}:
    if request.rhs == 0:
        outcome = failure("ZeroDivisionError")
        return Candidate(outcome=outcome, content="candidate:error:" + outcome_error(outcome))
    outcome = success(request.lhs // request.rhs)
    return Candidate(outcome=outcome, content="candidate:ok:" + str(outcome_value(outcome)))

def verdict_for(candidate: Candidate) -> Verdict !{}:
    if candidate.outcome.kind == OutcomeKind.failure:
        return Verdict(outcome=candidate.outcome,
                       content="final:error:" + outcome_error(candidate.outcome) + ":TERMINATE")
    return Verdict(outcome=candidate.outcome,
                   content="final:ok:" + str(outcome_value(candidate.outcome)) + ":TERMINATE")

@provides("agent.execute")
def scripted_model(packet: dict[str, any]) -> any !{}:
    if packet["agent"] == "solver":
        return candidate_for(packet["inputs"]["request"])
    if packet["agent"] == "verifier":
        return verdict_for(packet["inputs"]["candidate"])
    return None

agent solver(request: Request) -> Candidate by arithmetic_model:
    sem "Compute exact signed floor division or return a typed zero-division outcome"
    budget model_calls=1, tokens=64

agent verifier(candidate: Candidate) -> Verdict by verifier_model:
    sem "Verify the candidate and emit the bounded terminal verdict"
    budget model_calls=1, tokens=64

def user_event(request: Request) -> dict[str, any] !{}:
    return {"ordinal": 0, "source": "user", "kind": "TextMessage",
            "content": "floor_div " + str(request.lhs) + " " + str(request.rhs)}

def agent_event(ordinal: int, source: str, content: str) -> dict[str, any] !{}:
    return {"ordinal": ordinal, "source": source, "kind": "TextMessage", "content": content}

def outcome_payload(outcome: Outcome) -> dict[str, any] !{}:
    if outcome.kind == OutcomeKind.failure:
        return {"kind": "error", "value": None, "error": outcome_error(outcome)}
    return {"kind": "ok", "value": outcome_value(outcome), "error": None}

def outcome_from_payload(raw: dict[str, any]) -> Outcome !{}:
    require raw["kind"] == "ok" or raw["kind"] == "error"
    if raw["kind"] == "ok":
        ensure raw["value"] != None
        ensure raw["error"] == None
        return success(raw["value"])
    ensure raw["value"] == None
    ensure raw["error"] != None
    ensure len(raw["error"]) > 0
    return failure(raw["error"])

def save_midpoint(request: Request, candidate: Candidate, generation: int) -> TeamState !{}:
    return TeamState(request=request, candidate=candidate,
                     messages=[user_event(request), agent_event(1, "solver", candidate.content)],
                     next_speaker="verifier", generation=generation)

def serialize_midpoint(state: TeamState) -> str !{}:
    return json.dumps({"request": {"lhs": state.request.lhs, "rhs": state.request.rhs},
                       "candidate": {"outcome": outcome_payload(state.candidate.outcome),
                                     "content": state.candidate.content},
                       "messages": state.messages, "next_speaker": state.next_speaker,
                       "generation": state.generation})

def loaded_event(raw: dict[str, any], ordinal: int, source: str) -> dict[str, any] !{}:
    require raw["ordinal"] == ordinal
    require raw["source"] == source
    require raw["kind"] == "TextMessage"
    require len(raw["content"]) > 0
    return agent_event(ordinal, source, raw["content"])

def load_midpoint(encoded: str) -> TeamState !{}:
    raw = json.loads(encoded)
    ensure raw["next_speaker"] == "verifier"
    ensure raw["generation"] >= 0
    request = Request(lhs=raw["request"]["lhs"], rhs=raw["request"]["rhs"])
    outcome = outcome_from_payload(raw["candidate"]["outcome"])
    candidate = Candidate(outcome=outcome, content=raw["candidate"]["content"])
    ensure candidate.content == candidate_for(request).content
    ensure len(raw["messages"]) == 2
    first = loaded_event(raw["messages"][0], 0, "user")
    second = loaded_event(raw["messages"][1], 1, "solver")
    ensure first == user_event(request)
    ensure second == agent_event(1, "solver", candidate.content)
    return TeamState(request=request, candidate=candidate, messages=[first, second],
                     next_speaker="verifier", generation=raw["generation"])

def final_payload(outcome: Outcome) -> dict[str, any] !{}:
    if outcome.kind == OutcomeKind.failure:
        return {"status": "error", "value": None, "error": outcome_error(outcome)}
    return {"status": "ok", "value": outcome_value(outcome), "error": None}

def agent_message_count(messages: list[dict[str, any]]) -> int !{}:
    mut count = 0
    for message in messages:
        if message["source"] == "solver" or message["source"] == "verifier":
            count = count + 1
    return count

def duplicate_messages(prior: list[dict[str, any]], resumed: list[dict[str, any]]) -> int !{}:
    mut count = 0
    for later in resumed:
        for earlier in prior:
            if later["source"] == earlier["source"] and later["content"] == earlier["content"]:
                count = count + 1
    return count

def run_view(messages: list[dict[str, any]], verdict: Verdict) -> dict[str, any] !{}:
    return {"events": messages, "final": final_payload(verdict.outcome),
            "stop_reason": "text_mention", "message_count": len(messages),
            "model_calls": agent_message_count(messages)}

def initial_team(request: Request) -> ResetTeam !{}:
    return ResetTeam(request=request, messages=[], next_speaker="solver", generation=0)

def execute_from_start(team: ResetTeam) -> Execution !{model.invoke}:
    require team.next_speaker == "solver"
    candidate = solver(team.request)
    verdict = verifier(candidate)
    messages = [user_event(team.request), agent_event(1, "solver", candidate.content),
                agent_event(2, "verifier", verdict.content)]
    return Execution(view=run_view(messages, verdict),
                     used=UsedTeam(request=team.request, messages=messages,
                                   generation=team.generation))

def uninterrupted(request: Request) -> Execution !{model.invoke}:
    return execute_from_start(initial_team(request))

def resumed(request: Request) -> ResumeExecution !{model.invoke}:
    candidate = solver(request)
    encoded = serialize_midpoint(save_midpoint(request, candidate, 0))
    loaded = load_midpoint(encoded)
    verdict = verifier(loaded.candidate)
    resumed_messages = [agent_event(2, loaded.next_speaker, verdict.content)]
    midpoint_count = len(loaded.messages)
    duplicates = duplicate_messages(loaded.messages, resumed_messages)
    mut messages = loaded.messages
    messages.append(resumed_messages[0])
    execution = Execution(view=run_view(messages, verdict),
                          used=UsedTeam(request=loaded.request, messages=messages,
                                        generation=loaded.generation))
    return ResumeExecution(execution=execution, midpoint_message_count=midpoint_count,
                           duplicate_count=duplicates)

def reset_team(used: UsedTeam) -> ResetTeam !{}:
    return ResetTeam(request=Request(lhs=used.request.lhs, rhs=used.request.rhs),
                     messages=[], next_speaker="solver", generation=used.generation + 1)

def rerun_after_reset(used: UsedTeam) -> Execution !{model.invoke}:
    return execute_from_start(reset_team(used))

circuit case_paths(request: Request) -> dict[str, any] !{model.invoke}:
    budget agents=6, spawn_depth=0, model_calls=6, tokens=512
    full = uninterrupted(request)
    restored = resumed(request)
    reset = rerun_after_reset(full.used)
    ensure full.view == restored.execution.view
    ensure full.view == reset.view
    ensure reset.used.generation == full.used.generation + 1
    save_load_equal = full.view == restored.execution.view
    reset_equal = full.view == reset.view
    return {"id": str(request.lhs) + "//" + str(request.rhs),
            "input": {"lhs": request.lhs, "rhs": request.rhs},
            "uninterrupted": full.view, "resumed": restored.execution.view, "reset": reset.view,
            "midpoint_stop_reason": "max_messages",
            "midpoint_message_count": restored.midpoint_message_count,
            "resume_duplicate_count": restored.duplicate_count,
            "save_load_equal": save_load_equal, "reset_equal": reset_equal}

def corpus() -> dict[str, any] !{model.invoke}:
    inputs = [Request(lhs=7, rhs=3), Request(lhs=-7, rhs=3),
              Request(lhs=7, rhs=-3), Request(lhs=-7, rhs=-3),
              Request(lhs=0, rhs=5), Request(lhs=5, rhs=0)]
    mut cases = []
    for request in inputs:
        cases.append(case_paths(request))
    return {"schema": "autogen-arithmetic-parity/v1", "cases": cases}

test "signed floor division uses Python semantics":
    ensure outcome_value(candidate_for(Request(lhs=7, rhs=3)).outcome) == 2
    ensure outcome_value(candidate_for(Request(lhs=-7, rhs=3)).outcome) == -3
    ensure outcome_value(candidate_for(Request(lhs=7, rhs=-3)).outcome) == -3
    ensure outcome_value(candidate_for(Request(lhs=-7, rhs=-3)).outcome) == 2

test "division by zero is a mutually exclusive typed outcome":
    outcome = candidate_for(Request(lhs=5, rhs=0)).outcome
    ensure outcome.kind == OutcomeKind.failure
    ensure outcome.value == None
    ensure outcome_error(outcome) == "ZeroDivisionError"

test "midpoint crosses JSON and reset consumes used state":
    request = Request(lhs=-7, rhs=3)
    saved = save_midpoint(request, candidate_for(request), 0)
    encoded = serialize_midpoint(saved)
    loaded = load_midpoint(encoded)
    ensure serialize_midpoint(loaded) == encoded
    used = UsedTeam(request=request,
                    messages=[user_event(request), agent_event(1, "solver", saved.candidate.content),
                              agent_event(2, "verifier", verdict_for(saved.candidate).content)],
                    generation=0)
    reset = reset_team(used)
    ensure reset.generation == 1
    ensure len(reset.messages) == 0
    ensure reset.next_speaker == "solver"

def main() -> str !{model.invoke}:
    return json.dumps(corpus())

Reflected API

main

Bounded AutoGen arithmetic-team parity with persisted typed state.

enum OutcomeKind

Variants

• success
• failure

struct Request

Fields

field	type	descriptor
lhs	int
rhs	int

struct Outcome

Fields

field	type	descriptor
kind	OutcomeKind
value	Option[int]
error	Option[str]

struct Candidate

Fields

field	type	descriptor
outcome	Outcome
content	str

struct Verdict

Fields

field	type	descriptor
outcome	Outcome
content	str

struct TeamState

Fields

field	type	descriptor
request	Request
candidate	Candidate
messages	list[dict[str, any]]
next_speaker	str
generation	int

struct ResetTeam

Fields

field	type	descriptor
request	Request
messages	list[dict[str, any]]
next_speaker	str
generation	int

struct UsedTeam

Fields

field	type	descriptor
request	Request
messages	list[dict[str, any]]
generation	int

struct Execution

Fields

field	type	descriptor
view	dict[str, any]
used	UsedTeam

struct ResumeExecution

Fields

field	type	descriptor
execution	Execution
midpoint_message_count	int
duplicate_count	int

def success

def success(value: int) -> Outcome !{}

Parameters

name	type
value	int

Returns Outcome

Effects !{}

def failure

def failure(name: str) -> Outcome !{}

Parameters

name	type
name	str

Returns Outcome

Effects !{}

def outcome_value

def outcome_value(outcome: Outcome) -> int !{}

Parameters

name	type
outcome	Outcome

Returns int

Effects !{}

def outcome_error

def outcome_error(outcome: Outcome) -> str !{}

Parameters

name	type
outcome	Outcome

Returns str

Effects !{}

def outcome_tag

def outcome_tag(outcome: Outcome) -> str !{}

Parameters

name	type
outcome	Outcome

Returns str

Effects !{}

def candidate_for

def candidate_for(request: Request) -> Candidate !{}

Parameters

name	type
request	Request

Returns Candidate

Effects !{}

def verdict_for

def verdict_for(candidate: Candidate) -> Verdict !{}

Parameters

name	type
candidate	Candidate

Returns Verdict

Effects !{}

def scripted_model

def scripted_model(packet: dict[str, any]) -> any !{}

Parameters

name	type
packet	dict[str, any]

Returns any

Effects !{}

agent solver

agent solver(request: Request) -> Candidate

Parameters

name	type
request	Request

Returns Candidate

agent verifier

agent verifier(candidate: Candidate) -> Verdict

Parameters

name	type
candidate	Candidate

Returns Verdict

def user_event

def user_event(request: Request) -> dict[str, any] !{}

Parameters

name	type
request	Request

Returns dict[str, any]

Effects !{}

def agent_event

def agent_event(ordinal: int, source: str, content: str) -> dict[str, any] !{}

Parameters

name	type
ordinal	int
source	str
content	str

Returns dict[str, any]

Effects !{}

def outcome_payload

def outcome_payload(outcome: Outcome) -> dict[str, any] !{}

Parameters

name	type
outcome	Outcome

Returns dict[str, any]

Effects !{}

def outcome_from_payload

def outcome_from_payload(raw: dict[str, any]) -> Outcome !{}

Parameters

name	type
raw	dict[str, any]

Returns Outcome

Effects !{}

def save_midpoint

def save_midpoint(request: Request, candidate: Candidate, generation: int) -> TeamState !{}

Parameters

name	type
request	Request
candidate	Candidate
generation	int

Returns TeamState

Effects !{}

def serialize_midpoint

def serialize_midpoint(state: TeamState) -> str !{}

Parameters

name	type
state	TeamState

Returns str

Effects !{}

def loaded_event

def loaded_event(raw: dict[str, any], ordinal: int, source: str) -> dict[str, any] !{}

Parameters

name	type
raw	dict[str, any]
ordinal	int
source	str

Returns dict[str, any]

Effects !{}

def load_midpoint

def load_midpoint(encoded: str) -> TeamState !{}

Parameters

name	type
encoded	str

Returns TeamState

Effects !{}

def final_payload

def final_payload(outcome: Outcome) -> dict[str, any] !{}

Parameters

name	type
outcome	Outcome

Returns dict[str, any]

Effects !{}

def agent_message_count

def agent_message_count(messages: list[dict[str, any]]) -> int !{}

Parameters

name	type
messages	list[dict[str, any]]

Returns int

Effects !{}

def duplicate_messages

def duplicate_messages(prior: list[dict[str, any]], resumed: list[dict[str, any]]) -> int !{}

Parameters

name	type
prior	list[dict[str, any]]
resumed	list[dict[str, any]]

Returns int

Effects !{}

def run_view

def run_view(messages: list[dict[str, any]], verdict: Verdict) -> dict[str, any] !{}

Parameters

name	type
messages	list[dict[str, any]]
verdict	Verdict

Returns dict[str, any]

Effects !{}

def initial_team

def initial_team(request: Request) -> ResetTeam !{}

Parameters

name	type
request	Request

Returns ResetTeam

Effects !{}

def execute_from_start

def execute_from_start(team: ResetTeam) -> Execution !{model.invoke}

Parameters

name	type
team	ResetTeam

Returns Execution

Effects !{model.invoke}

def uninterrupted

def uninterrupted(request: Request) -> Execution !{model.invoke}

Parameters

name	type
request	Request

Returns Execution

Effects !{model.invoke}

def resumed

def resumed(request: Request) -> ResumeExecution !{model.invoke}

Parameters

name	type
request	Request

Returns ResumeExecution

Effects !{model.invoke}

def reset_team

def reset_team(used: UsedTeam) -> ResetTeam !{}

Parameters

name	type
used	UsedTeam

Returns ResetTeam

Effects !{}

def rerun_after_reset

def rerun_after_reset(used: UsedTeam) -> Execution !{model.invoke}

Parameters

name	type
used	UsedTeam

Returns Execution

Effects !{model.invoke}

circuit case_paths

circuit case_paths(request: Request) -> dict[str, any] !{model.invoke}

Parameters

name	type
request	Request

Returns dict[str, any]

Effects !{model.invoke}

def corpus

def corpus() -> dict[str, any] !{model.invoke}

Returns dict[str, any]

Effects !{model.invoke}

def main

def main() -> str !{model.invoke}

Returns str

Effects !{model.invoke}