A simple set of modules, wrappers and utils that are essential for LLM-based assistants and agents using the OpenAI API and function tools. It is useful for:
Yet to come:
pip install agentools
A high-level interface to use ChatGPT or other LLM-based assistants! The default implementation of ChatGPT has:
from agentools import *
# empty chat history and default model (gpt-3.5)
model = ChatGPT()
You can then simply call the model as if it was a function, with a prompt:
await model("Hey!")
'Hello! How can I assist you today?'
As you can see, the model is async and it simply returns the resonse as a string.
Both your prompt and the response are stored in the history, so you can keep calling the model with new prompts and it will remember the conversation so far.
await model("Can you repeat my last message please?")
'Of course! You said, "Hey!"'
model.messages.history
[{'role': 'user', 'content': 'Hey!'},
{'content': 'Hello! How can I assist you today?', 'role': 'assistant'},
{'role': 'user', 'content': 'Can you repeat my last message please?'},
{'content': 'Of course! You said, "Hey!"', 'role': 'assistant'}]
MessageHistory
Notice that our model has no system prompt in the beginning. ChatGPT
's constructor by default creates an empty chat history, but you can explicitly create a MessageHistory
object and pass it to the constructor:
translate = ChatGPT(
messages=SimpleHistory.system("Translate the user message to English")
)
# SimpleHistory.system(s) is just shorthand for SimpleHistory([msg(system=s)])
print(await translate("Ich liebe Katzen!"))
print(await translate("고양이랑 강아지 둘다 좋아!"))
I love cats!
I like both cats and dogs!
translate.messages.history
[{'role': 'system', 'content': 'Translate the user message to English'},
{'role': 'user', 'content': 'Ich liebe Katzen!'},
{'content': 'I love cats!', 'role': 'assistant'},
{'role': 'user', 'content': '고양이랑 강아지 둘다 좋아!'},
{'content': 'I like both cats and dogs!', 'role': 'assistant'}]
Notice that here, we're wasting tokens by remembering the chat history, since it's not really a conversation. There's a simple GPT
class, which simply resets the message history after each prompt:
translate = GPT(messages=SimpleHistory.system("Translate the user message to English"))
await translate("Ich liebe Katzen!")
await translate("고양이랑 강아지 둘다 좋아!")
translate.messages.history
[{'role': 'system', 'content': 'Translate the user message to English'}]
You can set the default model in the constructor, or override it for each prompt:
# default model is now gpt-4 💸
model = ChatGPT(model="gpt-4")
# but you can override it for each prompt anyways
await model("Heyo!", model="mocked")
'Hello, world!'
As you see, our wrapper provides a simple mocked "model", which will simply return "Hello, world!"
for any prompt, with some simulated latency. This will also work with streaming responses, and in either cases, you won't be able to tell the difference between the real API and the mocked one.
There are more mocked models for your convinience:
mocked
: always returns "Hello, world!"
mocked:TEST123
: returns the string after the colon, e.g. "TEST123"
echo
: returns the user prompt itselfLet's print all events to the console to take a peek at the event-based generator:
await model("Heya!", model="echo", event_logger=print)
[ResponseStartEvent]: prompt=Heya!, tools=None, model=echo, max_function_calls=100, openai_kwargs={}
[CompletionStartEvent]: call_index=0
[CompletionEvent]: completion=ChatCompletion(id='mock', choices=[Choice(finish_reason='stop', index=0, logprobs=None, message=ChatCompletionMessage(content='Heya!', role='assistant', function_call=None, tool_calls=None))], created=1721161834, model='mock', object='chat.completion', service_tier=None, system_fingerprint=None, usage=None), call_index=0
[FullMessageEvent]: message=ChatCompletionMessage(content='Heya!', role='assistant', function_call=None, tool_calls=None), choice_index=0
[TextMessageEvent]: content=Heya!
[ResponseEndEvent]: content=Heya!
'Heya!'
Wow, quite a lot going on for a simple prompt! While it might seem like too many events, this offers a lot of flexibility and customizability.
You can easily handle only the events you are interested in, useful when e.g:
For instance, the GPT
class from above is as simple as:
async for event in self.response_events(prompt, **openai_kwargs):
match event:
case self.ResponseEndEvent():
await self.messages.reset()
return event.content
This generator-based architecture is a good balance between flexibility and simplicity!
While we won't go deeper into the low-level API in this quickstart, you can look at the advanced.ipynb
notebook for more details.
@function_tool
You can turn any function into a tool usable by the model by decorating it with @function_tool
:
@function_tool
def print_to_console(text: str) -> str:
"""
Print text to console
Args:
text: text to print
"""
print(text)
return "success" # the model will see the return value
# normal call
print_to_console("Hello from python!")
Hello from python!
'success'
You can use the tool from python as you normally would, and the model will also be able to use it simply by passing it to the tools
parameter during init (as default) or prompting it (as a one-off).
model = ChatGPT(tools=print_to_console)
await model("Say 'hello from GPT' to console!")
hello from GPT
'The message "hello from GPT" has been successfully printed to the console.'
To make the function a @function_tool
, you must do the following:
str
, int
, float
, bool
, list
, dict
, None
, etc).str
or something that can be converted to a str
.'''docstring'''
, including each parameter (most formats supported, e.g. Google-style, NumPy-style, sphinx-style, etc, see this overview)Showing off some more goodies:
await
them.@fail_with_message(err)
is a decorator that will catch any exceptions thrown by the function and instead return the error message. This is useful for when you want to handle errors in a more graceful way than just crashing the model. It also takes an optional logger, which by default takes the print
function, but any callable that takes a string will work, such as logger.error
from the logging
module.@function_tool
decorator will throw an assertion error if you forget to provide the description for any of the function or their parameters. If you really don't want to provide descriptions for some (or all), maybe because it's so self-explanatory or you need to save tokens, then you can explicitly turn off the docstring parsing by passing @function_tool(check_description=False)
. This is not recommended, but it's there if you need it.Note that by returning descriptive error strings, the model can read the error message and retry, increasing the robustness!
import asyncio
import logging
@function_tool(name="Fibonacci", require_doc=False)
@fail_with_message("Error", logger=logging.error)
async def fib(n: int):
if n < 0:
raise ValueError("n must be >= 0")
if n < 2:
return n
await asyncio.sleep(0.1)
return sum(await asyncio.gather(fib(n - 1), fib(n - 2)))
await fib(-10)
ERROR:root:Tool call fib(-10) failed: n must be >= 0
'Error: n must be >= 0'
class Toolkit
Toolkits are a collection of related function tools, esp. useful when they share a state. Also good for keeping the state bound to a single instance of the toolkit, rather than a global state.
To create a toolkit, simply subclass Toolkit
and decorate its methods with @function_tool
.
class Notepad(Toolkit):
def __init__(self):
super().__init__()
self.content = "<Fill me in>"
@function_tool
def write(self, text: str):
"""
Write text to the notepad
Args:
text: The text to write
"""
self.content = text
@function_tool(require_doc=False)
def read(self):
return self.content
notes = Notepad()
notes.write("Shhh... here's a secret: 42")
notes.read()
"Shhh... here's a secret: 42"
As before, simply pass the toolkit to the model. To use multiple tools and toolkits, simply put them in a list:
model = ChatGPT(
tools=[notes, print_to_console, fib],
)
await model("What's on my notepad?")
'On your notepad, it says: "Shhh... here\'s a secret: 42"'
await model(
"Can you calculate the 8th fibonacci number, add it to the number in my notes, and write it? also print it to console as well.",
event_logger=lambda x: print(x) if x.startswith("[Tool") else None,
parallel_tool_calls=False,
)
[ToolCallsEvent]: tool_calls=[ChatCompletionMessageToolCall(id='call_wxaisBbFMYRa0XNcTnP9MH1b', function=Function(arguments='{"n":8}', name='Fibonacci'), type='function')]
[ToolResultEvent]: result=21, tool_call=ChatCompletionMessageToolCall(id='call_wxaisBbFMYRa0XNcTnP9MH1b', function=Function(arguments='{"n":8}', name='Fibonacci'), type='function'), index=0
[ToolCallsEvent]: tool_calls=[ChatCompletionMessageToolCall(id='call_gt5ZnA5v2VJL5R2gyPeHRN0a', function=Function(arguments='{"text":"The sum of the 8th Fibonacci number (21) and the number on your notepad (42) is 63"}', name='write'), type='function')]
[ToolResultEvent]: result=None, tool_call=ChatCompletionMessageToolCall(id='call_gt5ZnA5v2VJL5R2gyPeHRN0a', function=Function(arguments='{"text":"The sum of the 8th Fibonacci number (21) and the number on your notepad (42) is 63"}', name='write'), type='function'), index=0
[ToolCallsEvent]: tool_calls=[ChatCompletionMessageToolCall(id='call_ErYx6g7gpVTnLsqg59oxHI9C', function=Function(arguments='{"text":"The sum of the 8th Fibonacci number (21) and the number on your notepad (42) is 63"}', name='print_to_console'), type='function')]
The sum of the 8th Fibonacci number (21) and the number on your notepad (42) is 63
[ToolResultEvent]: result=success, tool_call=ChatCompletionMessageToolCall(id='call_ErYx6g7gpVTnLsqg59oxHI9C', function=Function(arguments='{"text":"The sum of the 8th Fibonacci number (21) and the number on your notepad (42) is 63"}', name='print_to_console'), type='function'), index=0
'I have written on your notepad. The sum of the 8th Fibonacci number (21) and the number on your notepad (42) is 63. I have also printed it to the console.'
notes.read()
'The sum of the 8th Fibonacci number (21) and the number on your notepad (42) is 63'
Notice how since our write
function doesn't return anything, it defaults to None
and our model gets confused! So don't forget to return an encouraging success message to make our model happy :)
When using streaming, and you're using function tools with a long input, you might want to preview the tool's output before it's fully processed. With the help of the json_autocomplete
package, the JSON argument generated by the model can be parsed before it's fully complete, and the preview can be shown to the user.
@function_tool(require_doc=False)
async def create_slogan(title: str, content: str):
print(f"\n\n[Final Slogan] {title}: {content}")
return "Slogan created and shown to user! Simply tell the user that it was created."
@create_slogan.preview
async def preview(title: str = "", content: str = ""):
assert isinstance(title, str) and isinstance(content, str)
print(f"[Preview] {title}: {content}", flush=True)
model = ChatGPT(tools=create_slogan)
await model(
"Create a 1-sentence slogan about how ducks can help with debugging.", stream=True
)
[Preview] :
[Preview] D:
[Preview] Ducks:
[Preview] Ducks and:
[Preview] Ducks and Debug:
[Preview] Ducks and Debugging:
[Preview] Ducks and Debugging:
[Preview] Ducks and Debugging: Qu
[Preview] Ducks and Debugging: Quack
[Preview] Ducks and Debugging: Quack your
[Preview] Ducks and Debugging: Quack your code
[Preview] Ducks and Debugging: Quack your code bugs
[Preview] Ducks and Debugging: Quack your code bugs away
[Preview] Ducks and Debugging: Quack your code bugs away with
[Preview] Ducks and Debugging: Quack your code bugs away with the
[Preview] Ducks and Debugging: Quack your code bugs away with the help
[Preview] Ducks and Debugging: Quack your code bugs away with the help of
[Preview] Ducks and Debugging: Quack your code bugs away with the help of a
[Preview] Ducks and Debugging: Quack your code bugs away with the help of a debugging
[Preview] Ducks and Debugging: Quack your code bugs away with the help of a debugging duck
[Preview] Ducks and Debugging: Quack your code bugs away with the help of a debugging duck by
[Preview] Ducks and Debugging: Quack your code bugs away with the help of a debugging duck by your
[Preview] Ducks and Debugging: Quack your code bugs away with the help of a debugging duck by your side
[Preview] Ducks and Debugging: Quack your code bugs away with the help of a debugging duck by your side.
[Preview] Ducks and Debugging: Quack your code bugs away with the help of a debugging duck by your side.
[Final Slogan] Ducks and Debugging: Quack your code bugs away with the help of a debugging duck by your side.
'I have created a slogan about how ducks can help with debugging!'
If you need a more coherent logic shared between the @preview
and the final @function_tool
, e.g. do something at the start of the function call, share some data between previews, etc... It gets messy very fast!
Instead, you can use the @streaming_function_tool()
decorator, which receives a single arg_stream
parameter, which is an async generator that yields the partial arguments, as streamed from the model. Therefore, you simply need to iterate through it, and perform the actual function call at the end of the iteration. The following is the equivalent of the previous example:
Note that currently, you must pass the parameter as a
schema
(either JSON Schema or Pydantic BaseModel).
from pydantic import BaseModel, Field
class Slogan(BaseModel):
"""A slogan for a product"""
title: str = Field(description="MUST BE EXACTLY 3 WORDS!")
content: str = Field(description="less than 10 words")
@streaming_function_tool(schema=Slogan)
async def create_slogan(arg_stream):
print("Starting slogan creation...")
async for args in arg_stream:
title, content = args.get("title", ""), args.get("content", "")
print(f'{args} -> "{title}", "{content}"', flush=True)
print(f"\n\n[Final Slogan] {title}: {content}")
return "Slogan created and shown to user! Simply tell the user that it was created."
model = ChatGPT(tools=create_slogan)
await model(
"Create a 1-sentence slogan about how ducks can help with debugging.", stream=True
)
Starting slogan creation...
{'': None} -> "", ""
{'title': None} -> "None", ""
{'title': ''} -> "", ""
{'title': 'Debug'} -> "Debug", ""
{'title': 'Debugging'} -> "Debugging", ""
{'title': 'Debugging Ducks'} -> "Debugging Ducks", ""
{'title': 'Debugging Ducks', '': None} -> "Debugging Ducks", ""
{'title': 'Debugging Ducks', 'content': None} -> "Debugging Ducks", "None"
{'title': 'Debugging Ducks', 'content': ''} -> "Debugging Ducks", ""
{'title': 'Debugging Ducks', 'content': 'Qu'} -> "Debugging Ducks", "Qu"
{'title': 'Debugging Ducks', 'content': 'Quack'} -> "Debugging Ducks", "Quack"
{'title': 'Debugging Ducks', 'content': 'Quack through'} -> "Debugging Ducks", "Quack through"
{'title': 'Debugging Ducks', 'content': 'Quack through errors'} -> "Debugging Ducks", "Quack through errors"
{'title': 'Debugging Ducks', 'content': 'Quack through errors effortlessly'} -> "Debugging Ducks", "Quack through errors effortlessly"
{'title': 'Debugging Ducks', 'content': 'Quack through errors effortlessly.'} -> "Debugging Ducks", "Quack through errors effortlessly."
{'title': 'Debugging Ducks', 'content': 'Quack through errors effortlessly.'} -> "Debugging Ducks", "Quack through errors effortlessly."
{'title': 'Debugging Ducks', 'content': 'Quack through errors effortlessly.'} -> "Debugging Ducks", "Quack through errors effortlessly."
[Final Slogan] Debugging Ducks: Quack through errors effortlessly.
'I have created a slogan: "Debugging Ducks - Quack through errors effortlessly."'
We can very easily define a Pydantic model that can be generated by the LLM model, with validation and retries:
from enum import StrEnum
from pydantic import BaseModel, Field
class Language(StrEnum):
EN = "en"
DE = "de"
KO = "ko"
class Song(BaseModel):
title: str
genres: list[str] = Field(description="AT LEAST 3 genres!")
duration: float
language: Language
has_lyrics: bool
# normal use
Song(title="Hello", genres=["pop"], duration=3.5, language=Language.EN, has_lyrics=True)
Song(title='Hello', genres=['pop'], duration=3.5, language=<Language.EN: 'en'>, has_lyrics=True)
Create a StructGPT
object with your pydantic model, and prompting it will always return a valid instance of the model, or raise an exception if it fails to generate a valid instance after the maximum number of retries. Your docstring and field descriptions will also be visible to the model, so make sure to write good descriptions!
generate_song = StructGPT(Song)
await generate_song("Come up with an all-time best K-hiphop song")
Song(title='Eternal Sunshine', genres=['Hip-hop', 'R&B', 'K-pop'], duration=240.0, language=<Language.KO: 'ko'>, has_lyrics=True)
Streaming can be enabled as usual by passing stream=True
when prompting, and handle the partial events as they come in. Check the Assistant
class for a list of events including the ones for streaming.
There are some other useful utilities in the utils
module, such as:
tokens
: for token countingtrackers
: for transparent token tracking and prompt/response logging