Skip to main content

Agent

The agent resource submits and manages conversational agent jobs. It is available via client.agent.

An agent job reasons over indexed files and conversation history to produce an answer. Unlike Question Answering (one answer per question) or Extraction (one value per field), the agent runs an iterative loop and can optionally return a structured answer constrained by a JSON Schema.

You can run the agent in two modes — both return the same kind of async job:

  • Stateful: pass a single new message; include a conversation_id to continue a server-side conversation (prior turns load automatically).
  • Stateless: pass the full messages list yourself; nothing is stored server-side, so each call carries its own history.

Stateful

import asyncio
from biolevate import BiolevateClient

async def main():
async with BiolevateClient(
base_url="https://<api-domain>",
token="<your-pat>",
) as client:
job = await client.agent.create_job(
message="Summarize the attached report and list the main risks.",
file_ids=["a1b2c3d4-e5f6-7890-abcd-ef1234567890"],
)
print(f"Job created: {job.job_id} (status: {job.status})")

asyncio.run(main())

Stateless

from biolevate import AgentMessage

job = await client.agent.create_job(
messages=[
AgentMessage(role="USER", content="What are the main risks in this study?"),
AgentMessage(role="ASSISTANT", content="The main risks are dropout rate, dosing variance, and small sample size."),
AgentMessage(role="USER", content="Which of those is most severe?"),
],
file_ids=["a1b2c3d4-e5f6-7890-abcd-ef1234567890"],
)
print(f"Job created: {job.job_id} (status: {job.status})")

Methods

create_job()

Submit a new agent job. Provide either message (stateful turn) or messages (stateless list); the two are mutually exclusive.

from biolevate import AgentMessage, AgentCompletionConfig

# Stateless: full message list
job = await client.agent.create_job(
messages=[
AgentMessage(role="USER", content="What are the main risks in this study?"),
],
file_ids=["a1b2c3d4-e5f6-7890-abcd-ef1234567890"],
)

# Structured output constrained by a JSON Schema
job = await client.agent.create_job(
message="Give a verdict on whether the primary endpoint was met.",
collection_ids=["c0ffee00-dead-beef-cafe-123456789abc"],
output_model_schema={
"type": "object",
"properties": {"verdict": {"type": "string"}},
"required": ["verdict"],
"additionalProperties": False,
},
completion_config=AgentCompletionConfig(model="BALANCED", temperature=0.2),
max_iterations=10,
)

# Stateful: continue an existing conversation
job = await client.agent.create_job(
message="And what about the secondary endpoints?",
conversation_id="a1b2c3d4-e5f6-7890-abcd-ef1234567890",
)
ParameterTypeDefaultDescription
messagestr | NoneNoneNew user message for a stateful turn
messageslist[AgentMessage] | NoneNoneFull list of role/content items for a stateless run
file_idslist[str] | NoneNoneFile IDs the agent can read from
collection_idslist[str] | NoneNoneCollection IDs the agent can read from
output_model_schemadict | NoneNoneStrict JSON Schema constraining the final answer (see note below)
completion_configAgentCompletionConfig | NoneNoneModel preset (FAST/BALANCED), temperature, max tokens
max_iterationsint | NoneNoneHard cap on the number of agent-loop iterations
conversation_idstr | NoneNoneContinue an existing server-side conversation (stateful)
idempotency_keystr | NoneNoneOptional idempotency key for safe retries
configJobLaunchConfig | NoneNoneJob launch behaviour, e.g. JobLaunchConfig(skip_unindexed_files=True)

See Common Patterns: Job Launch Config for details on config.

output_model_schema uses strict structured-output mode: every object must set "additionalProperties": False and list all its keys in required. There are no optional keys — to make a value optional, keep the key in required and make it nullable ("type": ["string", "null"]); it is always returned, possibly as None. See the Agent guide for a full example.

Returns Job

Raises AuthenticationError, APIError


get_job()

Get an agent job by ID.

job = await client.agent.get_job(job_id)
print(job.status)
ParameterTypeDescription
job_idstrUUID of the agent job

Returns Job

Raises NotFoundError, AuthenticationError, APIError


get_job_outputs()

Get the agent's answer for a completed job. answer may be a structured object (when output_model_schema was provided) or plain text.

outputs = await client.agent.get_job_outputs(job_id)
print(outputs.answer)
print(outputs.explanation)
print(outputs.reference_ids)

Returns AgentJobOutputs

Raises NotFoundError, AuthenticationError, APIError


get_job_inputs()

Get the inputs the job was created with.

inputs = await client.agent.get_job_inputs(job_id)
print(inputs.message)
print(inputs.max_iterations)

Returns AgentJobInputs

Raises NotFoundError, AuthenticationError, APIError


get_job_annotations()

Get the annotations linking the answer back to source passages.

annotations = await client.agent.get_job_annotations(job_id)
for annotation in annotations:
print(annotation.type, annotation.status)

Returns list[Annotation]

Raises NotFoundError, AuthenticationError, APIError


list_jobs()

List agent jobs for the current user, optionally restricted to a single conversation.

page = await client.agent.list_jobs(page=0, page_size=20)
for job in page.data:
print(f"{job.job_id}: {job.status}")

# Only jobs from one conversation
page = await client.agent.list_jobs(conversation_id="a1b2c3d4-e5f6-7890-abcd-ef1234567890")
ParameterTypeDefaultDescription
pageint0Page number (0-based)
page_sizeint20Number of jobs per page
conversation_idstr | NoneNoneRestrict results to one conversation

Returns JobPage

Raises AuthenticationError, APIError

Full Example — Create, Poll, and Read the Answer

import asyncio
from biolevate import BiolevateClient

TERMINAL_STATUSES = {"SUCCESS", "FAILED", "ABORTED"}

async def main():
async with BiolevateClient(
base_url="https://<api-domain>",
token="<your-pat>",
) as client:
job = await client.agent.create_job(
message="Summarize the attached report and list the main risks.",
file_ids=["a1b2c3d4-e5f6-7890-abcd-ef1234567890"],
)
print(f"Job created: {job.job_id}")

while True:
job = await client.agent.get_job(job.job_id)
if job.status in TERMINAL_STATUSES:
break
await asyncio.sleep(3)

if job.status == "SUCCESS":
outputs = await client.agent.get_job_outputs(job.job_id)
print("Answer:", outputs.answer)
print("References:", outputs.reference_ids)

asyncio.run(main())

Next Steps