Memory and Chat

Vision Agents provides a conversation system that maintains context across interactions. The system supports persistent storage through Stream Chat (default) and in-memory storage for development.

Vision Agents uses Stream Video for real-time WebRTC transport by default. External WebRTC transports are supported as well. Most AI providers offer free tiers to get started.

Persistent conversations use Stream Chat automatically.

Persistent Conversations (Default)

By default, agents use StreamConversation which persists messages to Stream Chat. No additional setup required.

from vision_agents.core import User, Agent, Runner
from vision_agents.core.agents import AgentLauncher
from vision_agents.plugins import getstream, gemini

async def create_agent(**kwargs) -> Agent:
    return Agent(
        edge=getstream.Edge(),
        agent_user=User(name="Assistant", id="agent"),
        instructions="Remember details from our conversation across sessions.",
        llm=gemini.Realtime(),
    )

async def join_call(agent: Agent, call_type: str, call_id: str, **kwargs) -> None:
    await agent.create_user()
    call = await agent.create_call(call_type, call_id)

    async with agent.join(call):
        # Conversation automatically:
        # - Stores user messages from STT
        # - Stores agent responses from LLM
        # - Persists to Stream Chat
        # - Maintains context across sessions
        await agent.simple_response("Hello! I'll remember our conversation.")
        await agent.finish()

if __name__ == "__main__":
    Runner(AgentLauncher(create_agent=create_agent, join_call=join_call)).cli()

Messages are streamed to an ephemeral endpoint before persisting, ensuring real-time UI updates without affecting performance.

In-Memory Conversations

For development and testing, use InMemoryConversation:

from vision_agents.core.agents.conversation import InMemoryConversation

async def create_agent(**kwargs) -> Agent:
    llm = gemini.LLM()
    llm.set_conversation(InMemoryConversation("Be friendly", []))

    return Agent(
        edge=getstream.Edge(),
        agent_user=User(name="Assistant", id="agent"),
        instructions="You're a conversational AI assistant.",
        llm=llm,
    )

Custom Conversation Storage

Implement the Conversation abstract base class for custom storage:

from vision_agents.core.agents.conversation import Conversation, Message

class CustomConversation(Conversation):
    def add_message(self, message: Message, completed: bool = True):
        """Add a message to your custom storage."""
        pass

    def update_message(self, message_id: str, input_text: str, user_id: str,
                      replace_content: bool, completed: bool):
        """Update an existing message."""
        pass

Message Structure

Each message includes:

Field	Description
content	Message text
role	user or assistant
user_id	Sender identifier
timestamp	When the message was sent
id	Unique message identifier

Next Steps

Event System

React to conversation events

RAG for Agents

Add knowledge base access