soev_agents.adapters
Connect agents to a model service, keep their traces in a store, and turn search hits into elements.
An adapter implements a port of soev_agents.ports over one technology. OpenAIChatModel needs the openai
extra and PostgresTraceStore needs the postgres extra. The package imports without them, and each of the two
names imports its library only when you use it.
Models
A model adapter is a ChatModel over the API of a model service.
OpenAIChatModel
dataclassOpenAIChatModel(client, model, settings=dict())A ChatModel over an OpenAI-compatible chat completions API.
Each call goes to the model that it names, else to model, at the base URL of the client. The adapter sends the
messages and the tools of the ChatRequest, and an image part as a base64 data URL. It reads the reasoning of
the model's output from a reasoning_content or reasoning field, which OpenAI-compatible services add. It never
sends reasoning back, because the API has no field for the reasoning of an earlier output.
When the service is down, slow or overloaded, the adapter raises TransientError after the retries of the client.
When the service refuses a call with another HTTP code below 500, the openai.APIStatusError of the SDK
propagates.
Example
from openai import AsyncOpenAI
from soev_agents.adapters import OpenAIChatModel
model = OpenAIChatModel(AsyncOpenAI(), model="gpt-4.1-mini", settings={"temperature": 0.2})paramclientAsyncOpenAIThe client of the OpenAI SDK. Its base URL, API key and retries apply to every call.
parammodelstrThe model that a call goes to when it names no model.
paramsettingsMapping[str, JsonValue]= dict()The extra fields for the body of every call, such as {"temperature": 0.2}.
methodresolve(model) -> strmethodcompleteasync (request, *, model=None) -> ModelOutputSend request to the model in one call, and return the model's output.
The output holds the content, the reasoning, the tool calls, the finish reason and the token usage of the
first choice. A finish reason other than stop, length, tool_calls and content_filter becomes None.
paramrequestChatRequestThe messages and the tools to send.
parammodelstr | None= NoneThe model to call. None calls the adapter's model.
Returns
ModelOutputThe model's output. Its model is the name that resolve returns.
Raises:
TransientError: The service failed with HTTP code 408, 409, 429 or 500 and above, the connection failed, the body did not parse, the response held no choice, or the model wrote tool arguments that are not a JSON object.openai.APIStatusError: The service refused the call with another HTTP code below 500.
methodstream_completeasync (request, *, model=None) -> AsyncIterator[Fragment | ModelOutput]Send request to the model, and yield the parts of the model's output as they arrive.
The adapter asks the service for the token usage at the end of the response, and reads only the first choice.
paramrequestChatRequestThe messages and the tools to send.
parammodelstr | None= NoneThe model to call. None calls the adapter's model.
Yields:
Fragment | ModelOutput: AReasoningDeltafor each part of the reasoning, a string for each part of the content, and at the end the completeModelOutput, ascompletereturns it.
Raises:
TransientError: The service failed as forcomplete, or the response ended without a finish reason.openai.APIStatusError: The service refused the call with another HTTP code below 500.
Trace stores
A trace store keeps the events of each thread, and the lease of the run that writes them.
InMemoryTraceStore
classInMemoryTraceStore(clock=lambda: datetime.now(UTC))TraceStore in memory, for tests and single-process serving.
An append does not await between the check of expected and its last write. So on one event loop, no other
append can come between them.
The threads are lost when the process ends. The store keeps a copy of each payload, made through JSON, so a later change to a payload object changes nothing stored.
Make an empty store.
methodappendasync (thread_id, events, *, expected=None, lease=None) -> Sequence[StoredEvent]methodreadasync (thread_id, *, after=0, follow=False) -> AsyncGenerator[LogEntry[StoredEvent], None]Yield the log entries of the thread after the position after, in the order of their positions.
paramfollowbool= FalseFalse yields the entries that exist now, and stops. True then waits, and yields each new
entry as an append adds it. A read with follow ends when the thread is deleted or loaded again.
Yields:
methodread_backwardsasync (thread_id) -> AsyncGenerator[LogEntry[StoredEvent], None]methoddeleteasync (thread_id) -> Nonemethodacquireasync (thread_id, *, owner, ttl) -> LeaseTake the lease of the thread for ttl seconds, and return it.
The lease is free when no owner holds it, when the held lease expired, or when owner holds it already.
Each acquisition adds one to the counter of the thread, so the new lease replaces the lease before it.
paramownerstrThe name of the process that takes the lease.
paramttlfloatThe seconds until the lease expires, unless the owner renews it.
Returns
LeaseThe new lease.
Raises:
LeaseHeldError: Another owner holds a lease that has not expired.
methodrenewasync (lease, *, ttl) -> Nonemethodreleaseasync (lease) -> Nonemethodleaseasync (thread_id) -> Lease | Nonemethodloadasync (thread_id, events) -> Nonemethodthreadsasync (*, limit, before=None) -> Sequence[tuple[str, StoredEvent]]Return one page of the threads, newest first, each with its first event.
Newest means the ts of the first event, then the thread id, both descending. A thread without events is
not listed.
SqliteTraceStore
classSqliteTraceStore(path, *, wake_interval=1.0, clock=lambda: datetime.now(UTC))A TraceStore in one SQLite file, which needs no separate database process.
The file holds every thread. Each statement runs in a worker of asyncio.to_thread, one at a time. A write holds
the write lock of the database for its whole check and insert, so another process that writes the same file
cannot interleave. A read with follow wakes on each append of this store, and every wake_interval seconds for
the appends of another process.
Nothing works before open, and close releases the file. async with does both. A method that you call
before open raises RuntimeError.
Make a store on the file at path. The store does not touch the file before open.
parampathstr | PathThe SQLite file. open creates the file and its tables when they do not exist.
paramwake_intervalfloat= 1.0The seconds after which read(follow=True) reads again without an append of this store,
so that the read sees what another process appended.
methodopenasync () -> NoneOpen the file, and create the file and its tables when they do not exist.
A table that an earlier release of soev-agents created gets the columns that it lacks.
Raises:
RuntimeError: The table of events lacks a column that cannot be null, so the stored rows have no value for it.
methodcloseasync () -> NoneClose the file. A store that is not open stays as it is.
methodappendasync (thread_id, events, *, expected=None, lease=None) -> Sequence[StoredEvent]methodloadasync (thread_id, events) -> Nonemethodreadasync (thread_id, *, after=0, follow=False) -> AsyncGenerator[LogEntry[StoredEvent], None]Yield the log entries of the thread after the position after, in the order of their positions.
paramfollowbool= FalseFalse yields the entries that exist now, and stops. True then waits, and yields each new
entry after an append of this store, or after wake_interval seconds. A read with follow ends when
the thread is deleted or loaded again.
Yields:
methodread_backwardsasync (thread_id) -> AsyncGenerator[LogEntry[StoredEvent], None]methodthreadsasync (*, limit, before=None) -> Sequence[tuple[str, StoredEvent]]Return one page of the threads, newest first, each with its first event.
Newest means the ts of the first event, then the thread id, both descending. A thread without events is
not listed.
methoddeleteasync (thread_id) -> Nonemethodacquireasync (thread_id, *, owner, ttl) -> LeaseTake the lease of the thread for ttl seconds, and return it.
The lease is free when no owner holds it, when the held lease expired, or when owner holds it already.
Each acquisition adds one to the counter of the thread, so the new lease replaces the lease before it.
paramownerstrThe name of the process that takes the lease.
paramttlfloatThe seconds until the lease expires, unless the owner renews it.
Returns
LeaseThe new lease.
Raises:
LeaseHeldError: Another owner holds a lease that has not expired.
methodrenewasync (lease, *, ttl) -> Nonemethodreleaseasync (lease) -> Nonemethodleaseasync (thread_id) -> Lease | NonePostgresTraceStore
classPostgresTraceStore(connection_string, *, schema, pool_size=8, wake_interval=5.0)A TraceStore on Postgres, in a schema of its own, for many processes that serve the same threads.
An append holds the row lock of the thread until it commits. So the positions of one thread are given and
become visible in order, and no other append can come between the check of expected and the insert. Each
append sends a notification, and a read with follow wakes on it. The read also wakes every
wake_interval seconds, because a pooler in transaction mode never delivers notifications.
Nothing works before open, which opens the connection pool and one more connection that listens for the
notifications. close releases both. async with does the same, and also calls setup on entry. The schema
must exist: the store creates only its tables.
The times of the leases come from the clock of the database. The ts of a batch comes from the clock of the
process.
Make a store on the database at connection_string. Nothing connects before open.
paramconnection_stringstrWhere the database is, as postgresql://user:password@host:port/database or as
the key=value pairs of libpq. The store gives it to psycopg unchanged.
paramschemastrThe Postgres schema that holds the tables of the store. Two stores on one database with different schemas hold different threads.
parampool_sizeint= 8The maximum number of connections in the pool, without the connection that listens.
paramwake_intervalfloat= 5.0The seconds after which read(follow=True) reads again without a notification, for a
database behind a pooler that drops notifications.
methodopenasync () -> NoneOpen the connection pool and the connection that listens for notifications.
open creates no tables. Call setup for them, or use async with.
methodcloseasync () -> NoneStop the connection that listens, and close the connection pool.
methodsetupasync () -> NoneCreate the tables of the schema. Call it at every process start: tables that exist stay as they are.
A table that an earlier release of soev-agents created gets the columns that it lacks.
Raises:
RuntimeError: The table of events lacks a column that cannot be null, so the stored rows have no value for it.
methodappendasync (thread_id, events, *, expected=None, lease=None) -> Sequence[StoredEvent]Seal events into the streams of the thread, and return them as stored.
The store appends all events of the batch or none, in one transaction. Every event of the batch gets the
same ts. Postgres cannot keep the character U+0000 in JSON, so the store removes it from each string and
key of the payloads. The returned events show the payloads as stored.
paramexpectedMapping[Stream, int] | NoneRaises:
methodloadasync (thread_id, events) -> Nonemethodreadasync (thread_id, *, after=0, follow=False) -> AsyncGenerator[LogEntry[StoredEvent], None]Yield the log entries of the thread after the position after, in the order of their positions.
paramfollowbool= FalseFalse yields the entries that exist now, and stops. True then waits, and yields each new
entry after a notification of an append, or after wake_interval seconds. A read with follow ends
when the thread is deleted or loaded again.
Yields:
methodread_backwardsasync (thread_id) -> AsyncGenerator[LogEntry[StoredEvent], None]Yield the log entries of the thread from the newest to the oldest.
The store reads the entries in pages of 200, and reads the next page only when the reader asks for it.
Yields:
methodthreadsasync (*, limit, before=None) -> Sequence[tuple[str, StoredEvent]]Return one page of the threads, newest first, each with its first event.
Newest means the ts of the first event, then the thread id, both descending. A thread without events is
not listed.
methoddeleteasync (thread_id) -> Nonemethodacquireasync (thread_id, *, owner, ttl) -> LeaseTake the lease of the thread for ttl seconds, and return it.
The lease is free when no owner holds it, when the held lease expired, or when owner holds it already.
Each acquisition adds one to the counter of the thread, so the new lease replaces the lease before it.
paramownerstrThe name of the process that takes the lease.
paramttlfloatThe seconds until the lease expires, unless the owner renews it.
Returns
LeaseThe new lease.
Raises:
LeaseHeldError: Another owner holds a lease that has not expired.
methodrenewasync (lease, *, ttl) -> Nonemethodreleaseasync (lease) -> Nonemethodleaseasync (thread_id) -> Lease | NoneSearch results
A search tool records the hits of a soev-retrieval search as elements.
sources_of
functionsources_ofasync (search, elements) -> list[Element]Await search and record what elements makes of each hit, in ranking order.
elements decides what a hit is: a passage after its document, a product that is its own whole. It sees the
hit's properties as JSON values (a date becomes its ISO text). A real id is recorded once per call, as the first
hit made it, so a document shared by several hits appears once, before its first passage.
A tool awaits its search through this, so a store problem reaches the model as the tool's output instead of failing the turn.
paramsearchAwaitable[Sequence[Hit]]The search of a soev-retrieval store, not yet awaited.
paramelementsCallable[[Hit], Sequence[Element]]Makes the elements of one hit, in the order to record them.
Returns
list[Element]The elements of all hits, in the order of the hits. Each id occurs once.
Raises:
TransientError: The store or a reranker raisedStoreUnavailable.InvalidQueryError: The store refused the call, also for a denial. The error keeps the message of the store.soev_retrieval.ports.Unconfigured: The store cannot serve the call as it is built.sources_ofdoes not convert it, so the turn fails.
Writing a trace store
A new TraceStore seals each batch of events as the stores of this package do.
seal
functionseal(batch, lengths, ts) -> tuple[StoredEvent, ...]Set the seq of each event to its place in its stream after lengths, and set ts.
A TraceStore calls it in append, after its checks and before it writes. The events of one stream get the
next numbers in batch order, from the length of that stream. Every event gets the same ts. The given events
do not change.
paramlengthsMapping[Stream, int]paramtsdatetimeThe time of the append.
Raises:
SealedEventError: An event ofbatchhas aseqor atsalready.