aiq.data_models.api_server

Attributes

UserContent

Classes

Message
AIQChatRequest	AIQChatRequest is a data model that represents a request to the AgentIQ chat API.
AIQChoiceMessage
AIQChoice
AIQUsage
AIQResponseSerializable	AIQChatResponseSerializable is an abstract class that defines the interface for serializing output for the AgentIQ
AIQResponseBaseModelOutput
AIQResponseBaseModelIntermediate
AIQChatResponse	AIQChatResponse is a data model that represents a response from the AgentIQ chat API.
AIQChatResponseChunk	AIQChatResponseChunk is a data model that represents a response chunk from the AgentIQ chat streaming API.
AIQResponseIntermediateStep	AIQResponseSerializedStep is a data model that represents a serialized step in the AgentIQ chat streaming API.
AIQResponsePayloadOutput
AIQGenerateResponse
UserMessageContentRoleType	str(object='') -> str
ChatContentType	ChatContentType is an Enum that represents the type of Chat content.
WebSocketMessageType	WebSocketMessageType is an Enum that represents WebSocket Message types.
WorkflowSchemaType	WorkflowSchemaType is an Enum that represents Workkflow response types.
WebSocketMessageStatus	WebSocketMessageStatus is an Enum that represents the status of a WebSocket message.
InputAudio
AudioContent
ImageUrl
ImageContent
TextContent
Security
UserMessages
UserMessageContent
User
ErrorTypes	str(object='') -> str
Error
WebSocketUserMessage	For more details, refer to the API documentation:
WebSocketUserInteractionResponseMessage	For more details, refer to the API documentation:
SystemIntermediateStepContent
WebSocketSystemIntermediateStepMessage	For more details, refer to the API documentation:
SystemResponseContent
WebSocketSystemResponseTokenMessage	For more details, refer to the API documentation:
WebSocketSystemInteractionMessage	For more details, refer to the API documentation:

Functions

_generate_response_to_str(→ str)
_generate_response_to_chat_response(→ AIQChatResponse)
_aiq_chat_request_to_string(→ str)
_string_to_aiq_chat_request(→ AIQChatRequest)
_aiq_chat_response_to_string(→ str)
_string_to_aiq_chat_response(→ AIQChatResponse)	Converts a string to an AIQChatResponse object
_chat_response_to_chat_response_chunk(...)
_aiq_chat_response_chunk_to_string(→ str)
_string_to_aiq_chat_response_chunk(→ AIQChatResponseChunk)	Converts a string to an AIQChatResponseChunk object

Module Contents

class Message(/, **data: Any)

Bases: pydantic.BaseModel

content: str

role: str

class AIQChatRequest(/, **data: Any)

Bases: pydantic.BaseModel

AIQChatRequest is a data model that represents a request to the AgentIQ chat API.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

messages: Annotated[list[Message], conlist(Message, min_length=1)]

model: str | None = None

temperature: float | None = None

max_tokens: int | None = None

top_p: float | None = None

static from_string(

data: str,

*,

model: str | None = None,

temperature: float | None = None,

max_tokens: int | None = None,

top_p: float | None = None,

) → AIQChatRequest

class AIQChoiceMessage(/, **data: Any)

Bases: pydantic.BaseModel

content: str | None = None

role: str | None = None

class AIQChoice(/, **data: Any)

Bases: pydantic.BaseModel

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

message: AIQChoiceMessage

finish_reason: Literal['stop', 'length', 'tool_calls', 'content_filter', 'function_call'] | None = None

index: int

class AIQUsage(/, **data: Any)

Bases: pydantic.BaseModel

prompt_tokens: int

completion_tokens: int

total_tokens: int

class AIQResponseSerializable

Bases: abc.ABC

AIQChatResponseSerializable is an abstract class that defines the interface for serializing output for the AgentIQ chat streaming API.

abstractmethod get_stream_data() → str

class AIQResponseBaseModelOutput(/, **data: Any)

Bases: pydantic.BaseModel, AIQResponseSerializable

get_stream_data() → str

class AIQResponseBaseModelIntermediate(/, **data: Any)

Bases: pydantic.BaseModel, AIQResponseSerializable

get_stream_data() → str

class AIQChatResponse(/, **data: Any)

Bases: AIQResponseBaseModelOutput

AIQChatResponse is a data model that represents a response from the AgentIQ chat API.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

id: str

object: str

model: str = ''

created: datetime.datetime

choices: list[AIQChoice]

usage: AIQUsage | None = None

static from_string(

data: str,

*,

id_: str | None = None,

object_: str | None = None,

model: str | None = None,

created: datetime.datetime | None = None,

usage: AIQUsage | None = None,

) → AIQChatResponse

class AIQChatResponseChunk(/, **data: Any)

Bases: AIQResponseBaseModelOutput

AIQChatResponseChunk is a data model that represents a response chunk from the AgentIQ chat streaming API.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

id: str

choices: list[AIQChoice]

created: datetime.datetime

model: str = ''

object: str = 'chat.completion.chunk'

static from_string(

data: str,

*,

id_: str | None = None,

created: datetime.datetime | None = None,

model: str | None = None,

object_: str | None = None,

) → AIQChatResponseChunk

class AIQResponseIntermediateStep(/, **data: Any)

Bases: AIQResponseBaseModelIntermediate

AIQResponseSerializedStep is a data model that represents a serialized step in the AgentIQ chat streaming API.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

id: str

parent_id: str | None = None

type: str = 'markdown'

name: str

payload: str

class AIQResponsePayloadOutput(/, **data: Any)

Bases: pydantic.BaseModel, AIQResponseSerializable

payload: Any

get_stream_data() → str

class AIQGenerateResponse(/, **data: Any)

Bases: pydantic.BaseModel

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

intermediate_steps: list[tuple] | None = None

output: str

value: str | None = 'default'

class UserMessageContentRoleType

Bases: str, enum.Enum

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

Initialize self. See help(type(self)) for accurate signature.

USER = 'user'

ASSISTANT = 'assistant'

class ChatContentType

Bases: str, enum.Enum

ChatContentType is an Enum that represents the type of Chat content.

Initialize self. See help(type(self)) for accurate signature.

TEXT = 'text'

IMAGE_URL = 'image_url'

INPUT_AUDIO = 'input_audio'

class WebSocketMessageType

Bases: str, enum.Enum

WebSocketMessageType is an Enum that represents WebSocket Message types.

Initialize self. See help(type(self)) for accurate signature.

USER_MESSAGE = 'user_message'

RESPONSE_MESSAGE = 'system_response_message'

INTERMEDIATE_STEP_MESSAGE = 'system_intermediate_message'

SYSTEM_INTERACTION_MESSAGE = 'system_interaction_message'

USER_INTERACTION_MESSAGE = 'user_interaction_message'

ERROR_MESSAGE = 'error_message'

class WorkflowSchemaType

Bases: str, enum.Enum

WorkflowSchemaType is an Enum that represents Workkflow response types.

Initialize self. See help(type(self)) for accurate signature.

GENERATE_STREAM = 'generate_stream'

CHAT_STREAM = 'chat_stream'

GENERATE = 'generate'

CHAT = 'chat'

class WebSocketMessageStatus

Bases: str, enum.Enum

WebSocketMessageStatus is an Enum that represents the status of a WebSocket message.

Initialize self. See help(type(self)) for accurate signature.

IN_PROGRESS = 'in_progress'

COMPLETE = 'complete'

class InputAudio(/, **data: Any)

Bases: pydantic.BaseModel

data: str = 'default'

format: str = 'default'

class AudioContent(/, **data: Any)

Bases: pydantic.BaseModel

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal[ChatContentType]

input_audio: InputAudio

class ImageUrl(/, **data: Any)

Bases: pydantic.BaseModel

url: pydantic.HttpUrl

class ImageContent(/, **data: Any)

Bases: pydantic.BaseModel

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal[ChatContentType]

image_url: ImageUrl

class TextContent(/, **data: Any)

Bases: pydantic.BaseModel

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal[ChatContentType]

text: str = 'default'

class Security(/, **data: Any)

Bases: pydantic.BaseModel

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

api_key: str = 'default'

token: str = 'default'

UserContent

class UserMessages(/, **data: Any)

Bases: pydantic.BaseModel

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

role: UserMessageContentRoleType

content: list[UserContent]

class UserMessageContent(/, **data: Any)

Bases: pydantic.BaseModel

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

messages: list[UserMessages]

class User(/, **data: Any)

Bases: pydantic.BaseModel

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str = 'default'

email: str = 'default'

class ErrorTypes

Bases: str, enum.Enum

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

Initialize self. See help(type(self)) for accurate signature.

UNKNOWN_ERROR = 'unknown_error'

INVALID_MESSAGE = 'invalid_message'

INVALID_MESSAGE_TYPE = 'invalid_message_type'

INVALID_USER_MESSAGE_CONTENT = 'invalid_user_message_content'

INVALID_DATA_CONTENT = 'invalid_data_content'

class Error(/, **data: Any)

Bases: pydantic.BaseModel

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

code: ErrorTypes

message: str = 'default'

details: str = 'default'

class WebSocketUserMessage(/, **data: Any)

Bases: pydantic.BaseModel

For more details, refer to the API documentation: docs/source/developer_guide/websockets.md

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal[WebSocketMessageType]

schema_type: WorkflowSchemaType

id: str = 'default'

thread_id: str = 'default'

content: UserMessageContent

user: User

security: Security

error: Error

schema_version: str = '1.0.0'

timestamp: str = ''

class WebSocketUserInteractionResponseMessage(/, **data: Any)

Bases: pydantic.BaseModel

For more details, refer to the API documentation: docs/source/developer_guide/websockets.md

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

type: Literal[WebSocketMessageType]

id: str = 'default'

thread_id: str = 'default'

content: UserMessageContent

user: User

security: Security

error: Error

schema_version: str = '1.0.0'

timestamp: str = ''

class SystemIntermediateStepContent(/, **data: Any)

Bases: pydantic.BaseModel

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

payload: str

class WebSocketSystemIntermediateStepMessage(/, **data: Any)

Bases: pydantic.BaseModel

For more details, refer to the API documentation: docs/source/developer_guide/websockets.md

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal[WebSocketMessageType]

id: str = 'default'

thread_id: str | None = 'default'

parent_id: str = 'default'

intermediate_parent_id: str | None = 'default'

update_message_id: str | None = 'default'

content: SystemIntermediateStepContent

status: WebSocketMessageStatus

timestamp: str = ''

class SystemResponseContent(/, **data: Any)

Bases: pydantic.BaseModel

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

text: str | None = None

class WebSocketSystemResponseTokenMessage(/, **data: Any)

Bases: pydantic.BaseModel

For more details, refer to the API documentation: docs/source/developer_guide/websockets.md

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal[WebSocketMessageType, WebSocketMessageType]

id: str | None = 'default'

thread_id: str | None = 'default'

parent_id: str = 'default'

content: SystemResponseContent | Error | AIQGenerateResponse

status: WebSocketMessageStatus

timestamp: str = ''

classmethod validate_content_by_type(

value: SystemResponseContent | Error | AIQGenerateResponse,

info: pydantic_core.core_schema.ValidationInfo,

)

class WebSocketSystemInteractionMessage(/, **data: Any)

Bases: pydantic.BaseModel

For more details, refer to the API documentation: docs/source/developer_guide/websockets.md

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal[WebSocketMessageType]

id: str | None = 'default'

thread_id: str | None = 'default'

parent_id: str = 'default'

content: aiq.data_models.interactive.HumanPrompt

status: WebSocketMessageStatus

timestamp: str = ''

_generate_response_to_str(response: AIQGenerateResponse) → str

_generate_response_to_chat_response(

response: AIQGenerateResponse,

) → AIQChatResponse

_aiq_chat_request_to_string(data: AIQChatRequest) → str

_string_to_aiq_chat_request(data: str) → AIQChatRequest

_aiq_chat_response_to_string(data: AIQChatResponse) → str

_string_to_aiq_chat_response(data: str) → AIQChatResponse

Converts a string to an AIQChatResponse object

_chat_response_to_chat_response_chunk(

data: AIQChatResponse,

) → AIQChatResponseChunk

_aiq_chat_response_chunk_to_string(data: AIQChatResponseChunk) → str

_string_to_aiq_chat_response_chunk(data: str) → AIQChatResponseChunk

Converts a string to an AIQChatResponseChunk object