How do kick off two tasks concurrently and based on the first task result cancel the second task?

[vinodkumars]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the Asyncer documentation, with the integrated search.
• I already searched in Google "How to X in Asyncer" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to Asyncer but to AnyIO.
• I already checked if it is not related to Asyncer but to Trio.

Commit to Help

• I commit to help with one of those options 👆

Example Code

import asyncio

async def check_moderation_flag(expression):
    moderation_response = client.moderations.create(input=expression)
    flagged = moderation_response.results[0].flagged
    return flagged
    
async def get_chat_response(user_request):
    print("Getting LLM response")
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_request},
    ]
    response = client.chat.completions.create(
        model=GPT_MODEL, messages=messages, temperature=0.5
    )
    print("Got LLM response")
    return response.choices[0].message.content


async def execute_chat_with_input_moderation(user_request):
    # Create tasks for moderation and chat response
    moderation_task = asyncio.create_task(check_moderation_flag(user_request))
    chat_task = asyncio.create_task(get_chat_response(user_request))

    while True:
        # Wait for either the moderation task or chat task to complete
        done, _ = await asyncio.wait(
            [moderation_task, chat_task], return_when=asyncio.FIRST_COMPLETED
        )

        # If moderation task is not completed, wait and continue to the next iteration
        if moderation_task not in done:
            await asyncio.sleep(0.1)
            continue

        # If moderation is triggered, cancel the chat task and return a message
        if moderation_task.result() == True:
            chat_task.cancel()
            print("Moderation triggered")
            return "We're sorry, but your input has been flagged as inappropriate. Please rephrase your input and try again."

        # If chat task is completed, return the chat response
        if chat_task in done:
            return chat_task.result()

        # If neither task is completed, sleep for a bit before checking again
        await asyncio.sleep(0.1)

Description

I am investigating how to add guardrails to a LLM app. I came across this very useful OpenAI cookbook. In the cookbook, they use asyncio. They kick off two tasks concurrently: one for guardrail check and one for actual LLM processing. Based on the guardrail task result, they either use or cancel the other task.

I can't figure out how to do them using asyncer? My understading of soonify() in a taskgroup is that both tasks will be done when the code block is exited.

What is the recommended way to implement the guardrails using asyncer?

Operating System

macOS

Operating System Details

No response

asyncer Version

0.0.8

Python Version

3.11.8

Additional Context

No response

[YuriiMotov]

Add tasks to TaskGroup

    async with asyncer.create_task_group() as task_group:
        result1 = task_group.soonify(do_work)(name="task 1", tm=1)
        _ = task_group.soonify(do_work)(name="task 2", tm=2)

and use

        while not result1.ready:  # Untill task 1 is done
            await anyio.sleep(0)  # Give event loop time to execute tasks

to execute them until first task is done.
Then cancel tasks

        task_group.cancel_scope.cancel()

Full code example:

import anyio
import asyncer


async def do_work(name: str, tm: int) -> str:
    await anyio.sleep(tm)
    print(f"Hello {name}")
    return f"Hello {name}"


async def main():
    async with asyncer.create_task_group() as task_group:
        result1 = task_group.soonify(do_work)(name="task 1", tm=1)
        _ = task_group.soonify(do_work)(name="task 2", tm=2)

        while not result1.ready:  # Untill task 1 is done
            await anyio.sleep(0)  # Give event loop time to execute tasks

        task_group.cancel_scope.cancel()


anyio.run(main)

Output:

Hello task 1