Skip to content

OpenAILLM

BONUS Material: OpenAI LLM.

OpenAILLM

Bases: LLM

OpenAI LLM integration.

Source code in src/llm_agents_from_scratch/llms/openai/llm.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 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
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
class OpenAILLM(LLM):
    """OpenAI LLM integration."""

    def __init__(
        self,
        model: str,
        *,
        api_key: str | None = None,
        **kwargs: Any,
    ) -> None:
        """Create an OpenAILLM instance.

        Args:
            model (str): The name of the OpenAI LLM.
            api_key (str | None, optional): An OpenAI api key. Defaults to None.
                If None, will fallback to OpenAI's api key resolution, looking
                for an OPENAI_API_KEY env var.
            **kwargs (Any): Additional keyword arguments. Passed to the
                construction of an ~openai.AsyncOpenAI
        """
        check_extra_was_installed(extra="openai", packages="openai")
        from openai import AsyncOpenAI  # noqa: PLC0415

        # Avoid passing duplicate `api_key` if provided both explicitly and
        # in kwargs.
        kwargs.pop("api_key", None)
        self.model = model
        self.client = AsyncOpenAI(api_key=api_key, **kwargs)

    async def complete(self, prompt: str, **kwargs: Any) -> CompleteResult:
        """Implements complete LLM interaction mode."""
        response: "Response" = await self.client.responses.create(
            model=self.model,
            input=prompt,
            **kwargs,
        )
        return CompleteResult(response=str(response.output_text), prompt=prompt)

    async def structured_output(
        self,
        prompt: str,
        mdl: type[StructuredOutputType],
        **kwargs: Any,
    ) -> StructuredOutputType:
        """Implements structured output LLM interaction mode.

        Args:
            prompt (str): The prompt to elicit the structured output response.
            mdl (type[StructuredOutputType]): The ~pydantic.BaseModel to output.
            **kwargs (Any): Additional keyword arguments.

        Returns:
            StructuredOutputType: The structured output as the specified `mdl`
                type.
        """
        response: "ParsedResponse" = await self.client.responses.parse(
            model=self.model,
            input=prompt,
            text_format=mdl,
            **kwargs,
        )
        return response.output_parsed  # type: ignore[no-any-return]

    def _prepare_input_and_instructions_from_history(
        self,
        chat_history: Sequence[ChatMessage],
    ) -> tuple[list["ResponseInputItemParam"], str | None]:
        """Prepare response inputs and instructions.

        With OpenAI Responses API system messages can be supplied in
        instructions params. This helper extracts system messages from the chat
        history and prepares instructions by concatenating the content of such
        messages.

        Returns:
            tuple[list[openai.ResponseInputParam], str | None]: The input and
                instructions when invoking openai.client.responses.create()
        """
        context = []
        instruction_messages = []
        for cm in chat_history:
            if cm.role == "system":
                instruction_messages.append(cm.content)
            else:
                context.extend(chat_message_to_openai_response_input_param(cm))
        instructions = (
            "\n".join(instruction_messages) if instruction_messages else None
        )
        return context, instructions

    async def chat(
        self,
        input: str,
        chat_history: Sequence[ChatMessage] | None = None,
        tools: Sequence[Tool] | None = None,
        **kwargs: Any,
    ) -> tuple[ChatMessage, ChatMessage]:
        """Implements chat LLM interaction mode.

        Args:
            input (str): The user's current input.
            chat_history (list[ChatMessage] | None, optional): The chat
                history.
            tools (list[BaseTool] | None, optional): The tools available to the
                LLM.
            return_history (bool): Whether to return the update chat history.
                Defaults to False.
            **kwargs (Any): Additional keyword arguments.

        Returns:
            tuple[ChatMessage, ChatMessage]: A tuple of ChatMessage with the
                first message corresponding to the ChatMessage created from the
                supplied input string, and the second ChatMessage is the
                response from the LLM.
        """
        # prepare chat history and instructions
        chat_history = chat_history or []
        context, instructions = (
            self._prepare_input_and_instructions_from_history(
                chat_history,
            )
        )

        user_message = ChatMessage(role="user", content=input)
        context.extend(
            chat_message_to_openai_response_input_param(user_message),
        )

        # prepare tools
        openai_tools = (
            [tool_to_openai_tool(t) for t in tools] if tools else None
        )

        response = await self.client.responses.create(
            model=self.model,
            instructions=instructions,
            input=context,
            tools=openai_tools,
            **kwargs,
        )
        return user_message, openai_response_to_chat_message(response)

    async def continue_chat_with_tool_results(
        self,
        tool_call_results: Sequence[ToolCallResult],
        chat_history: Sequence[ChatMessage],
        tools: Sequence[Tool] | None = None,
        **kwargs: Any,
    ) -> tuple[list[ChatMessage], ChatMessage]:
        """Implements continue chat with tool results.

        Args:
            tool_call_results (Sequence[ToolCallResult]): The tool call results.
            chat_history (Sequence[ChatMessage]): The chat history.
            tools (Sequence[BaseTool]|None, optional): tools that the LLM
                can call.
            **kwargs (Any): Additional keyword arguments.

        Returns:
            tuple[list[ChatMessage], ChatMessage]: A tuple whose first element
                is a list of ChatMessage objects corresponding to the
                supplied ToolCallResult converted objects. The second element
                is the response ChatMessage from the LLM.
        """
        # prepare input from chat history
        chat_history = chat_history or []
        context_from_chat_history, instructions = (
            self._prepare_input_and_instructions_from_history(
                chat_history,
            )
        )

        # prepare input from tool call results
        tool_messages = [
            ChatMessage.from_tool_call_result(tc) for tc in tool_call_results
        ]
        context_from_tool_messages = []
        for tm in tool_messages:
            context_from_tool_messages.extend(
                chat_message_to_openai_response_input_param(tm),
            )

        openai_response_input_params = (
            context_from_chat_history + context_from_tool_messages
        )

        # prepare tools
        openai_tools = (
            [tool_to_openai_tool(t) for t in tools] if tools else None
        )

        # send response
        response = await self.client.responses.create(
            model=self.model,
            instructions=instructions,
            input=openai_response_input_params,
            tools=openai_tools,
            **kwargs,
        )

        return tool_messages, openai_response_to_chat_message(response)

__init__ 

__init__(model, *, api_key=None, **kwargs)

Create an OpenAILLM instance.

Parameters:

Name Type Description Default
model str

The name of the OpenAI LLM.

 required
api_key str | None

An OpenAI api key. Defaults to None. If None, will fallback to OpenAI's api key resolution, looking for an OPENAI_API_KEY env var.

 None
**kwargs Any

Additional keyword arguments. Passed to the construction of an ~openai.AsyncOpenAI

 {}
Source code in src/llm_agents_from_scratch/llms/openai/llm.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def __init__(
    self,
    model: str,
    *,
    api_key: str | None = None,
    **kwargs: Any,
) -> None:
    """Create an OpenAILLM instance.

    Args:
        model (str): The name of the OpenAI LLM.
        api_key (str | None, optional): An OpenAI api key. Defaults to None.
            If None, will fallback to OpenAI's api key resolution, looking
            for an OPENAI_API_KEY env var.
        **kwargs (Any): Additional keyword arguments. Passed to the
            construction of an ~openai.AsyncOpenAI
    """
    check_extra_was_installed(extra="openai", packages="openai")
    from openai import AsyncOpenAI  # noqa: PLC0415

    # Avoid passing duplicate `api_key` if provided both explicitly and
    # in kwargs.
    kwargs.pop("api_key", None)
    self.model = model
    self.client = AsyncOpenAI(api_key=api_key, **kwargs)

complete async 

complete(prompt, **kwargs)

Implements complete LLM interaction mode.

Source code in src/llm_agents_from_scratch/llms/openai/llm.py 
57
58
59
60
61
62
63
64
async def complete(self, prompt: str, **kwargs: Any) -> CompleteResult:
    """Implements complete LLM interaction mode."""
    response: "Response" = await self.client.responses.create(
        model=self.model,
        input=prompt,
        **kwargs,
    )
    return CompleteResult(response=str(response.output_text), prompt=prompt)

structured_output async 

structured_output(prompt, mdl, **kwargs)

Implements structured output LLM interaction mode.

Parameters:

Name Type Description Default
prompt str

The prompt to elicit the structured output response.

 required
mdl type[StructuredOutputType]

The ~pydantic.BaseModel to output.

 required
**kwargs Any

Additional keyword arguments.

 {}

Returns:

Name Type Description
StructuredOutputType StructuredOutputType

The structured output as the specified mdl type.
Source code in src/llm_agents_from_scratch/llms/openai/llm.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
async def structured_output(
    self,
    prompt: str,
    mdl: type[StructuredOutputType],
    **kwargs: Any,
) -> StructuredOutputType:
    """Implements structured output LLM interaction mode.

    Args:
        prompt (str): The prompt to elicit the structured output response.
        mdl (type[StructuredOutputType]): The ~pydantic.BaseModel to output.
        **kwargs (Any): Additional keyword arguments.

    Returns:
        StructuredOutputType: The structured output as the specified `mdl`
            type.
    """
    response: "ParsedResponse" = await self.client.responses.parse(
        model=self.model,
        input=prompt,
        text_format=mdl,
        **kwargs,
    )
    return response.output_parsed  # type: ignore[no-any-return]

chat async 

chat(input, chat_history=None, tools=None, **kwargs)

Implements chat LLM interaction mode.

Parameters:

Name Type Description Default
input str

The user's current input.

 required
chat_history list[ChatMessage] | None

The chat history.

 None
tools list[BaseTool] | None

The tools available to the LLM.

 None
return_history bool

Whether to return the update chat history. Defaults to False.

 required
**kwargs Any

Additional keyword arguments.

 {}

Returns:

Type Description
tuple[ChatMessage, ChatMessage]

tuple[ChatMessage, ChatMessage]: A tuple of ChatMessage with the first message corresponding to the ChatMessage created from the supplied input string, and the second ChatMessage is the response from the LLM.
Source code in src/llm_agents_from_scratch/llms/openai/llm.py 
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
162
163
164
165
166
167
168
async def chat(
    self,
    input: str,
    chat_history: Sequence[ChatMessage] | None = None,
    tools: Sequence[Tool] | None = None,
    **kwargs: Any,
) -> tuple[ChatMessage, ChatMessage]:
    """Implements chat LLM interaction mode.

    Args:
        input (str): The user's current input.
        chat_history (list[ChatMessage] | None, optional): The chat
            history.
        tools (list[BaseTool] | None, optional): The tools available to the
            LLM.
        return_history (bool): Whether to return the update chat history.
            Defaults to False.
        **kwargs (Any): Additional keyword arguments.

    Returns:
        tuple[ChatMessage, ChatMessage]: A tuple of ChatMessage with the
            first message corresponding to the ChatMessage created from the
            supplied input string, and the second ChatMessage is the
            response from the LLM.
    """
    # prepare chat history and instructions
    chat_history = chat_history or []
    context, instructions = (
        self._prepare_input_and_instructions_from_history(
            chat_history,
        )
    )

    user_message = ChatMessage(role="user", content=input)
    context.extend(
        chat_message_to_openai_response_input_param(user_message),
    )

    # prepare tools
    openai_tools = (
        [tool_to_openai_tool(t) for t in tools] if tools else None
    )

    response = await self.client.responses.create(
        model=self.model,
        instructions=instructions,
        input=context,
        tools=openai_tools,
        **kwargs,
    )
    return user_message, openai_response_to_chat_message(response)

continue_chat_with_tool_results async 

continue_chat_with_tool_results(
    tool_call_results, chat_history, tools=None, **kwargs
)

Implements continue chat with tool results.

Parameters:

Name Type Description Default
tool_call_results Sequence[ToolCallResult]

The tool call results.

 required
chat_history Sequence[ChatMessage]

The chat history.

 required
tools Sequence[BaseTool] | None

tools that the LLM can call.

 None
**kwargs Any

Additional keyword arguments.

 {}

Returns:

Type Description
tuple[list[ChatMessage], ChatMessage]

tuple[list[ChatMessage], ChatMessage]: A tuple whose first element is a list of ChatMessage objects corresponding to the supplied ToolCallResult converted objects. The second element is the response ChatMessage from the LLM.
Source code in src/llm_agents_from_scratch/llms/openai/llm.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
async def continue_chat_with_tool_results(
    self,
    tool_call_results: Sequence[ToolCallResult],
    chat_history: Sequence[ChatMessage],
    tools: Sequence[Tool] | None = None,
    **kwargs: Any,
) -> tuple[list[ChatMessage], ChatMessage]:
    """Implements continue chat with tool results.

    Args:
        tool_call_results (Sequence[ToolCallResult]): The tool call results.
        chat_history (Sequence[ChatMessage]): The chat history.
        tools (Sequence[BaseTool]|None, optional): tools that the LLM
            can call.
        **kwargs (Any): Additional keyword arguments.

    Returns:
        tuple[list[ChatMessage], ChatMessage]: A tuple whose first element
            is a list of ChatMessage objects corresponding to the
            supplied ToolCallResult converted objects. The second element
            is the response ChatMessage from the LLM.
    """
    # prepare input from chat history
    chat_history = chat_history or []
    context_from_chat_history, instructions = (
        self._prepare_input_and_instructions_from_history(
            chat_history,
        )
    )

    # prepare input from tool call results
    tool_messages = [
        ChatMessage.from_tool_call_result(tc) for tc in tool_call_results
    ]
    context_from_tool_messages = []
    for tm in tool_messages:
        context_from_tool_messages.extend(
            chat_message_to_openai_response_input_param(tm),
        )

    openai_response_input_params = (
        context_from_chat_history + context_from_tool_messages
    )

    # prepare tools
    openai_tools = (
        [tool_to_openai_tool(t) for t in tools] if tools else None
    )

    # send response
    response = await self.client.responses.create(
        model=self.model,
        instructions=instructions,
        input=openai_response_input_params,
        tools=openai_tools,
        **kwargs,
    )

    return tool_messages, openai_response_to_chat_message(response)