Skip to main content

​
Use as Context Manager

Chat implements async context manager protocol for automatic resource cleanup.

​
Basic Usage

import asyncio
from upsonic import Agent, Chat


async def main():
    agent = Agent("openai/gpt-4o")
    
    async with Chat(session_id="session1", user_id="user1", agent=agent) as chat:
        response = await chat.invoke("Hello")
        print(response)
        # Storage connection automatically closed on exit


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

​
With Custom Storage

import asyncio
from upsonic import Agent, Chat
from upsonic.storage.providers import SqliteStorage


async def main():
    storage = SqliteStorage("sessions", "profiles", "chat.db")
    agent = Agent("openai/gpt-4o")

    async with Chat(
        session_id="session1",
        user_id="user1",
        agent=agent,
        storage=storage
    ) as chat:
        response = await chat.invoke("Hello")
        print(response)
        # Storage automatically disconnected


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

​
Error Handling with Context Manager

import asyncio
from upsonic import Agent, Chat


async def main():
    agent = Agent("openai/gpt-4o")
    
    try:
        async with Chat(session_id="session1", user_id="user1", agent=agent) as chat:
            response = await chat.invoke("Hello")
            print(response)
    except Exception as e:
        print(f"Error: {e}")
        # Resources still cleaned up automatically


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

​
Multiple Sessions

import asyncio
from upsonic import Agent, Chat


async def main():
    agent = Agent("openai/gpt-4o")

    async with Chat(session_id="session1", user_id="user1", agent=agent) as chat1:
        await chat1.invoke("Hello")

    async with Chat(session_id="session2", user_id="user1", agent=agent) as chat2:
        await chat2.invoke("Hi")
        # Each session manages its own resources


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