Create OpenAI Chat Server via Hypha

openai_server.py

import asyncio
import random
from hypha_rpc.utils.serve import create_openai_chat_server


async def text_generator(request: dict):
    max_tokens = request.get("max_tokens", 50)
    words = [
        "hello",
        "world",
        "foo",
        "bar",
        "chatbot",
        "test",
        "api",
        "response",
        "markdown",
        "collapsible",
        "details",
        "block",
        "section",
        "content",
    ]

    def generate_markdown_parts():
        """Helper function to generate markdown text split into chunks."""
        # Create a collapsible details block in parts
        details_start = "<details>\n  <summary>Click to expand</summary>\n\n"
        details_content = f"  **Here is some more information:** {' '.join(random.choices(words, k=random.randint(10, 20)))}\n"
        details_end = "</details>\n"

        # Randomly generate code blocks
        code_block_start = "\n```markdown\n"
        code_content = f"print('{random.choice(words)}')\n" * random.randint(1, 10)
        code_block_end = "```\n\n"

        normal_text = " ".join(random.choices(words, k=max_tokens // 4))

        # Return as broken up chunks
        parts = [
            normal_text,
            details_start,
            details_content[
                : len(details_content) // 2
            ],  # first part of the details content
            details_content[
                len(details_content) // 2 :
            ],  # second part of the details content
            details_end,
            code_block_start,
            code_content[: len(code_content) // 2],  # first half of the code
            code_content[len(code_content) // 2 :],  # second half of the code
            code_block_end,
        ]

        return parts

    # Simulate streaming random markdown text in chunks
    for _ in range(10):  # Stream 10 chunks of markdown text
        parts = generate_markdown_parts()
        for part in parts:
            yield part  # Yield each part to simulate streaming


model_registry = {"test-chat-model": text_generator}
app = create_openai_chat_server(model_registry)

# start by running
# python -m hypha_rpc.utils.serve openai_server:app --id=chat --name=chat --server-url=https://hypha.aicell.io --login

test_openai_server.py

from openai import AsyncOpenAI
import asyncio
from hypha_rpc import login


async def main():
    token = await login({"server_url": "https://hypha.aicell.io"})
    print("Token:", token)

    client = AsyncOpenAI(
        base_url=f"https://hypha.aicell.io/ws-user-github|478667/apps/chat/v1",
        api_key=token,
    )
    response = await client.chat.completions.create(
        model="test-chat-model",
        messages=[{"role": "user", "content": "Tell me a story."}],
        temperature=0.8,
        max_tokens=50,
        stream=True,
    )

    async for chunk in response:
        print("Chunk:", chunk.choices[0].delta.content)
        assert chunk.choices[0].delta.role == "assistant"


if __name__ == "__main__":
    asyncio.run(main())