shuvi

English | Chinese

A lightweight, extensible async agent loop for building AI companions with Python. shuvi covers tool-using conversational agents, context management, completion verification, callbacks, and multi-agent handoffs.

Features

• Async agent runtime for iterative LLM calls and tool execution.
• Tool calling support through hydrogen-ai, including text and image results.
• Configurable loop controls for max iterations, retries, and completion verification.
• Callback hooks for streaming chunks, assistant messages, tool calls, tool results, errors, and state updates.

Development

Setup

# Clone the repo
git clone https://github.com/waifu-lab/Shuvi
cd Shuvi

# Install all workspace dependencies
uv sync

Example

Basic usage:

import asyncio
import os

from hai.providers.openai import OpenAI
from shuvi.agent import Agent
from shuvi.types import AgentConfig, UserMessage


async def main() -> None:
    provider = OpenAI(
        api_key=os.environ["OPENAI_API_KEY"],
        model="gpt-4o-mini",
        stream=False,
    )
    agent = Agent(
        AgentConfig(
            provider=provider,
            system_prompt="You are a helpful AI companion.",
        )
    )

    result = await agent.run(
        [
            UserMessage(
                content=[
                    {
                        "type": "text",
                        "text": "Hello!",
                    }
                ]
            )
        ]
    )
    print(result.messages[-1].content)


asyncio.run(main())

With tools:

import asyncio
import os

from hai.providers.openai import OpenAI
from hai.tool import tool
from shuvi.agent import Agent
from shuvi.types import AgentConfig, LoopConfig, UserMessage


@tool(
    description="Add two numbers together",
    parameters={
        "type": "object",
        "properties": {
            "a": {"type": "number"},
            "b": {"type": "number"},
        },
        "required": ["a", "b"],
    },
)
def add(a: float, b: float) -> float:
    return a + b


async def main() -> None:
    provider = OpenAI(
        api_key=os.environ["OPENAI_API_KEY"],
        model="gpt-4o-mini",
        stream=False,
    )
    agent = Agent(
        AgentConfig(
            provider=provider,
            system_prompt="You are a calculator assistant. Use tools when needed.",
            tools=[add],
            loop=LoopConfig(max_inner_iterations=10),
        )
    )

    result = await agent.run(
        [
            UserMessage(
                content=[
                    {
                        "type": "text",
                        "text": "Please calculate 123 + 456.",
                    }
                ]
            )
        ]
    )
    print(result.messages[-1].content)


asyncio.run(main())

Streaming mode:

import asyncio
import os

from hai.providers.openai import OpenAI
from shuvi.agent import Agent
from shuvi.types import AgentCallbacks, AgentConfig, UserMessage


def print_chunk(chunk: str) -> None:
    print(chunk, end="", flush=True)


async def main() -> None:
    provider = OpenAI(
        api_key=os.environ["OPENAI_API_KEY"],
        model="gpt-4o-mini",
        stream=True,
    )
    agent = Agent(
        AgentConfig(
            provider=provider,
            system_prompt="You are a helpful AI companion.",
            callbacks=AgentCallbacks(on_chunk=print_chunk),
        )
    )

    await agent.run(
        [
            UserMessage(
                content=[
                    {
                        "type": "text",
                        "text": "Introduce shuvi in three sentences.",
                    }
                ]
            )
        ]
    )
    print()


asyncio.run(main())

Testing

uv run pytest