Pipe Function

Overview

Think of IntraLLM AI as a system where messages flow through a processing pipeline. Pipes introduce new processing routes that can apply custom logic, call external services, and return results as if they were generated by a native model.

In this guide you will learn:

• How a Pipe fits into the IntraLLM AI workflow
• How to structure a Pipe (Valves, initialization, core logic)
• How to expose multiple selectable models from a single Pipe (manifold pattern)
• How to build a provider proxy (IntrallmAI example)
• How to call internal IntraLLM AI functions
• Best practices and security considerations

Conceptual model: Pipes and Valves

A useful mental model is a plumbing system:

• Pipes define the route and processing behavior.
• Valves define configuration parameters that control how the Pipe operates.

In IntraLLM AI:

• A Pipe is the Python class that contains your logic.
• Valves are persistent configuration fields (settings) exposed to the environment/UI.

Understanding the Pipe structure

A minimal Pipe follows a consistent structure:

1. Valves class at the top (configuration)
2. __init__ (initialize state and self.valves)
3. pipe(...) method (core processing logic)

Barebones example

from pydantic import BaseModel, Field

class Pipe:
    class Valves(BaseModel):
        MODEL_ID: str = Field(default="")

    def __init__(self):
        self.valves = self.Valves()

    def pipe(self, body: dict):
        # Custom logic goes here
        print(self.valves, body)  # Configuration + input payload
        return "Hello, World!"

Core components explained

The Pipe class

• Definition: The Pipe class is where you define your custom logic.
• Purpose: It acts as the blueprint for your plugin and determines how it behaves inside IntraLLM AI.

Valves: configuration for your Pipe

• Definition: Valves is a nested class inside Pipe, typically inheriting from pydantic.BaseModel.
• Purpose: It contains configuration options (parameters) that persist across uses of your Pipe.
• Concept: Treat Valves as persistent settings that control routing, provider configuration, feature toggles, and credentials.

Example:

from pydantic import BaseModel, Field

class Pipe:
    class Valves(BaseModel):
        MODEL_ID: str = Field(default="")

__init__: initialize Pipe state

• Definition: The constructor method for the Pipe class.
• Purpose: Initializes the Pipe’s state and sets up required components.
• Best practice: Keep initialization minimal; usually only initialize self.valves.

Example:

from pydantic import BaseModel, Field

class Pipe:
    class Valves(BaseModel):
        MODEL_ID: str = Field(default="")

    def __init__(self):
        self.valves = self.Valves()

pipe(...): core processing function

• Definition: The main function where your custom logic runs.
• Parameters:
  • body: a dictionary containing the input payload.
• Purpose: Processes input data and returns the result.

Example:

from pydantic import BaseModel, Field

class Pipe:
    class Valves(BaseModel):
        MODEL_ID: str = Field(default="")

    def __init__(self):
        self.valves = self.Valves()

    def pipe(self, body: dict):
        print(self.valves, body)
        return "Hello, World!"

Recommended class layout

To keep Pipes consistent and readable, follow this order:

• Valves (configuration)
• __init__ (initialization)
• pipes() (optional: manifold model list)
• pipe() (core logic)

Creating multiple models with a single Pipe (manifold pattern)

If you want one Pipe to expose multiple model entries in the UI, define a pipes() method that returns a list of model definitions. Each entry appears separately in the model selector.

from pydantic import BaseModel, Field

class Pipe:
    class Valves(BaseModel):
        MODEL_ID: str = Field(default="")

    def __init__(self):
        self.valves = self.Valves()

    def pipes(self):
        return [
            {"id": "model_id_1", "name": "model_1"},
            {"id": "model_id_2", "name": "model_2"},
            {"id": "model_id_3", "name": "model_3"},
        ]

    def pipe(self, body: dict):
        print(self.valves, body)  # Configuration + input payload
        model = body.get("model", "")
        return f"{model}: Hello, World!"

How it works:

• pipes() returns a list of {id, name} objects; each becomes a selectable model in the UI.
• pipe() can branch logic based on body.get("model").

Example: IntrallmAI Proxy Pipe

This example proxies requests to the IntrallmAI API. It:

• Fetches available models from IntrallmAI and exposes them in the UI
• Routes chat completion calls to the selected model
• Supports streaming responses when stream=True

from pydantic import BaseModel, Field
import requests

class Pipe:
    class Valves(BaseModel):
        NAME_PREFIX: str = Field(
            default="IntrallmAI/",
            description="Prefix added to model names in the UI.",
        )
        INTRALLMAI_API_BASE_URL: str = Field(
            default="https://api.intrallmai.com/v1",
            description="Base URL for IntrallmAI API endpoints.",
        )
        INTRALLMAI_API_KEY: str = Field(
            default="",
            description="API key used to authenticate IntrallmAI requests.",
        )

    def __init__(self):
        self.valves = self.Valves()

    def pipes(self):
        if self.valves.IntrallmAI_API_KEY:
            try:
                headers = {
                    "Authorization": f"Bearer {self.valves. INTRALLMAI_API_KEY}",
                    "Content-Type": "application/json",
                }

                r = requests.get(
                    f"{self.valves.INTRALLMAI_API_BASE_URL}/models",
                    headers=headers,
                )
                models = r.json()

                return [
                    {
                        "id": model["id"],
                        "name": f'{self.valves.NAME_PREFIX}{model.get("name", model["id"])}',
                    }
                    for model in models.get("data", [])
                    if "gpt" in model.get("id", "")
                ]

            except Exception:
                return [
                    {
                        "id": "error",
                        "name": "Error fetching models. Please check your API Key.",
                    },
                ]
        else:
            return [
                {
                    "id": "error",
                    "name": "API Key not provided.",
                },
            ]

    def pipe(self, body: dict, __user__: dict):
        headers = {
            "Authorization": f"Bearer {self.valves.INTRALLMAI_API_KEY}",
            "Content-Type": "application/json",
        }

        # Extract model id from the selected model string
        model_id = body["model"][body["model"].find(".") + 1 :]

        # Replace model with the extracted provider model id
        payload = {**body, "model": model_id}

        try:
            r = requests.post(
                url=f"{self.valves.INTRALLMAI_API_BASE_URL}/chat/completions",
                json=payload,
                headers=headers,
                stream=True,
            )
            r.raise_for_status()

            if body.get("stream", False):
                return r.iter_lines()
            return r.json()

        except Exception as e:
            return f"Error: {e}"

Extending the proxy Pipe to other providers

You can adapt the proxy pattern for other services (e.g., Anthropic, Perplexity, Vertex AI) by updating:

• API base URL and endpoints
• Authentication headers
• Request/response payload formats
• Model listing logic inside pipes()

Using internal IntraLLM AI functions

You can call internal IntraLLM AI utilities from the intrallmai package. Internal APIs can evolve, so confirm signatures against your current codebase.

from fastapi import Request
from intrallmai.models.users import Users
from intrallmai.utils.chat import generate_chat_completion

class Pipe:
    def __init__(self):
        pass

    async def pipe(
        self,
        body: dict,
        __user__: dict,
        __request__: Request,
    ) -> str:
        user = Users.get_user_by_id(__user__["id"])
        body["model"] = "llama3.2:latest"
        return await generate_chat_completion(__request__, body, user)

Best practices and security

• Structure consistently: keep Valves at the top, then __init__, then pipes() (optional), then pipe().
• Do not hard-code secrets: store API keys in Valves or secure configuration.
• Avoid logging secrets: do not print API keys or sensitive payload fields.
• Graceful error handling: wrap external calls in try/except and return readable error messages.
• Schema resilience: validate provider responses and handle missing keys safely.
• Dependency management: ensure required libraries exist in your deployment environment.

Frequently Asked Questions

Q1: Why should I use Pipes?

Pipes allow you to add new model-like capabilities with custom logic, provider proxies, and integrations without modifying the core platform code.

Q2: What are Valves, and why are they important?

Valves are persistent configuration parameters (settings) that control how your Pipe operates, enabling behavior changes without editing the code.

Q3: Can I create a Pipe without Valves?

Yes. Valves are optional, but recommended for flexibility and future scalability.

Q4: How do I ensure my Pipe is secure when using API keys?

Never hard-code API keys. Use Valves for secrets and ensure they are not logged or exposed.

Q5: What is the difference between pipe() and pipes()?

• pipe() processes requests and returns outputs for a selected model.
• pipes() returns a list of model definitions so one Pipe can appear as multiple models.

Q6: How can I handle errors in my Pipe?

Use try/except blocks inside pipes() and pipe(), and return meaningful error messages to the user.

Q7: Can I use external libraries in my Pipe?

Yes, as long as the dependencies are installed and compatible with your runtime environment.

Q8: How do I test my Pipe?

Run IntraLLM AI in a development environment, select the Pipe from the model selector, and validate behavior using different inputs and configurations.

Q9: Are there best practices for organizing my Pipe code?

Yes:

• Keep Valves at the top of the class
• Initialize self.valves in __init__
• Put pipe() after initialization
• Use clear naming and minimal logging

Q10: Where can I find the latest documentation?

Refer to your current IntraLLM AI repository or documentation for updated signatures and examples.

Conclusion

Pipes provide a flexible, model-like extension mechanism for IntraLLM AI. They can proxy providers, orchestrate workflows, and integrate external services while remaining modular and configurable through Valves. For production use, prioritize consistent structure, secure secret handling, and robust error management.