Decision Examples

#Decision Examples

These examples describe the prepared decision-feature release. Use an SDK build with maivn.decisions and set your ordinary MAIVN_API_KEY in the process environment. Enabled terminal calls contact the platform; they are not free offline demonstrations.

Run the Python blocks in order in one script; later blocks reuse the first definitions. Use async terminal methods from an async event loop. The guide explains privacy, cost and fallback rules, and the complete reference covers every exported type and option.

#Define an authorized request and fallback

maivn.decisions answers categorical questions through the mAIvn platform and falls back to your own baseline when a decision is unavailable, refused or not confident enough. The SDK never calls a decision model directly: requests go to POST /v1/decisions with the same API key and MAIVN_BASE_URL as every other call. The platform chooses the decision model, shields explicitly marked private values and bills the call to your project.

from maivn.decisions import (
    DecisionAnswer, DecisionChoice, DecisionClient, DecisionConfig, DecisionRequest,
)

question = DecisionRequest(
    name='urgent',
    state='Public bulletin: planned API maintenance on Saturday.',
    instructions='Does this bulletin need immediate action?',
    privacy='public',
    choices=(
        DecisionChoice(key='yes', description='Immediate action'),
        DecisionChoice(key='no', description='No immediate action'),
    ),
)


async def baseline(request: DecisionRequest) -> DecisionAnswer:
    return DecisionAnswer(value='yes', provider='local', token_category=None)


decisions = DecisionClient(
    fallback=baseline,
    config=DecisionConfig(min_confidence=0.8),
    authorize=lambda request: request == question,
)
result = decisions.decide(question)
print(result.value, result.source, result.primary.status)

Only requests declared privacy='public' and approved by authorize are sent. Mark private values inside otherwise public text with DecisionRequest(..., private_data={'customer_name': 'Priya Raman'}): the platform replaces them with placeholders before the decision model sees anything. A placeholder verdict is resolved locally and must still match a declared choice. Raw marked values reach the platform shield, never the decision model. Plain unmarked text receives no automatic redaction. Your fallback receives the original request and must enforce your application's privacy policy. Pass client=Client(...) to reuse an existing client's settings.

#Ordered native batch

batch_requests = (question, question.model_copy(update={'name': 'urgent_again'}))
batch_client = DecisionClient(
    fallback=baseline, config=DecisionConfig(),
    authorize=lambda request: request in batch_requests,
)
results = batch_client.batch(batch_requests, max_concurrency=2, native_batch=True)
print([result.source for result in results])

batch / abatch return results in input order. max_concurrency=4 bounds outstanding work by default and must be a positive integer. native_batch=False runs scalar decisions; True combines questions with identical state and private_data, splitting at max_batch_questions. native_batch_question_limit reports that limit (one when the primary is disabled). Approve every full request in your authorization policy. If any member of a shared packet is not approved, that packet falls back without dispatch. Private or unknown requests fall back individually. On caller cancellation or failure, outstanding batch tasks are cancelled and drained.

A native packet has one token receipt. The first member owns it; followers expose usage_scope='batch', the batch size and usage_owner=False, without duplicating usage. Aggregate the whole result collection to retain the owned receipt.

#Gate and review an invocation

Agent.decide and Swarm.decide return a DecisionInvocationBuilder. The same operation composes with their structured-output, event and follow-up builders. Stage names and question labels start with a letter and contain letters, digits, underscores, dots or hyphens, up to 128 characters; stage names must be unique.

from maivn import Agent

plan = (
    Agent(name='Bulletin assistant')
    .decide(
        'triage', client=decisions, questions={'urgent': question},
        require={'urgent': 'yes'},
    )
    .decide('review', questions={'urgent': question}, at='after')
)
outcome = plan.invoke('Summarize the public maintenance bulletin.')
print(outcome.status, outcome.skipped_by)
if outcome.response is not None:
    print(outcome.response.response)

Each .decide(name, client=..., questions=..., at='before', require=None, max_concurrency=4, native_batch=False) stage has 1–32 questions. Subsequent stages may omit client to reuse the previous stage's client. DecisionQuestion values are requests or synchronous factories taking DecisionContext. Each factory must explicitly construct its own request. Context contains the original input (also original_input), output (absent before invocation), and prior nested decisions[stage][question]. Context does not authorize exporting input or output.

A require gate skips invocation only when an accepted primary answer differs from the required choice. Any fallback preserves normal invocation, even when its value differs. After stages annotate the completed response and cannot have require. .map_input(mapper) explicitly changes invocation input after before stages; without it, decisions do not change messages.

invoke / ainvoke return DecisionInvocationResult: status is completed or skipped, response is an InvokeResponse or None, decisions is the nested result mapping, and skipped_by identifies a blocking stage. batch / abatch accept an iterable of inputs and max_concurrency=4, preserve order and share a bound across decisions and invocations. Invocation keyword arguments pass through to the bound scope.

.structured_output(Model), .events(include=None, exclude=None, on_event=None, auto_verbose=True) and .allow_followup_questions(**kwargs) retain the decision stages and configure the existing invocation wrappers.

stream / astream forward ordinary invocation events and add decision.stage.started, decision.stage.completed and decision.plan.completed. The final DecisionStreamEvent.outcome holds the typed result locally and is excluded from serialized event data; read it before discarding the event.

from maivn.decisions import DecisionStreamEvent

for event in plan.stream('Summarize the public maintenance bulletin.'):
    if isinstance(event, DecisionStreamEvent) and event.outcome is not None:
        print(event.outcome.status)

#Prepare explicit arguments

from maivn.decisions import DecisionArgument, DecisionArgumentPreparer

prepared = DecisionArgumentPreparer(decisions).prepare([
    DecisionArgument(
        name='notify', question=question, values={'yes': True, 'no': False},
    ),
])
# Preserve the existing value when the optional decision falls back.
notify = prepared.values.get('notify', True)
print(notify, prepared.decisions['notify'].source)

DecisionArgument(name, question, values) maps every choice exactly once to a boolean or string; its name must be a valid Python identifier, excluding keywords. DecisionArgumentPreparer.prepare / aprepare accept unique named arguments, max_concurrency=4 and native_batch=False. DecisionArgumentPreparation.values contains only accepted primary mappings; fallbacks leave the field absent so your existing argument preparation can continue. .decisions retains every result. It does not register a tool or automatically inject arguments into one.

#More patterns

See function/local-tool dependencies and optional candidate matching for complete runnable patterns. Matching stays disabled until explicitly enabled. Duplicate questions in the batch example illustrate receipt ownership; use distinct instructions for useful application questions.