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 aconversation_idto continue a server-side conversation (prior turns load automatically). - Stateless: pass the full
messageslist 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",
)
| Parameter | Type | Default | Description |
|---|---|---|---|
message | str | None | None | New user message for a stateful turn |
messages | list[AgentMessage] | None | None | Full list of role/content items for a stateless run |
file_ids | list[str] | None | None | File IDs the agent can read from |
collection_ids | list[str] | None | None | Collection IDs the agent can read from |
output_model_schema | dict | None | None | Strict JSON Schema constraining the final answer (see note below) |
completion_config | AgentCompletionConfig | None | None | Model preset (FAST/BALANCED), temperature, max tokens |
max_iterations | int | None | None | Hard cap on the number of agent-loop iterations |
conversation_id | str | None | None | Continue an existing server-side conversation (stateful) |
idempotency_key | str | None | None | Optional idempotency key for safe retries |
config | JobLaunchConfig | None | None | Job 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)
| Parameter | Type | Description |
|---|---|---|
job_id | str | UUID 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")
| Parameter | Type | Default | Description |
|---|---|---|---|
page | int | 0 | Page number (0-based) |
page_size | int | 20 | Number of jobs per page |
conversation_id | str | None | None | Restrict 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
- Files — index files before the agent can read them
- Question Answering — single-question answers instead of an agent loop
- Agent guide — the same workflow with REST examples
- API Reference for complete endpoint schemas