Extending Adapters

from langchat.adapters.llm._base_llm import BaseLLM, AIMessage
from typing import Any

def my_model_call(messages: list[Any]) -> str:
    """Call your model and return the response text."""
    # Extract the last user message
    last_message = messages[-1].content if hasattr(messages[-1], "content") else str(messages[-1])

    # Call your custom model here
    response = my_custom_api.generate(last_message)
    return response.text

# Wrap in BaseLLM
llm = BaseLLM(invoke_func=my_model_call)

# Use like any other provider
from langchat import LangChat
from langchat.providers import Pinecone, Supabase

lc = LangChat(
    llm=llm,
    vector_db=Pinecone("my-index"),
    db=Supabase(),
)

from openai import AzureOpenAI
from langchat.adapters.llm._base_llm import BaseLLM

azure_client = AzureOpenAI(
    api_key=os.environ["AZURE_OPENAI_KEY"],
    api_version="2024-02-01",
    azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
)

def azure_call(messages: list) -> str:
    formatted = [{"role": "user", "content": m.content} for m in messages]
    response = azure_client.chat.completions.create(
        model="gpt-4o",
        messages=formatted,
    )
    return response.choices[0].message.content

llm = BaseLLM(invoke_func=azure_call)

import os
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

class ChromaAdapter:
    """Custom Chroma vector database adapter."""

    def __init__(self, collection_name: str, persist_dir: str = "./chroma_db"):
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
        self.store = Chroma(
            collection_name=collection_name,
            embedding_function=self.embeddings,
            persist_directory=persist_dir,
        )

    def get_retriever(self, k: int = 5):
        return self.store.as_retriever(search_kwargs={"k": k})

# Use with LangChat
from langchat import LangChat
from langchat.providers import OpenAI, Supabase

lc = LangChat(
    llm=OpenAI("gpt-4o-mini"),
    vector_db=ChromaAdapter("my-collection"),
    db=Supabase(),
)

class MockDatabaseAdapter:
    """In-memory database adapter for testing."""

    def __init__(self):
        self._data = {}

    @property
    def client(self):
        return self  # return self as the "client"

    def table(self, name: str):
        return self

    def insert(self, data: dict):
        return self

    def select(self, cols: str = "*"):
        return self

    def eq(self, col: str, val):
        return self

    def execute(self):
        class Result:
            data = []
        return Result()