Custom LLM

In Conversational AI Engine interaction scenarios, your use case may require a custom large language model (Custom LLM). This document explains how to integrate a custom LLM into Agora's Conversational AI Engine.

Understand the tech

Agora's Conversational AI Engine interacts with LLM services using the OpenAI API protocol. To integrate a custom LLM, you need to provide an HTTP service compatible with the OpenAI API, capable of handling requests and responses in the OpenAI API format.

This approach enables you to implement additional custom functionalities, including but not limited to:

Retrieval-Augmented Generation (RAG): Allows the model to retrieve information from a specific knowledge base.
Multimodal Capabilities: Enables the model to generate output in both text and audio formats.
Tool Invocation: Allows the model to call external tools.
Function Calling: Enables the model to return structured data in the form of function calls.

Prerequisites

Before you begin, ensure that you have:

Implemented the basic logic for interacting with a Conversational AI agent by following the REST Quickstart.
Set up access to a custom LLM service.
Prepared a vector database or retrieval system if using Retrieval-Augmented Generation (RAG).

Implementation

Take the following steps to integrate your custom LLM into Agora's Conversational AI Engine.

Create an OpenAI API-compatible service

To integrate successfully with Agora's Conversational AI Engine, your custom LLM service must provide an interface compatible with the OpenAI Chat Completions API. The key requirements include:

Endpoint: A request-handling endpoint, such as https://your-custom-llm-service/chat/completions.
Request format: Must accept request parameters adhering to the OpenAI API protocol.
Response format: Should return OpenAI API-compatible responses and support the Server-Sent Events (SSE) standard for streaming.

The following example demonstrates how to implement an OpenAI API-compliant interface:

class TextContent(BaseModel):
    type: str = "text"
    text: str
class ImageContent(BaseModel):
    type: str = "image"
    image_url: HttpUrl
class AudioContent(BaseModel):
    type: str = "input_audio"
    input_audio: Dict[str, str]
class ToolFunction(BaseModel):
    name: str
    description: Optional[str]
    parameters: Optional[Dict]
    strict: bool = False
class Tool(BaseModel):
    type: str = "function"
    function: ToolFunction
class ToolChoice(BaseModel):
    type: str = "function"
    function: Optional[Dict]
class ResponseFormat(BaseModel):
    type: str = "json_schema"
    json_schema: Optional[Dict[str, str]]
class SystemMessage(BaseModel):
    role: str = "system"
    content: Union[str, List[str]]
class UserMessage(BaseModel):
    role: str = "user"
    content: Union[str, List[Union[TextContent, ImageContent, AudioContent]]]
class AssistantMessage(BaseModel):
    role: str = "assistant"
    content: Union[str, List[TextContent]] = None
    audio: Optional[Dict[str, str]] = None
    tool_calls: Optional[List[Dict]] = None
class ToolMessage(BaseModel):
    role: str = "tool"
    content: Union[str, List[str]]
    tool_call_id: str
# Define the complete request format
class ChatCompletionRequest(BaseModel):
    context: Optional[Dict] = None  # Context information
    model: Optional[str] = None  # Model name being used
    messages: List[Union[SystemMessage, UserMessage, AssistantMessage, ToolMessage]]  # List of messages
    response_format: Optional[ResponseFormat] = None  # Response format
    modalities: List[str] = ["text"]  # Default modality is text
    audio: Optional[Dict[str, str]] = None  # Assistant's audio response
    tools: Optional[List[Tool]] = None  # List of tools
    tool_choice: Optional[Union[str, ToolChoice]] = "auto"  # Tool selection
    parallel_tool_calls: bool = True  # Whether to call tools in parallel
    stream: bool = True  # Default to streaming response
    stream_options: Optional[Dict] = None  # Streaming options
@app.post("/chat/completions")
async def create_chat_completion(request: ChatCompletionRequest):
    try:
        logger.info(f"Received request: {request.model_dump_json()}")
        client = AsyncOpenAI(api_key=os.getenv("YOUR_LLM_API_KEY"))
        response = await client.chat.completions.create(
            model=request.model,
            messages=request.messages,  # Directly use request messages
            tool_choice=(
                request.tool_choice if request.tools and request.tool_choice else None
            ),
            tools=request.tools if request.tools else None,
            modalities=request.modalities,
            audio=request.audio,
            response_format=request.response_format,
            stream=request.stream,
            stream_options=request.stream_options,
        )
        if not request.stream:
            raise HTTPException(
                status_code=400, detail="chat completions require streaming"
            )
        async def generate():
            try:
                async for chunk in response:
                    logger.debug(f"Received chunk: {chunk}")
                    yield f"data: {json.dumps(chunk.to_dict())}\n\n"
                yield "data: [DONE]\n\n"
            except asyncio.CancelledError:
                logger.info("Request was cancelled")
                raise
        return StreamingResponse(generate(), media_type="text/event-stream")
    except asyncio.CancelledError:
        logger.info("Request was cancelled")
        raise HTTPException(status_code=499, detail="Request was cancelled")
    except Exception as e:
        traceback_str = "".join(traceback.format_tb(e.__traceback__))
        error_message = f"{str(e)}\n{traceback_str}"
        logger.error(error_message)
        raise HTTPException(status_code=500, detail=error_message)

When calling the POST method to Start a conversational AI agent, use the LLM configuration to point your agent to the custom service:

{
  "llm": {
     "url": "https://your-custom-llm-service/chat/completions",
    "api_key": "",
     "system_messages": [
      {
        "role": "system",
        "content": "You are a helpful assistant."
      }
    ]
  }
}

info

If accessing your custom LLM service requires identity verification, provide the authentication information in the api_key field.

Advanced features

To integrate advanced features such as Retrieval-Augmented Generation and generating outputs in multimodal forms, refer to the following sections.

Retrieval-Augmented Generation

To improve the accuracy and relevance of the agent's responses, use the Retrieval-Augmented Generation (RAG) feature. This feature allows your custom LLM to retrieve information from a specific knowledge base and use the retrieved results as context for generating responses.

The following example simulates the process of retrieving and returning content from a knowledge base and creates the /rag/chat/completions endpoint to incorporate RAG retrieval results when generating responses with the LLM.

async def perform_rag_retrieval(messages: Optional[Dict]) -> str:
    """
    Retrieve relevant content from the knowledge base using the RAG model.
    Args:
        messages: The original message list.
    Returns:
        str: The retrieved text content.
    """
    # TODO: Implement the actual RAG retrieval logic.
    # You can choose the first or last message from the message list as the query,
    # then send the query to the RAG model to retrieve relevant content.
    # Return the retrieval result.
    return "This is relevant content retrieved from the knowledge base."
def refact_messages(context: str, messages: Optional[Dict] = None) -> Optional[Dict]:
    """
    Modify the message list by adding the retrieved context to the original messages.
    Args:
        context: The retrieved context.
        messages: The original message list.
    Returns:
        List: The modified message list.
    """
    # TODO: Implement the actual message modification logic.
    # This should add the retrieved context to the original message list.
    return messages
# Random waiting messages.
waiting_messages = [
    "Just a moment, I'm thinking...",
    "Let me think about that for a second...",
    "Good question, let me find out...",
]
@app.post("/rag/chat/completions")
async def create_rag_chat_completion(request: ChatCompletionRequest):
    try:
        logger.info(f"Received RAG request: {request.model_dump_json()}")
        if not request.stream:
            raise HTTPException(
                status_code=400, detail="chat completions require streaming"
            )
        async def generate():
            # First, send a "please wait" prompt.
            waiting_message = {
                "id": "waiting_msg",
                "choices": [
                    {
                        "index": 0,
                        "delta": {
                            "role": "assistant",
                            "content": random.choice(waiting_messages),
                        },
                        "finish_reason": None,
                    }
                ],
            }
            yield f"data: {json.dumps(waiting_message)}\n\n"
            # Perform RAG retrieval.
            retrieved_context = await perform_rag_retrieval(request.messages)
            # Modify messages.
            refacted_messages = refact_messages(retrieved_context, request.messages)
            # Request LLM completion.
            client = AsyncOpenAI(api_key=os.getenv("<YOUR_LLM_API_KEY>"))
            response = await client.chat.completions.create(
                model=request.model,
                messages=refacted_messages,
                tool_choice=(
                    request.tool_choice
                    if request.tools and request.tool_choice
                    else None
                ),
                tools=request.tools if request.tools else None,
                modalities=request.modalities,
                audio=request.audio,
                response_format=request.response_format,
                stream=True,  # Force streaming.
                stream_options=request.stream_options,
            )
            try:
                async for chunk in response:
                    logger.debug(f"Received RAG chunk: {chunk}")
                    yield f"data: {json.dumps(chunk.to_dict())}\n\n"
                yield "data: [DONE]\n\n"
            except asyncio.CancelledError:
                logger.info("RAG stream was cancelled")
                raise
        return StreamingResponse(generate(), media_type="text/event-stream")
    except asyncio.CancelledError:
        logger.info("RAG request was cancelled")
        raise HTTPException(status_code=499, detail="Request was cancelled")
    except Exception as e:
        traceback_str = "".join(traceback.format_tb(e.__traceback__))
        error_message = f"{str(e)}\n{traceback_str}"
        logger.error(error_message)
        raise HTTPException(status_code=500, detail=error_message)

When calling the POST method to Start a conversational AI agent, simply point the LLM URL to your RAG interface:

{
  "llm": {
    "url": "http://your-custom-llm-service/rag/chat/completions",
    "api_key": ""
    "system_messages": [
      {
        "role": "system",
        "content": "Please answer the user's question based on the following retrieved information: ..."
      }
    ]
  }
}

info

If accessing your custom LLM service requires identity verification, provide the authentication information in the api_key field.

Multimodal capabilities

Conversational AI Engine supports LLMs in generating output in multimodal formats, including text and audio. You can create dedicated multimodal interfaces to meet personalized requirements.

The following example demonstrates how to read text and audio files and send them to an LLM to generate an audio response.

async def read_text_file(file_path: str) -> str:
    """
    Read a text file and return its contents.
    Args:
        file_path: Path to the text file.
    Returns:
        str: Contents of the text file.
    """
    async with aiofiles.open(file_path, "r") as file:
        content = await file.read()
    return content
async def read_pcm_file(
    file_path: str, sample_rate: int, duration_ms: int
) -> List[bytes]:
    """
    Read a PCM file and return a list of audio chunks.
    Args:
        file_path: Path to the PCM file.
        sample_rate: Sample rate of the audio.
        duration_ms: Duration of each audio chunk in milliseconds.
    Returns:
        List: List of audio chunks.
    """
    async with aiofiles.open(file_path, "rb") as file:
        content = await file.read()
    chunk_size = int(sample_rate * 2 * (duration_ms / 1000))
    return [content[i : i + chunk_size] for i in range(0, len(content), chunk_size)]
@app.post("/audio/chat/completions")
async def create_audio_chat_completion(request: ChatCompletionRequest):
    try:
        logger.info(f"Received audio request: {request.model_dump_json()}")
        if not request.stream:
            raise HTTPException(
                status_code=400, detail="chat completions require streaming"
            )
        # Example usage: Reading text and audio files
        # Replace with your actual logic
        text_file_path = "./file.txt"
        pcm_file_path = "./file.pcm"
        sample_rate = 16000  # Example sample rate
        duration_ms = 40  # 40ms audio chunks
        text_content = await read_text_file(text_file_path)
        audio_chunks = await read_pcm_file(pcm_file_path, sample_rate, duration_ms)
        async def generate():
            try:
                # Send text content
                audio_id = uuid.uuid4().hex
                text_message = {
                    "id": uuid.uuid4().hex,
                    "choices": [
                        {
                            "index": 0,
                            "delta": {
                                "audio": {
                                    "id": audio_id,
                                    "transcript": text_content,
                                },
                            },
                            "finish_reason": None,
                        }
                    ],
                }
                yield f"data: {json.dumps(text_message)}\n\n"
                # Send audio chunks
                for chunk in audio_chunks:
                    audio_message = {
                        "id": uuid.uuid4().hex,
                        "choices": [
                            {
                                "index": 0,
                                "delta": {
                                    "audio": {
                                        "id": audio_id,
                                        "data": base64.b64encode(chunk).decode("utf-8"),
                                    },
                                },
                                "finish_reason": None,
                            }
                        ],
                    }
                    yield f

When calling the POST method to Start a conversational AI agent, use the following configuration to specify the output_modalities:

{
  "llm": {
    "url": "http://your-custom-llm-service/audio/chat/completions",
    "api_key": "your_api_key",
    "input_modalities": ["text"],
    "output_modalities": ["text", "audio"]
    "system_messages": [
      {
        "role": "system",
        "content": "You are a helpful assistant."
      }
    ]
  }
}

Request format: Contains parameters such as model, message, and tool call configuration.
Response format: Contains the response generated by the model, metadata, and other information
Streaming response: Compliant with the SSE (Server-Sent Events) specification

For detailed interface standards, please refer to:

Custom LLM

Understand the tech

Prerequisites

Implementation

Create an OpenAI API-compatible service

Advanced features

Retrieval-Augmented Generation

Multimodal capabilities

Reference

Sample project

Interface standards