Skip to content
/ turbo-ai Public

Idiomatic way to build chatgpt apps using async generators in python

License

MIT license
18 stars 1 fork Branches Tags Activity
Star
Notifications You must be signed in to change notification settings

julep-ai/turbo-ai

BranchesTags

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

233 Commits
.autodoc/docs
.autodoc/docs
 
 
.github/workflows
.github/workflows
 
 
tests
tests
 
 
turbo_chat
turbo_chat
 
 
.gitignore
.gitignore
 
 
FAQ.md
FAQ.md
 
 
LICENSE
LICENSE
 
 
README.md
README.md
 
 
TODOS.md
TODOS.md
 
 
autodoc.config.json
autodoc.config.json
 
 
poetry.lock
poetry.lock
 
 
pyproject.toml
pyproject.toml
 
 
pytype.toml
pytype.toml
 
 

Repository files navigation

turbo-chat

Idiomatic way to build chatgpt apps using async generators in python

turbo

About

The ChatGPT API uses a new input format called chatml. In openai's python client, the format is used something like this:

messages = [
    {"role": "system", "content": "Greet the user!"},
    {"role": "user", "content": "Hello world!"},
]

The idea here is to incrementally build the messages using an async generator and then use that to generate completions. Async generators are incredibly versatile and simple abstraction for doing this kind of stuff. They can also be composed together very easily.

# Equivalent turbo-chat generator
async def example():
    yield System(content="Greet the user!")
    yield User(content="Hello World!")

    # To run generation, just yield Generate(),
    # the lib will take care of correctly running the app, and
    # return the value back here.
    output = yield Generate()
    print(output.content)

See more detailed example below.

Installation

pip install turbo-chat

Example

from typing import AsyncGenerator, Union

from turbo_chat import (
    turbo,
    System,
    User,
    Assistant,
    GetInput,
    Generate,
    run,
)

# Get user
async def get_user(id):
    return {"zodiac": "pisces"}

# Set user zodiac mixin
# Notice that no `@turbo()` decorator used here
async def set_user_zodiac(user_id: int):

    user_data: dict = await get_user(user_id)
    zodiac: str = user_data["zodiac"]

    yield User(content=f"My zodiac sign is {zodiac}")


# Horoscope app
@turbo(temperature=0.0)
async def horoscope(user_id: int):

    yield System(content="You are a fortune teller")

    # Yield from mixin
    async for output in set_user_zodiac(user_id):
        yield output

    # Prompt runner to ask for user input
    input = yield GetInput(message="What do you want to know?")

    # Yield the input
    yield User(content=input)

    # Generate (overriding the temperature)
    value = yield Generate(temperature=0.9)

# Let's run this
app: AsyncGenerator[Union[Assistant, GetInput], str] = horoscope({"user_id": 1})

_input = None
while not (result := await (app.run(_input)).done:
    if result.needs_input:
        # Prompt user with the input message
        _input = input(result.content)
        continue

    print(result.content)

# Output
# >>> What do you want to know? Tell me my fortune
# >>> As an AI language model, I cannot predict the future or provide supernatural fortune-telling. However, I can offer guidance and advice based on your current situation and past experiences. Is there anything specific you would like me to help you with?
#

Custom memory

You can also customize how the messages are persisted in-between the executions.

from turbo_chat import turbo, BaseMemory

class RedisMemory(BaseMemory):
    """Implement BaseMemory methods here"""

    async def setup(self, **kwargs) -> None:
        ...

    async def append(self, item) -> None:
        ...

    async def clear(self) -> None:
        ...


# Now use the memory in a turbo_chat app
@turbo(memory_class=RedisMemory)
async def app():
    ...

Get access to memory object directly (just declare an additional param)

@turbo()
async def app(some_param: Any, memory: BaseMemory):

    messages = await memory.get()
    ...

Generate a response to use internally but don't yield downstream

@turbo()
async def example():
    yield System(content="You are a good guy named John")
    yield User(content="What is your name?")
    result = yield Generate(forward=False)

    yield User(content="How are you doing?")
    result = yield Generate()

b = example()
results = [output async for output in b]

assert len(results) == 1

Add a simple in-memory cache

You can also subclass the BaseCache class to create a custom cache.

cache = SimpleCache()

@turbo(cache=cache)
async def example():
    yield System(content="You are a good guy named John")
    yield User(content="What is your name?")
    result = yield Generate()

b = example()
results = [output async for output in b]

assert len(cache.cache) == 1

Latest Changes

About

Idiomatic way to build chatgpt apps using async generators in python

Topics

python asyncio gpt-3 gpt-4 llm chatgpt gpt-35-turbo gpt-3-5-turbo

Resources

Readme

License

MIT license
Activity
Custom properties

Stars

18 stars

Watchers

4 watching

Forks

1 fork
Report repository

Releases 21

0.3.11 Latest
May 2, 2023
+ 20 releases

Contributors 2

  •  
  •  

Languages