Async-open-ai-call-app-w-gradio

async_gradio_app.py

from typing import AsyncIterator

import gradio as gr
from openai import AsyncOpenAI

client = AsyncOpenAI()


async def fetch_gpt_response(
    prompt: str, model: str = "gpt-3.5-turbo"
) -> AsyncIterator[str]:
    """
    GPTからのレスポンスを非同期で取得し、内容を連結して返す。
    ex) こん→こんにちは→こんにちは、元気ですか？
    """
    stream = await client.chat.completions.create(
        model=model,
        messages=[
            {"role": "user", "content": prompt},
        ],
        stream=True,
    )

    all_content = ""
    async for chunk in stream:
        chunk_content = chunk.choices[0].delta.content
        if chunk_content is not None:
            all_content += chunk_content
            yield all_content


async def merge_async_iterators(*iterables: AsyncIterator) -> AsyncIterator[tuple[str]]:
    """
    複数のAyncIteratorを受け取り、それぞれの最新の値をタプルにまとめて返す。
    いずれかのAsyncIteratorが終了した場合は、その値を最後の値として返す。
    すべてのAsyncIteratorが終了した場合は、処理を終了する。
    """
    iterators = [iterable.__aiter__() for iterable in iterables]
    last_values = [None] * len(iterators)
    done_flags = [False] * len(iterators)

    while not all(done_flags):
        results = []
        for idx, it in enumerate(iterators):
            if done_flags[idx]:
                results.append(last_values[idx])
                continue
            try:
                next_item = await it.__anext__()
                last_values[idx] = next_item
                results.append(next_item)
            except StopAsyncIteration:
                done_flags[idx] = True
                results.append(last_values[idx])
        yield tuple(results)


def build_whole_prompt(user_prompt: str, detail_prompt: str) -> str:
    return f"""
    課題の解像度を上げるために、対象に「{user_prompt}」ついて、アクション:{detail_prompt}を行ってください。
    会話形式や補足情報は不要です。
    各項目には、対象「{user_prompt}」に直接関連する情報を、簡潔に記述してください。
    具体性を持たせるため、実際のデータや事実に基づいた内容を想定しています。
    下記に続くように回答してください。

    会話形式や補足情報は不要です。アクションだけ行ってください。

    ---
    課題:「{user_prompt}」に「{detail_prompt}」を行う、

    """


async def main(user_prompt: str) -> AsyncIterator[str]:
    detail_prompts = [
        "深く考え直す。なぜ?を9回繰り返し問う。",
        "広く考え直す。異なるアプローチや視点を検討する。",
        "構造を見直す。原因や要因の関係性を把握する。",
        "時間軸について考える。時間的な変化を捉えて洞察につなげる。",
    ]

    gpt_responses = [
        fetch_gpt_response(build_whole_prompt(user_prompt, detail_prompt))
        for detail_prompt in detail_prompts
    ]
    async for merged_responses in merge_async_iterators(*gpt_responses):
        yield merged_responses


if __name__ == "__main__":
    with gr.Blocks(theme=gr.themes.Soft()) as demo:
        greet_btn = gr.Textbox(
            label="課題", placeholder="解像度を上げたい課題を入力してください。"
        )
        btn = gr.Button("Start")

        with gr.Row():
            with gr.Column():
                t1 = gr.Textbox(label="深さ")
            with gr.Column():
                t2 = gr.Textbox(label="広さ")

        with gr.Row():
            with gr.Column():
                t3 = gr.Textbox(label="構造")
            with gr.Column():
                t4 = gr.Textbox(label="時間")

        outputs = [t1, t2, t3, t4]

        btn.click(
            fn=main,
            inputs=greet_btn,
            outputs=outputs,
        )
    demo.launch()