async utils

async_utils.py

from marvin import ai_fn
from pydantic import BaseModel
import asyncio


class Fruit(BaseModel):
    name: str


@ai_fn
async def get_fruits(n) -> list[Fruit]:
    """Generates n random fruits"""


async def async_map(items, fn):
    tasks = []
    for i in items:
        tasks.append(fn(i))
    results = await asyncio.gather(*tasks)
    return results


async def async_filter(items, fn):
    results = await async_map(items, fn)
    print(results)
    return [i for i, r in zip(items, results) if r]


@ai_fn
async def get_color(fruit: Fruit) -> str:
    """Returns the color of the fruit"""


x = type(Fruit(name="apple"))
print(x)


@ai_fn
async def is_red(fruit: x) -> bool:
    """Returns True if fruit is red"""


async def main():
    fruits = await get_fruits(10)
    print(fruits)

    colors = await async_map(fruits, get_color)
    print(colors)

    red_fruits = await async_filter(fruits, is_red)
    print(red_fruits)


asyncio.run(main())

async_utils_ai.py

from marvin import ai_fn
from pydantic import BaseModel
import asyncio


class Fruit(BaseModel):
    name: str


@ai_fn
async def get_fruits(n) -> list[Fruit]:
    """Generates n random fruits"""


async def async_map(items, fn):
    tasks = []
    for i in items:
        tasks.append(fn(i))
    results = await asyncio.gather(*tasks)
    return results


async def async_filter(items, fn):
    results = await async_map(items, fn)
    print(results)
    return [i for i, r in zip(items, results) if r]


async def ai_map(items, q, OT):
    if not len(items):
        return []
    T = type(items[0])

    @ai_fn
    async def fn(item: T) -> OT:
        f"Answer the question {q}"

    return await async_map(items, fn)


async def ai_filter(items, criteria):
    if not len(items):
        return []
    T = type(items[0])

    @ai_fn
    async def fn(item: T) -> bool:
        f"Returns True if {criteria}"

    return await async_filter(items, fn)


@ai_fn
async def get_color(fruit: Fruit) -> str:
    """Returns the color of the fruit"""


x = type(Fruit(name="apple"))
print(x)


@ai_fn
async def is_red(fruit: x) -> bool:
    """Returns True if fruit is red"""


async def main():
    fruits = await get_fruits(10)
    print(fruits)

    # colors = await async_map(fruits, get_color)
    # print(colors)
    colors = await ai_map(fruits, "What is the color of the fruit?", str)

    # red_fruits = await async_filter(fruits, is_red)
    # print(red_fruits)

    print(await ai_filter(fruits, "The fruit is red."))


asyncio.run(main())