Skip to main content

Error Handling

Chat includes built-in error handling with automatic retries for transient failures.

Retry Configuration

from upsonic import Agent, Chat

agent = Agent("openai/gpt-4o")

chat = Chat(
    session_id="session1",
    user_id="user1",
    agent=agent,
    retry_attempts=3,
    retry_delay=1.0
)

Error States

Chat tracks session state and handles errors:
from upsonic import Agent, Chat

agent = Agent("openai/gpt-4o")
chat = Chat(session_id="session1", user_id="user1", agent=agent)

# Check session state
if chat.state.value == "error":
    # Handle error state
    chat.reset_session()

Handling Errors

import asyncio
from upsonic import Agent, Chat

async def main():
    agent = Agent("openai/gpt-4o")
    chat = Chat(session_id="session1", user_id="user1", agent=agent)

    try:
        response = await chat.invoke("Hello")
        print(response)
    except Exception as e:
        print(f"Error: {e}")
        # Reset session if needed
        chat.reset_session()

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

Retryable Errors

Chat automatically retries on:
  • Network errors (ConnectionError, TimeoutError)
  • Rate limiting
  • Temporary service unavailability
  • Internal server errors
Non-retryable errors (validation errors, authentication failures) are raised immediately.

Concurrent Invocation Limits

from upsonic import Agent, Chat

agent = Agent("openai/gpt-4o")

chat = Chat(
    session_id="session1",
    user_id="user1",
    agent=agent,
    max_concurrent_invocations=2
)

# Can handle up to 2 concurrent invocations
# Additional calls will raise RuntimeError

Error Recovery

import asyncio
from upsonic import Agent, Chat


async def main():
    agent = Agent("openai/gpt-4o")
    chat = Chat(session_id="session1", user_id="user1", agent=agent)

    try:
        response = await chat.invoke("Hello")
        print(response)
    except RuntimeError as e:
        if "error state" in str(e):
            chat.reset_session()
            response = await chat.invoke("Hello")
            print(response)


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