Skip to content

OllamaLLM

Ollama LLM integration.

OllamaLLM

Bases: LLM

Ollama LLM class.

Integration to ollama library for running open source models locally.

Source code in src/llm_agents_from_scratch/llms/ollama/llm.py 
 22
 23
 24
 25
 26
 27
 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
class OllamaLLM(LLM):
    """Ollama LLM class.

    Integration to `ollama` library for running open source models locally.
    """

    def __init__(
        self,
        model: str,
        host: str | None = None,
        *,
        think: bool = False,
        **kwargs: Any,
    ) -> None:
        """Create an OllamaLLM instance.

        Args:
            model (str): The name of the LLM model.
            host (str | None): Host of running Ollama service. Defaults to None.
            think (bool): Enable/disable thinking mode. Defaults to False.
            **kwargs (Any): Additional keyword arguments.
        """
        super().__init__(**kwargs)
        self.model = model
        self._client = AsyncClient(host=host)
        self.think = think

    async def complete(self, prompt: str, **kwargs: Any) -> CompleteResult:
        """Complete a prompt with an Ollama LLM.

        Args:
            prompt (str): The prompt to complete.
            **kwargs (Any): Additional keyword arguments.

        Returns:
            CompleteResult: The text completion result.
        """
        response = await self._client.generate(
            model=self.model,
            prompt=prompt,
            **kwargs,
        )
        return CompleteResult(
            response=response.response,
            prompt=prompt,
        )

    async def structured_output(
        self,
        prompt: str,
        mdl: type[StructuredOutputType],
        **kwargs: Any,
    ) -> StructuredOutputType:
        """Structured output interface implementation for Ollama LLM.

        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.
        """
        o_messages = [
            chat_message_to_ollama_message(
                ChatMessage(role="user", content=prompt),
            ),
        ]
        result = await self._client.chat(
            model=self.model,
            messages=o_messages,
            format=mdl.model_json_schema(),
            think=self.think,
            **kwargs,
        )
        return mdl.model_validate_json(result.message.content)

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

        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
        o_messages = (
            [chat_message_to_ollama_message(cm) for cm in chat_history]
            if chat_history
            else []
        )

        user_message = ChatMessage(role="user", content=input)
        o_messages.append(
            chat_message_to_ollama_message(
                user_message,
            ),
        )

        # prepare tools
        o_tools = [tool_to_ollama_tool(t) for t in tools] if tools else None

        result = await self._client.chat(
            model=self.model,
            messages=o_messages,
            tools=o_tools,
            think=self.think,
        )

        return user_message, ollama_message_to_chat_message(result.message)

    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 method.

        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.
        """
        # augment chat messages and convert to Ollama messages
        tool_messages = [
            ChatMessage.from_tool_call_result(tc) for tc in tool_call_results
        ]
        o_messages = [
            chat_message_to_ollama_message(cm) for cm in chat_history
        ] + [chat_message_to_ollama_message(tm) for tm in tool_messages]

        # prepare tools
        o_tools = [tool_to_ollama_tool(t) for t in tools] if tools else None

        # send chat request
        o_result = await self._client.chat(
            model=self.model,
            messages=o_messages,
            tools=o_tools,
            think=self.think,
        )

        return tool_messages, ollama_message_to_chat_message(o_result.message)

__init__ 

__init__(model, host=None, *, think=False, **kwargs)

Create an OllamaLLM instance.

Parameters:

Name Type Description Default
model str

The name of the LLM model.

 required
host str | None

Host of running Ollama service. Defaults to None.

 None
think bool

Enable/disable thinking mode. Defaults to False.

 False
**kwargs Any

Additional keyword arguments.

 {}
Source code in src/llm_agents_from_scratch/llms/ollama/llm.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def __init__(
    self,
    model: str,
    host: str | None = None,
    *,
    think: bool = False,
    **kwargs: Any,
) -> None:
    """Create an OllamaLLM instance.

    Args:
        model (str): The name of the LLM model.
        host (str | None): Host of running Ollama service. Defaults to None.
        think (bool): Enable/disable thinking mode. Defaults to False.
        **kwargs (Any): Additional keyword arguments.
    """
    super().__init__(**kwargs)
    self.model = model
    self._client = AsyncClient(host=host)
    self.think = think

complete async 

complete(prompt, **kwargs)

Complete a prompt with an Ollama LLM.

Parameters:

Name Type Description Default
prompt str

The prompt to complete.

 required
**kwargs Any

Additional keyword arguments.

 {}

Returns:

Name Type Description
CompleteResult CompleteResult

The text completion result.
Source code in src/llm_agents_from_scratch/llms/ollama/llm.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
async def complete(self, prompt: str, **kwargs: Any) -> CompleteResult:
    """Complete a prompt with an Ollama LLM.

    Args:
        prompt (str): The prompt to complete.
        **kwargs (Any): Additional keyword arguments.

    Returns:
        CompleteResult: The text completion result.
    """
    response = await self._client.generate(
        model=self.model,
        prompt=prompt,
        **kwargs,
    )
    return CompleteResult(
        response=response.response,
        prompt=prompt,
    )

structured_output async 

structured_output(prompt, mdl, **kwargs)

Structured output interface implementation for Ollama LLM.

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/ollama/llm.py 
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
async def structured_output(
    self,
    prompt: str,
    mdl: type[StructuredOutputType],
    **kwargs: Any,
) -> StructuredOutputType:
    """Structured output interface implementation for Ollama LLM.

    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.
    """
    o_messages = [
        chat_message_to_ollama_message(
            ChatMessage(role="user", content=prompt),
        ),
    ]
    result = await self._client.chat(
        model=self.model,
        messages=o_messages,
        format=mdl.model_json_schema(),
        think=self.think,
        **kwargs,
    )
    return mdl.model_validate_json(result.message.content)

chat async 

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

Chat with an Ollama LLM.

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/ollama/llm.py 
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
async def chat(
    self,
    input: str,
    chat_history: Sequence[ChatMessage] | None = None,
    tools: Sequence[Tool] | None = None,
    **kwargs: Any,
) -> tuple[ChatMessage, ChatMessage]:
    """Chat with an Ollama LLM.

    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
    o_messages = (
        [chat_message_to_ollama_message(cm) for cm in chat_history]
        if chat_history
        else []
    )

    user_message = ChatMessage(role="user", content=input)
    o_messages.append(
        chat_message_to_ollama_message(
            user_message,
        ),
    )

    # prepare tools
    o_tools = [tool_to_ollama_tool(t) for t in tools] if tools else None

    result = await self._client.chat(
        model=self.model,
        messages=o_messages,
        tools=o_tools,
        think=self.think,
    )

    return user_message, ollama_message_to_chat_message(result.message)

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 method.

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/ollama/llm.py 
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
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 method.

    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.
    """
    # augment chat messages and convert to Ollama messages
    tool_messages = [
        ChatMessage.from_tool_call_result(tc) for tc in tool_call_results
    ]
    o_messages = [
        chat_message_to_ollama_message(cm) for cm in chat_history
    ] + [chat_message_to_ollama_message(tm) for tm in tool_messages]

    # prepare tools
    o_tools = [tool_to_ollama_tool(t) for t in tools] if tools else None

    # send chat request
    o_result = await self._client.chat(
        model=self.model,
        messages=o_messages,
        tools=o_tools,
        think=self.think,
    )

    return tool_messages, ollama_message_to_chat_message(o_result.message)