Large Language Models (LLMs)
The LLM class is the interface between Agentswarm and the various AI model providers.
The Abstract Concept
Agentswarm is model-agnostic. The LLM abstract base class defines a standard way to:
1. Generate text: Given a list of messages.
2. Handle Tools: define function definitions (schemas) and parse tool calls from the model's response.
By implementing this interface, you can add support for any model provider that supports function calling (or even emulate it).
agentswarm.llms.LLM
Source code in src/agentswarm/llms/llm.py
| class LLM:
async def generate(
self,
messages: List[Message],
functions: List[LLMFunction] = None,
feedback: Optional[FeedbackSystem] = None,
) -> LLMOutput:
pass
|
Implementing a Custom Provider
To add a new provider (e.g., Anthropic, OpenAI, local Llama), create a subclass of LLM and implement generate.
from agentswarm.datamodels import Message, LLMFunction
from agentswarm.llms import LLM, LLMOutput
class MyCustomLLM(LLM):
def __init__(self, api_key: str, model_name: str = "gpt-4"):
self.client = ... # Initialize your client
self.model = model_name
async def generate(self, messages: list[Message], functions: list[LLMFunction] = None, feedback: FeedbackSystem = None) -> LLMOutput:
# 1. Convert Agentswarm Messages to provider format
# 2. Convert LLMFunctions to provider tool schemas
# 3. Call the API (optionally with streaming via feedback)
# 4. Parse the response into LLMOutput (text + function_calls)
pass
Supported Providers
Agentswarm currently includes support for Gemini.
Gemini
The GeminiLLM implementation connects to Google's Vertex AI or Generative AI SDKs.
agentswarm.llms.GeminiLLM
Bases: LLM
Source code in src/agentswarm/llms/gemini.py
13
14
15
16
17
18
19
20
21
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
193
194
195
196 | class GeminiLLM(LLM):
def __init__(
self,
api_key: str = None,
model: str = "gemini-3-flash-preview",
client: Client = None,
):
if api_key is None and client is None:
raise ValueError("api_key or client must be provided")
self.client = client if client is not None else Client(api_key=api_key)
self.model = model
async def generate(
self,
messages: List[Message],
functions: List[LLMFunction] = None,
feedback: Optional[FeedbackSystem] = None,
) -> LLMOutput:
contents = []
sys_instruct = []
for message in messages:
if message.type != "system":
role = "model" if message.type == "assistant" else message.type
contents.append(
types.Content(role=role, parts=[types.Part(text=message.content)])
)
else:
sys_instruct.append(message.content)
if len(sys_instruct) == 0:
sys_instruct = None
function_declarations = []
if functions is not None:
for fn in functions:
function_declarations.append(
{
"name": fn.name,
"description": fn.description,
"parameters": fn.parameters,
}
)
tools = [types.Tool(function_declarations=function_declarations)]
else:
tools = None
config = types.GenerateContentConfig(
temperature=0,
tools=tools,
system_instruction=sys_instruct,
safety_settings=[
types.SafetySetting(
category="HARM_CATEGORY_HATE_SPEECH", threshold="OFF"
),
types.SafetySetting(
category="HARM_CATEGORY_DANGEROUS_CONTENT", threshold="OFF"
),
types.SafetySetting(
category="HARM_CATEGORY_SEXUALLY_EXPLICIT", threshold="OFF"
),
types.SafetySetting(
category="HARM_CATEGORY_HARASSMENT", threshold="OFF"
),
],
)
if feedback:
text_parts = []
output_function_calls = []
usage = None
async for chunk in await self.client.aio.models.generate_content_stream(
model=self.model, config=config, contents=contents
):
if (
chunk.candidates
and chunk.candidates[0].content
and chunk.candidates[0].content.parts
):
for part in chunk.candidates[0].content.parts:
if part.text:
text_parts.append(part.text)
feedback.push(Feedback(payload=part.text, source="llm"))
if part.function_call:
args = part.function_call.args
if args is not None and not isinstance(args, dict):
try:
args = dict(args)
except Exception:
pass
output_function_calls.append(
LLMFunctionExecution(
name=part.function_call.name, arguments=args
)
)
if chunk.usage_metadata:
usg = chunk.usage_metadata
usage = LLMUsage(
model=self.model,
prompt_token_count=(
usg.prompt_token_count
if usg.prompt_token_count is not None
else 0
),
thoughts_token_count=(
usg.thoughts_token_count
if usg.thoughts_token_count is not None
else 0
),
tool_use_prompt_token_count=(
usg.tool_use_prompt_token_count
if usg.tool_use_prompt_token_count is not None
else 0
),
candidates_token_count=(
usg.candidates_token_count
if usg.candidates_token_count is not None
else 0
),
total_token_count=(
usg.total_token_count
if usg.total_token_count is not None
else 0
),
)
return LLMOutput(
text="".join(text_parts),
function_calls=output_function_calls,
usage=usage,
)
response = await self.client.aio.models.generate_content(
model=self.model, config=config, contents=contents
)
usg = response.usage_metadata
usage = LLMUsage(
model=self.model,
prompt_token_count=(
usg.prompt_token_count if usg.prompt_token_count is not None else 0
),
thoughts_token_count=(
usg.thoughts_token_count if usg.thoughts_token_count is not None else 0
),
tool_use_prompt_token_count=(
usg.tool_use_prompt_token_count
if usg.tool_use_prompt_token_count is not None
else 0
),
candidates_token_count=(
usg.candidates_token_count
if usg.candidates_token_count is not None
else 0
),
total_token_count=(
usg.total_token_count if usg.total_token_count is not None else 0
),
)
output_function_calls = []
text_parts = []
if (
response.candidates
and response.candidates[0].content
and response.candidates[0].content.parts
):
for part in response.candidates[0].content.parts:
if part.text:
text_parts.append(part.text)
if part.function_call:
args = part.function_call.args
if args is not None and not isinstance(args, dict):
try:
args = dict(args)
except Exception:
pass
output_function_calls.append(
LLMFunctionExecution(
name=part.function_call.name, arguments=args
)
)
return LLMOutput(
text="".join(text_parts), function_calls=output_function_calls, usage=usage
)
|
Reliable LLM (Wrapper)
The ReliableLLM is a wrapper that adds retry and timeout logic to any other LLM.
Learn more about Reliable LLM