Skip to content

trading_agent.backends.openai_compatible_backend

A backend for OpenAI, or any server that speaks OpenAI's Chat Completions protocol.

That covers OpenAI itself and local servers such as vLLM, LM Studio and llama.cpp. Tool calls stream in pieces: the arguments arrive as fragments of a JSON string, which are joined and parsed once the reply is complete.

Typical usage example:

backend = OpenAICompatibleBackend('openai', backend_configuration, api_key) reply = await backend.complete(conversation, tool_definitions, on_text)

OpenAICompatibleBackend 

OpenAICompatibleBackend(name, backend_configuration, api_key)

Bases: ModelBackend

One model on one OpenAI-compatible server.

Creates the backend without contacting the server.

Parameters:

Name Type Description Default
name str

The backend's name in trading_agent.toml.

 required
backend_configuration BackendConfiguration

Its settings, of kind "openai_compatible".

 required
api_key str

The server's API key, or an empty string for a local server that needs none.

 required
Source code in src/trading_agent/backends/openai_compatible_backend.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def __init__(
    self,
    name: str,
    backend_configuration: agent_configuration.BackendConfiguration,
    api_key: str,
):
    """Creates the backend without contacting the server.

    Args:
        name (str): The backend's name in trading_agent.toml.
        backend_configuration (agent_configuration.BackendConfiguration): Its settings, of kind "openai_compatible".
        api_key (str): The server's API key, or an empty string for a local server that needs none.
    """
    super().__init__(name, backend_configuration.model)
    self._vision = backend_configuration.vision
    self._client = openai.AsyncOpenAI(
        base_url=backend_configuration.base_url,
        api_key=api_key or _PLACEHOLDER_API_KEY,
        max_retries=0,
    )

complete async 

complete(conversation, tool_definitions, on_text)

Streams one reply from the server.

Parameters:

Name Type Description Default
conversation Conversation

The conversation so far.

 required
tool_definitions list[dict[str, Any]]

The tools the model may call.

 required
on_text Callable[[str], None]

Called with each piece of text as it streams in.

 required

Returns:

Type Description
ModelReply

The complete reply.

Raises:

Type Description
BackendUnavailableError

The server cannot be reached, refused the request, or is rate limiting.
Source code in src/trading_agent/backends/openai_compatible_backend.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
async def complete(
    self,
    conversation: conversation_module.Conversation,
    tool_definitions: list[dict[str, Any]],
    on_text: Callable[[str], None],
) -> backend.ModelReply:
    """Streams one reply from the server.

    Args:
        conversation (conversation_module.Conversation): The conversation so far.
        tool_definitions (list[dict[str, Any]]): The tools the model may call.
        on_text (Callable[[str], None]): Called with each piece of text as it streams in.

    Returns:
        backend.ModelReply: The complete reply.

    Raises:
        backend.BackendUnavailableError: The server cannot be reached, refused the request, or is rate limiting.
    """
    text_parts = []
    reasoning_parts = []
    call_parts = {}
    usage = None
    try:
        stream = await self._client.chat.completions.create(
            model=self.model,
            messages=self._wire_messages(conversation),
            tools=tool_definitions or openai.omit,
            stream=True,
            stream_options={
                'include_usage': True,
            },
        )
        async for chunk in stream:
            if chunk.usage is not None:
                usage = chunk.usage
            for choice in chunk.choices:
                thought = getattr(choice.delta, 'reasoning_content', None)
                if thought:
                    reasoning_parts.append(thought)
                piece = choice.delta.content
                if piece:
                    text_parts.append(piece)
                    on_text(piece)
                for delta_call in choice.delta.tool_calls or []:
                    self._add_call_delta(call_parts, delta_call)
    except openai.APIError as error:
        raise self._unavailable(error) from error
    tool_calls = []
    for index in sorted(call_parts):
        tool_calls.append(call_parts[index].to_tool_call())
    prompt_tokens = 0
    completion_tokens = 0
    cache_read_tokens = 0
    cost_usd = None
    if usage is not None:
        prompt_tokens = usage.prompt_tokens
        completion_tokens = usage.completion_tokens
        cache_read_tokens = self._cache_read_tokens(usage)
        cost_usd = self._cost_usd(usage, cache_read_tokens)
    return backend.ModelReply(
        text=''.join(text_parts),
        tool_calls=tool_calls,
        prompt_tokens=prompt_tokens,
        completion_tokens=completion_tokens,
        reasoning=''.join(reasoning_parts),
        cache_read_tokens=cache_read_tokens,
        cost_usd=cost_usd,
    )