Stateless Generation

Use ai-query’s core functions directly for maximum flexibility

The Agent class is optional. For simple use cases or when you need full control, use generate_text and stream_text directly.

Core Functions

ai-query provides two main functions:

Function	Description
generate_text()	Get a complete response
stream_text()	Stream response chunks

Both support tools, multiple providers, and conversation history.

Basic Usage

from ai_query import generate_text
from ai_query.providers import openai

result = await generate_text(
    model=openai("gpt-4o"),
    messages=[
        {"role": "user", "content": "What is Python?"}
    ]
)

print(result.text)
print(f"Tokens used: {result.usage.total_tokens}")

Streaming

from ai_query import stream_text
from ai_query.providers import openai

result = stream_text(
    model=openai("gpt-4o"),
    messages=[
        {"role": "user", "content": "Write a haiku about coding"}
    ]
)

async for chunk in result.text_stream:
    print(chunk, end="", flush=True)

# Access final usage after streaming
print(f"\nTokens: {result.usage.total_tokens}")

With Tools

from ai_query import generate_text
from ai_query.providers import openai, tool, Field

@tool(description="Get current weather for a location")
def get_weather(
    location: str = Field(description="City name")
) -> str:
    # Your weather API call here
    return f"Sunny, 72°F in {location}"

@tool(description="Search the web")
async def search(
    query: str = Field(description="Search query")
) -> str:
    # Your search implementation
    return f"Results for: {query}"

result = await generate_text(
    model=openai("gpt-4o"),
    messages=[{"role": "user", "content": "What's the weather in Tokyo?"}],
    tools={"get_weather": get_weather, "search": search}
)

print(result.text)

Conversation History

Manage history yourself with a list of messages.

from ai_query import generate_text
from ai_query.providers import openai

messages = []

async def chat(user_message: str) -> str:
    messages.append({"role": "user", "content": user_message})

    result = await generate_text(
        model=openai("gpt-4o"),
        messages=messages
    )

    messages.append({"role": "assistant", "content": result.text})
    return result.text

# Use it
await chat("Hello!")
await chat("What did I just say?")  # Model remembers

Simple Chatbot Class

Build a minimal chatbot without Agent.

from ai_query import generate_text, stream_text
from ai_query.providers import openai, tool, Field

class SimpleChatbot:
    def __init__(self, model, system: str = "You are a helpful assistant."):
        self.model = model
        self.system = system
        self.messages = []

    async def chat(self, message: str) -> str:
        self.messages.append({"role": "user", "content": message})

        result = await generate_text(
            model=self.model,
            system=self.system,
            messages=self.messages
        )

        self.messages.append({"role": "assistant", "content": result.text})
        return result.text

    async def stream(self, message: str):
        self.messages.append({"role": "user", "content": message})

        result = stream_text(
            model=self.model,
            system=self.system,
            messages=self.messages
        )

        full_response = ""
        async for chunk in result.text_stream:
            full_response += chunk
            yield chunk

        self.messages.append({"role": "assistant", "content": full_response})

    def clear(self):
        self.messages = []


# Usage
bot = SimpleChatbot(openai("gpt-4o"), system="You are a coding assistant.")

response = await bot.chat("How do I read a file in Python?")
print(response)

# Or stream
async for chunk in bot.stream("Now show me writing to a file"):
    print(chunk, end="")

Multi-Provider

Switch providers easily.

from ai_query import generate_text
from ai_query.providers import openai, anthropic, google

async def ask_all(question: str):
    providers = [
        ("OpenAI", openai("gpt-4o")),
        ("Anthropic", anthropic("claude-sonnet-4-20250514")),
        ("Google", google("gemini-2.0-flash")),
    ]

    for name, model in providers:
        result = await generate_text(
            model=model,
            messages=[{"role": "user", "content": question}]
        )
        print(f"{name}: {result.text[:100]}...")

With Stop Conditions

Control tool execution loops.

from ai_query import generate_text
from ai_query.providers import openai, tool, Field, step_count_is, has_tool_call

@tool(description="Execute a step")
def execute_step(step: str = Field(description="Step to execute")) -> str:
    return f"Completed: {step}"

# Stop after 5 tool calls
result = await generate_text(
    model=openai("gpt-4o"),
    messages=[{"role": "user", "content": "Plan and execute 3 steps"}],
    tools={"execute_step": execute_step},
    stop_when=step_count_is(5)
)

# Or stop when a specific tool is called
result = await generate_text(
    model=openai("gpt-4o"),
    messages=[{"role": "user", "content": "Complete the task"}],
    tools={"execute_step": execute_step, "finish": finish_tool},
    stop_when=has_tool_call("finish")
)

With Callbacks

Monitor execution with callbacks.

from ai_query import generate_text
from ai_query.providers import openai, tool, Field

@tool(description="Search")
def search(q: str = Field(description="Query")) -> str:
    return f"Results for {q}"

def on_step_start(event):
    print(f"Step {event.step_number} starting...")

def on_step_finish(event):
    if event.step.tool_calls:
        for tc in event.step.tool_calls:
            print(f"  Called: {tc.name}({tc.arguments})")

result = await generate_text(
    model=openai("gpt-4o"),
    messages=[{"role": "user", "content": "Search for Python tutorials"}],
    tools={"search": search},
    on_step_start=on_step_start,
    on_step_finish=on_step_finish
)

Embeddings

Generate embeddings without agents.

from ai_query import embed, embed_many
from ai_query.providers import openai

# Single text
embedding = await embed(
    model=openai("text-embedding-3-small"),
    text="Hello world"
)
print(f"Dimensions: {len(embedding.vector)}")

# Multiple texts
embeddings = await embed_many(
    model=openai("text-embedding-3-small"),
    texts=["Hello", "World", "Python"]
)
print(f"Got {len(embeddings.vectors)} embeddings")

When to Use What

Use Case	Approach
One-off generation	generate_text()
Simple streaming	stream_text()
Manual conversation	List of messages
Persistent state	Agent class
WebSocket server	Agent.serve()
Multi-user routing	AgentServer

Full Example: CLI Chat

A complete command-line chatbot.

import asyncio
from ai_query import stream_text
from ai_query.providers import openai, tool, Field

@tool(description="Get current time")
def get_time() -> str:
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

async def main():
    messages = []
    model = openai("gpt-4o")

    print("Chat (Ctrl+C to exit)")
    print("-" * 40)

    while True:
        try:
            user_input = input("\nYou: ")
            if not user_input.strip():
                continue

            messages.append({"role": "user", "content": user_input})

            print("Bot: ", end="", flush=True)

            result = stream_text(
                model=model,
                system="You are a helpful assistant.",
                messages=messages,
                tools={"get_time": get_time}
            )

            full_response = ""
            async for chunk in result.text_stream:
                print(chunk, end="", flush=True)
                full_response += chunk

            print()  # Newline
            messages.append({"role": "assistant", "content": full_response})

        except KeyboardInterrupt:
            print("\nGoodbye!")
            break

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