LLM Chain: Added foundation for chain execution (#616)

Author: vprashrex

backend/app/alembic/versions/048_create_llm_chain_table.py

@@ -0,0 +1,169 @@
+"""Create llm_chain table
+
+Revision ID: 048
+Revises: 047
+Create Date: 2026-02-20 00:00:00.000000
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects.postgresql import JSONB
+
+revision = "048"
+down_revision = "047"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # 1. Create llm_chain table
+    op.create_table(
+        "llm_chain",
+        sa.Column(
+            "id",
+            sa.Uuid(),
+            nullable=False,
+            comment="Unique identifier for the LLM chain record",
+        ),
+        sa.Column(
+            "job_id",
+            sa.Uuid(),
+            nullable=False,
+            comment="Reference to the parent job (status tracked in job table)",
+        ),
+        sa.Column(
+            "project_id",
+            sa.Integer(),
+            nullable=False,
+            comment="Reference to the project this LLM call belongs to",
+        ),
+        sa.Column(
+            "organization_id",
+            sa.Integer(),
+            nullable=False,
+            comment="Reference to the organization this LLM call belongs to",
+        ),
+        sa.Column(
+            "status",
+            sa.String(),
+            nullable=False,
+            server_default="pending",
+            comment="Chain execution status (pending, running, failed, completed)",
+        ),
+        sa.Column(
+            "error",
+            sa.Text(),
+            nullable=True,
+            comment="Error message if the chain execution failed",
+        ),
+        sa.Column(
+            "block_sequences",
+            JSONB(),
+            nullable=True,
+            comment="Ordered list of llm_call UUIDs as blocks complete",
+        ),
+        sa.Column(
+            "total_blocks",
+            sa.Integer(),
+            nullable=False,
+            comment="Total number of blocks to execute",
+        ),
+        sa.Column(
+            "number_of_blocks_processed",
+            sa.Integer(),
+            nullable=False,
+            server_default="0",
+            comment="Number of blocks processed so far (used for tracking progress)",
+        ),
+        sa.Column(
+            "input",
+            sa.String(),
+            nullable=False,
+            comment="First block user's input - text string, binary data, or file path for multimodal",
+        ),
+        sa.Column(
+            "output",
+            JSONB(),
+            nullable=True,
+            comment="Last block's final output (set on chain completion)",
+        ),
+        sa.Column(
+            "configs",
+            JSONB(),
+            nullable=True,
+            comment="Ordered list of block configs as submitted in the request",
+        ),
+        sa.Column(
+            "total_usage",
+            JSONB(),
+            nullable=True,
+            comment="Aggregated token usage: {input_tokens, output_tokens, total_tokens}",
+        ),
+        sa.Column(
+            "metadata",
+            JSONB(),
+            nullable=True,
+            comment="Future-proof extensibility catch-all",
+        ),
+        sa.Column(
+            "inserted_at",
+            sa.DateTime(),
+            nullable=False,
+            comment="Timestamp when the chain record was created",
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(),
+            nullable=False,
+            comment="Timestamp when the chain record was last updated",
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.ForeignKeyConstraint(["job_id"], ["job.id"], ondelete="CASCADE"),
+        sa.ForeignKeyConstraint(["project_id"], ["project.id"], ondelete="CASCADE"),
+        sa.ForeignKeyConstraint(
+            ["organization_id"], ["organization.id"], ondelete="CASCADE"
+        ),
+    )
+
+    op.create_index(
+        "idx_llm_chain_job_id",
+        "llm_chain",
+        ["job_id"],
+    )
+
+    # 2. Add chain_id FK column to llm_call table
+    op.add_column(
+        "llm_call",
+        sa.Column(
+            "chain_id",
+            sa.Uuid(),
+            nullable=True,
+            comment="Reference to the parent chain (NULL for standalone /llm/call requests)",
+        ),
+    )
+    op.create_foreign_key(
+        "fk_llm_call_chain_id",
+        "llm_call",
+        "llm_chain",
+        ["chain_id"],
+        ["id"],
+        ondelete="SET NULL",
+    )
+    op.create_index(
+        "idx_llm_call_chain_id",
+        "llm_call",
+        ["chain_id"],
+        postgresql_where=sa.text("chain_id IS NOT NULL"),
+    )
+
+    op.execute("ALTER TYPE jobtype ADD VALUE IF NOT EXISTS 'LLM_CHAIN'")
+
+
+def downgrade() -> None:
+    op.drop_index("idx_llm_call_chain_id", table_name="llm_call")
+    op.drop_constraint("fk_llm_call_chain_id", "llm_call", type_="foreignkey")
+    op.drop_column("llm_call", "chain_id")
+
+    op.drop_index("idx_llm_chain_job_id", table_name="llm_chain")
+    op.drop_table("llm_chain")

backend/app/api/docs/llm/llm_chain.md

@@ -0,0 +1,61 @@
+Execute a chain of LLM calls sequentially, where each block's output becomes the next block's input.
+
+This endpoint initiates an asynchronous LLM chain job. The request is queued
+for processing, and results are delivered via the callback URL when complete.
+
+### Key Parameters
+
+**`query`** (required) - Initial query input for the first block in the chain:
+- `input` (required): User question/prompt/query — accepts a plain string, a structured input object (`text`, `audio`, `image`, `pdf`), or a list of structured inputs
+- `conversation` (optional, object): Conversation configuration
+  - `id` (optional, string): Existing conversation ID to continue
+  - `auto_create` (optional, boolean, default false): Create new conversation if no ID provided
+  - **Note**: Cannot specify both `id` and `auto_create=true`
+
+
+**`blocks`** (required, array, min 1 block) - Ordered list of blocks to execute sequentially. Each block contains:
+
+- `config` (required) - Configuration for this block's LLM call (just choose one mode):
+
+  - **Mode 1: Stored Configuration**
+    - `id` (UUID): Configuration ID
+    - `version` (integer >= 1): Version number
+    - **Both required together**
+    - **Note**: When using stored configuration, do not include the `blob` field in the request body
+
+  - **Mode 2: Ad-hoc Configuration**
+    - `blob` (object): Complete configuration object
+      - `completion` (required, object): Completion configuration
+        - `provider` (required, string): Kaapi providers (`openai`, `google`, `sarvamai`) — params are validated and mapped internally. Native providers (`openai-native`, `google-native`, `sarvamai-native`) — params are passed through as-is
+        - `type` (required, string): Completion type — `"text"`, `"stt"`, or `"tts"`
+        - `params` (required, object): Parameters structure depends on provider and type (see schema for detailed structure)
+      - `input_guardrails` (optional, array): Guardrails applied to validate/sanitize input before the LLM call
+      - `output_guardrails` (optional, array): Guardrails applied to validate/sanitize output after the LLM call
+      - `prompt_template` (optional, object): Template for text interpolation
+        - `template` (required, string): Template string with `{{input}}` placeholder — replaced with the block's input before execution
+    - **Note**
+      - When using ad-hoc configuration, do not include `id` and `version` fields
+      - When using the Kaapi abstraction, parameters that are not supported by the selected provider or model are automatically suppressed. If any parameters are ignored, a list of warnings is included in the metadata.warnings.
+    - **Recommendation**: Use stored configs (Mode 1) for production; use ad-hoc configs only for testing/validation
+    - **Schema**: Check the API schema or examples below for the complete parameter structure for each provider type
+
+- `include_provider_raw_response` (optional, boolean, default false):
+  - When true, includes the unmodified raw response from the LLM provider for this block
+
+- `intermediate_callback` (optional, boolean, default false):
+  - When true, sends an intermediate callback after this block completes with the block's response, usage, and position in the chain
+
+**`callback_url`** (optional, HTTPS URL):
+- Webhook endpoint to receive the final response and intermediate callbacks
+- Must be a valid HTTPS URL
+- If not provided, response is only accessible through job status
+
+**`request_metadata`** (optional, object):
+- Custom JSON metadata
+- Passed through unchanged in the response
+
+### Note
+- If any block fails, the chain stops immediately — no subsequent blocks are executed
+- `warnings` list is automatically added in response metadata when using Kaapi configs if any parameters are suppressed or adjusted (e.g., temperature on reasoning models)
+
+---

backend/app/api/main.py

@@ -10,6 +10,7 @@
     login,
     languages,
     llm,
+    llm_chain,
     organization,
     openai_conversation,
     project,
@@ -41,6 +42,7 @@
 api_router.include_router(evaluations.router)
 api_router.include_router(languages.router)
 api_router.include_router(llm.router)
+api_router.include_router(llm_chain.router)
 api_router.include_router(login.router)
 api_router.include_router(onboarding.router)
 api_router.include_router(openai_conversation.router)

backend/app/api/routes/llm_chain.py

@@ -0,0 +1,62 @@
+import logging
+
+from fastapi import APIRouter, Depends
+from app.api.deps import AuthContextDep, SessionDep
+from app.api.permissions import Permission, require_permission
+from app.models import LLMChainRequest, LLMChainResponse, Message
+from app.services.llm.jobs import start_chain_job
+from app.utils import APIResponse, validate_callback_url, load_description
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(tags=["LLM"])
+llm_callback_router = APIRouter()
+
+
+@llm_callback_router.post(
+    "{$callback_url}",
+    name="llm_chain_callback",
+)
+def llm_callback_notification(body: APIResponse[LLMChainResponse]):
+    """
+    Callback endpoint specification for LLM chain completion.
+
+    The callback will receive:
+    - On success: APIResponse with success=True and data containing LLMChainResponse
+    - On failure: APIResponse with success=False and error message
+    - metadata field will always be included if provided in the request
+    """
+    ...
+
+
+@router.post(
+    "/llm/chain",
+    description=load_description("llm/llm_chain.md"),
+    response_model=APIResponse[Message],
+    callbacks=llm_callback_router.routes,
+    dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))],
+)
+def llm_chain(
+    _current_user: AuthContextDep, _session: SessionDep, request: LLMChainRequest
+):
+    """
+    Endpoint to initiate an LLM chain as a background job.
+    """
+    project_id = _current_user.project_.id
+    organization_id = _current_user.organization_.id
+
+    if request.callback_url:
+        validate_callback_url(str(request.callback_url))
+
+    start_chain_job(
+        db=_session,
+        request=request,
+        project_id=project_id,
+        organization_id=organization_id,
+    )
+
+    return APIResponse.success_response(
+        data=Message(
+            message="Your response is being generated and will be delivered via callback."
+        ),
+    )

backend/app/crud/llm.py

@@ -48,6 +48,7 @@ def create_llm_call(
     *,
     request: LLMCallRequest,
     job_id: UUID,
+    chain_id: UUID | None = None,
     project_id: int,
     organization_id: int,
     resolved_config: ConfigBlob,
@@ -128,6 +129,7 @@ def create_llm_call(
         job_id=job_id,
         project_id=project_id,
         organization_id=organization_id,
+        chain_id=chain_id,
         input=serialize_input(request.query.input),
         input_type=input_type,
         output_type=output_type,