Messages

Message System Overview

Flo AI uses a flexible message system to handle communication between users, agents, and tools. Messages support text, images, documents, and structured content, making it easy to build rich conversational experiences.

Message Types

Flo AI supports several message types, each serving a specific purpose in conversations:

BaseMessage

The base class for all messages. All message types inherit from BaseMessage.

from flo_ai.models import BaseMessage

# BaseMessage has:
# - content: The message content (str, ImageMessageContent, DocumentMessageContent, or TextMessageContent)
# - role: The message role ('system', 'user', 'assistant', or 'function')
# - metadata: Optional dictionary for additional metadata

UserMessage

Messages from the user to the agent. Supports text, images, and documents.

from flo_ai.models import UserMessage

# Simple text message
user_msg = UserMessage(content="Hello, how are you?")

# UserMessage automatically sets role to 'user'
print(user_msg.role)  # 'user'

AssistantMessage

Messages from the agent to the user. These are typically generated by the LLM.

from flo_ai.models import AssistantMessage

# Assistant message
assistant_msg = AssistantMessage(content="I'm doing well, thank you!")

# AssistantMessage automatically sets role to 'assistant' if not specified
print(assistant_msg.role)  # 'assistant'

# You can customize the role
custom_msg = AssistantMessage(content="Response", role="user")

SystemMessage

System-level instructions for the agent. Used for system prompts.

from flo_ai.models import SystemMessage

# System message
system_msg = SystemMessage(content="You are a helpful assistant.")

# SystemMessage automatically sets role to 'system'
print(system_msg.role)  # 'system'

FunctionMessage

Messages representing function/tool call results. Used when agents call tools.

from flo_ai.models import FunctionMessage

# Function message (result of a tool call)
function_msg = FunctionMessage(
    content="The weather in San Francisco is 72°F and sunny.",
    name="get_weather"  # Name of the function that was called
)

# FunctionMessage automatically sets role to 'function'
print(function_msg.role)  # 'function'
print(function_msg.name)  # 'get_weather'

Message Content Types

Messages can contain different types of content:

Text Content

Simple string content (default):

from flo_ai.models import UserMessage, TextMessageContent

# Direct string (automatically converted)
msg1 = UserMessage(content="Hello")

# Explicit TextMessageContent
msg2 = UserMessage(
    content=TextMessageContent(text="Hello")
)

Image Content

Send images to agents for vision tasks:

from flo_ai.models import UserMessage, ImageMessageContent
import base64

# Image from URL
image_msg_url = UserMessage(
    content=ImageMessageContent(
        url="https://example.com/image.jpg",
        mime_type="image/jpeg"
    )
)

# Image from base64
with open('photo.jpg', 'rb') as f:
    image_bytes = f.read()
    image_base64 = base64.b64encode(image_bytes).decode('utf-8')

image_msg_base64 = UserMessage(
    content=ImageMessageContent(
        base64=image_base64,
        mime_type="image/jpeg"
    )
)

# Image from bytes
with open('photo.jpg', 'rb') as f:
    image_bytes = f.read()

image_msg_bytes = UserMessage(
    content=ImageMessageContent(
        bytes=image_bytes,
        mime_type="image/jpeg"
    )
)

Document Content

Send documents (PDFs, text files, etc.) to agents:

from flo_ai.models import UserMessage, DocumentMessageContent
from flo_ai.models.document import DocumentType
import base64

# PDF document from base64
with open('report.pdf', 'rb') as f:
    pdf_bytes = f.read()
    pdf_base64 = base64.b64encode(pdf_bytes).decode('utf-8')

document_msg = UserMessage(
    content=DocumentMessageContent(
        base64=pdf_base64,
        mime_type=DocumentType.PDF.value  # "application/pdf"
    )
)

# Document from URL
document_msg_url = UserMessage(
    content=DocumentMessageContent(
        url="https://example.com/document.pdf",
        mime_type="application/pdf"
    )
)

# Document from bytes
with open('report.pdf', 'rb') as f:
    pdf_bytes = f.read()

document_msg_bytes = UserMessage(
    content=DocumentMessageContent(
        bytes=pdf_bytes,
        mime_type="application/pdf"
    )
)

Using Messages with Agents

Basic Text Messages

from flo_ai.agent import AgentBuilder
from flo_ai.llm import OpenAI
from flo_ai.models import UserMessage

agent = (
    AgentBuilder()
    .with_name('Assistant')
    .with_prompt('You are a helpful assistant.')
    .with_llm(OpenAI(model='gpt-4o-mini'))
    .build()
)

# Simple string (automatically converted to UserMessage)
response = await agent.run("Hello!")

# Explicit UserMessage
response = await agent.run([UserMessage(content="Hello!")])

# Multiple messages
response = await agent.run([
    UserMessage(content="My name is Alice."),
    UserMessage(content="What's my name?")
])

Messages with Images

from flo_ai.models import UserMessage, ImageMessageContent
import base64

# Create agent with vision capabilities
vision_agent = (
    AgentBuilder()
    .with_name('Vision Assistant')
    .with_prompt('You are an expert at analyzing images.')
    .with_llm(OpenAI(model='gpt-4o'))  # Use vision-capable model
    .build()
)

# Load and encode image
with open('chart.png', 'rb') as f:
    image_bytes = f.read()
    image_base64 = base64.b64encode(image_bytes).decode('utf-8')

# Send image with text prompt
response = await vision_agent.run([
    UserMessage(
        content=ImageMessageContent(
            base64=image_base64,
            mime_type="image/png"
        )
    ),
    UserMessage(content="What does this chart show?")
])

Messages with Documents

from flo_ai.models import UserMessage, DocumentMessageContent
from flo_ai.models.document import DocumentType
import base64

# Create agent for document analysis
doc_agent = (
    AgentBuilder()
    .with_name('Document Analyst')
    .with_prompt('You analyze documents and extract key information.')
    .with_llm(OpenAI(model='gpt-4o'))
    .build()
)

# Load PDF document
with open('report.pdf', 'rb') as f:
    pdf_bytes = f.read()
    pdf_base64 = base64.b64encode(pdf_bytes).decode('utf-8')

# Send document with analysis request
response = await doc_agent.run([
    UserMessage(
        content=DocumentMessageContent(
            base64=pdf_base64,
            mime_type=DocumentType.PDF.value
        )
    ),
    UserMessage(content="Summarize the key points from this document.")
])

Mixed Content Messages

You can combine text, images, and documents in a single conversation:

# Multiple content types in one conversation
response = await agent.run([
    UserMessage(content="I have a question about this image:"),
    UserMessage(
        content=ImageMessageContent(
            url="https://example.com/diagram.png",
            mime_type="image/png"
        )
    ),
    UserMessage(content="And this document:"),
    UserMessage(
        content=DocumentMessageContent(
            url="https://example.com/spec.pdf",
            mime_type="application/pdf"
        )
    ),
    UserMessage(content="How do they relate?")
])

Message Metadata

All messages support optional metadata for additional context:

from flo_ai.models import UserMessage

# Message with metadata
msg = UserMessage(
    content="Hello",
    metadata={
        "timestamp": "2024-01-15T10:30:00Z",
        "user_id": "user123",
        "session_id": "session456",
        "source": "web_app"
    }
)

print(msg.metadata)  # {'timestamp': '...', 'user_id': '...', ...}

Working with Message Responses

Accessing Response Messages

Agents return lists of messages representing the conversation history:

response = await agent.run("What is 2+2?")

# Response is a list of BaseMessage objects
print(type(response))  # <class 'list'>

# Get the last message (agent's response)
last_message = response[-1]
print(last_message.content)  # "2+2 equals 4"
print(last_message.role)  # "assistant"

# Iterate through all messages
for msg in response:
    print(f"{msg.role}: {msg.content}")

Extracting Content

response = await agent.run("Hello!")

# Get content from last message
if response:
    last_msg = response[-1]
    
    # Handle different content types
    if isinstance(last_msg.content, str):
        text_content = last_msg.content
    elif hasattr(last_msg.content, 'text'):
        text_content = last_msg.content.text
    else:
        text_content = str(last_msg.content)
    
    print(text_content)

Message History

Agents maintain conversation history across multiple calls:

agent = (
    AgentBuilder()
    .with_name('Chat Assistant')
    .with_prompt('You are a helpful assistant.')
    .with_llm(OpenAI(model='gpt-4o-mini'))
    .build()
)

# First message
response1 = await agent.run("My name is Bob.")
print(f"Response: {response1[-1].content}")

# Second message - agent remembers context
response2 = await agent.run("What's my name?")
print(f"Response: {response2[-1].content}")  # "Your name is Bob."

# Full conversation history
for msg in response2:
    print(f"{msg.role}: {msg.content}")

Using Messages in Workflows

Messages work seamlessly in Arium workflows:

from flo_ai.arium import AriumBuilder
from flo_ai.models import UserMessage, ImageMessageContent
import base64

# Create workflow
workflow = (
    AriumBuilder()
    .add_agent(agent1)
    .add_agent(agent2)
    .start_with(agent1)
    .connect(agent1, agent2)
    .end_with(agent2)
    .build()
)

# Run workflow with messages
response = await workflow.run([
    UserMessage(content="Process this request"),
    UserMessage(
        content=ImageMessageContent(
            url="https://example.com/image.jpg",
            mime_type="image/jpeg"
        )
    )
])

Message Type Reference

MessageType Constants

from flo_ai.models.chat_message import MessageType

MessageType.USER      # 'user'
MessageType.ASSISTANT # 'assistant'
MessageType.FUNCTION  # 'function'
MessageType.SYSTEM    # 'system'

Message Classes

Class	Role	Content Types	Use Case
`UserMessage`	`user`	str, ImageMessageContent, DocumentMessageContent, TextMessageContent	User inputs
`AssistantMessage`	`assistant`	str	Agent responses
`SystemMessage`	`system`	str	System prompts
`FunctionMessage`	`function`	str	Tool/function results

Content Classes

Class	Type	Fields	Use Case
`TextMessageContent`	`text`	`text: str`	Plain text content
`ImageMessageContent`	`image`	`url`, `base64`, `bytes`, `mime_type`	Image content
`DocumentMessageContent`	`document`	`url`, `base64`, `bytes`, `mime_type`	Document content

Best Practices

1. Use Appropriate Content Types

# ✅ Good: Use ImageMessageContent for images
image_msg = UserMessage(
    content=ImageMessageContent(url="https://example.com/img.jpg")
)

# ❌ Avoid: Don't embed images as text
bad_msg = UserMessage(content="<img src='...'>")  # Not recommended

2. Handle Base64 Encoding Properly

import base64

# ✅ Good: Proper base64 encoding
with open('file.pdf', 'rb') as f:
    file_bytes = f.read()
    file_base64 = base64.b64encode(file_bytes).decode('utf-8')

# ❌ Avoid: Don't forget to decode bytes to string
bad_base64 = base64.b64encode(file_bytes)  # Returns bytes, not string

3. Specify MIME Types

# ✅ Good: Always specify mime_type
image_msg = ImageMessageContent(
    base64=image_base64,
    mime_type="image/png"  # Explicit MIME type
)

# ❌ Avoid: Missing mime_type may cause issues
bad_msg = ImageMessageContent(base64=image_base64)  # No MIME type

4. Use Metadata for Context

# ✅ Good: Add metadata for tracking
msg = UserMessage(
    content="Hello",
    metadata={
        "user_id": "123",
        "timestamp": "2024-01-15T10:30:00Z"
    }
)

5. Handle Message Lists Properly

# ✅ Good: Pass list of messages
response = await agent.run([
    UserMessage(content="First message"),
    UserMessage(content="Second message")
])

# ✅ Also good: Single string (auto-converted)
response = await agent.run("Single message")

# ❌ Avoid: Don't pass raw strings in lists without UserMessage
# response = await agent.run(["raw string"])  # May cause issues

Examples

from flo_ai.agent import AgentBuilder
from flo_ai.llm import OpenAI
from flo_ai.models import UserMessage, ImageMessageContent, DocumentMessageContent
import base64

agent = (
    AgentBuilder()
    .with_name('Multi-Modal Assistant')
    .with_prompt('You can analyze images and documents.')
    .with_llm(OpenAI(model='gpt-4o'))
    .build()
)

# Load image
with open('screenshot.png', 'rb') as f:
    image_base64 = base64.b64encode(f.read()).decode('utf-8')

# Load document
with open('spec.pdf', 'rb') as f:
    doc_base64 = base64.b64encode(f.read()).decode('utf-8')

# Multi-modal conversation
response = await agent.run([
    UserMessage(content="I need help with this:"),
    UserMessage(
        content=ImageMessageContent(
            base64=image_base64,
            mime_type="image/png"
        )
    ),
    UserMessage(
        content=DocumentMessageContent(
            base64=doc_base64,
            mime_type="application/pdf"
        )
    ),
    UserMessage(content="Compare the image with the document.")
])

Function Message Handling

from flo_ai.models import FunctionMessage

# Function messages are automatically created when tools are called
# But you can also create them manually if needed

function_result = FunctionMessage(
    content="The result of the function call",
    name="calculate_total"
)

# Function messages are typically used internally by the agent system
# when tools are executed

Message Validation

from flo_ai.models import UserMessage, BaseMessage

# Validate message type
msg = UserMessage(content="Hello")

if isinstance(msg, BaseMessage):
    print("Valid message")
    print(f"Role: {msg.role}")
    print(f"Content: {msg.content}")

The message system provides a flexible foundation for building rich, multi-modal AI applications with Flo AI!

Getting started

Core Features

Advanced

Examples

Message System Overview

Message Types

BaseMessage

UserMessage

AssistantMessage

SystemMessage

FunctionMessage

Message Content Types

Text Content

Image Content

Document Content

Using Messages with Agents

Basic Text Messages

Messages with Images

Messages with Documents

Mixed Content Messages

Message Metadata

Working with Message Responses

Accessing Response Messages

Extracting Content

Message History

Using Messages in Workflows

Message Type Reference

MessageType Constants

Message Classes

Content Classes

Best Practices

1. Use Appropriate Content Types

2. Handle Base64 Encoding Properly

3. Specify MIME Types

4. Use Metadata for Context

5. Handle Message Lists Properly

Examples

Function Message Handling

Message Validation

​Message System Overview

​Message Types

​BaseMessage

​UserMessage

​AssistantMessage

​SystemMessage

​FunctionMessage

​Message Content Types

​Text Content

​Image Content

​Document Content

​Using Messages with Agents

​Basic Text Messages

​Messages with Images

​Messages with Documents

​Mixed Content Messages

​Message Metadata

​Working with Message Responses

​Accessing Response Messages

​Extracting Content

​Message History

​Using Messages in Workflows

​Message Type Reference

​MessageType Constants

​Message Classes

​Content Classes

​Best Practices

​1. Use Appropriate Content Types

​2. Handle Base64 Encoding Properly

​3. Specify MIME Types

​4. Use Metadata for Context

​5. Handle Message Lists Properly

​Examples

​Multi-Modal Conversation

​Function Message Handling

​Message Validation

Message System Overview

Message Types

BaseMessage

UserMessage

AssistantMessage

SystemMessage

FunctionMessage

Message Content Types

Text Content

Image Content

Document Content

Using Messages with Agents

Basic Text Messages

Messages with Images

Messages with Documents

Mixed Content Messages

Message Metadata

Working with Message Responses

Accessing Response Messages

Extracting Content

Message History

Using Messages in Workflows

Message Type Reference

MessageType Constants

Message Classes

Content Classes

Best Practices

1. Use Appropriate Content Types

2. Handle Base64 Encoding Properly

3. Specify MIME Types

4. Use Metadata for Context

5. Handle Message Lists Properly

Examples

Multi-Modal Conversation

Function Message Handling

Message Validation