Agent & Knowledge¶

Higher-level building blocks for retrieval-augmented agents.

Agent¶

Agent ¶

Agent(model: str = 'gpt-4o-mini', system_prompt: str | None = None, redis_url: str = 'redis://localhost:6379', redis_client: Redis | None = None, config: AgentConfig | None = None)

Framework-agnostic agent with LiteLLM backend.

Example

agent = Agent( model="gpt-4o-mini", # Or "anthropic/claude-3-5-sonnet" redis_url="redis://localhost:6379", )

Ingest knowledge¶

await agent.knowledge.ingest("Redis uses RDB for persistence...")

Add custom tools¶

agent.tools.add_provider(MyProvider())

Chat¶

response = await agent.chat("How does Redis persistence work?")

Initialize agent.

Parameters:

Name	Type	Description	Default
`model`	`str`	LiteLLM model name (e.g., "gpt-4o-mini", "anthropic/claude-3-5-sonnet").	`'gpt-4o-mini'`
`system_prompt`	`str \| None`	Optional system prompt.	`None`
`redis_url`	`str`	Redis connection URL.	`'redis://localhost:6379'`
`redis_client`	`Redis \| None`	Optional existing Redis client.	`None`
`config`	`AgentConfig \| None`	Optional full configuration.	`None`

Source code in redis_agent_kit/agent.py

def __init__(
    self,
    model: str = "gpt-4o-mini",
    system_prompt: str | None = None,
    redis_url: str = "redis://localhost:6379",
    redis_client: redis.Redis | None = None,
    config: AgentConfig | None = None,
):
    """Initialize agent.

    Args:
        model: LiteLLM model name (e.g., "gpt-4o-mini", "anthropic/claude-3-5-sonnet").
        system_prompt: Optional system prompt.
        redis_url: Redis connection URL.
        redis_client: Optional existing Redis client.
        config: Optional full configuration.
    """
    self._config = config or AgentConfig(model=model, system_prompt=system_prompt)
    self._client = redis_client or redis.from_url(redis_url, decode_responses=True)  # type: ignore[no-untyped-call]

    # Core components
    self.knowledge = KnowledgeStore(self._client)
    self.tools = ToolManager()

    # Register built-in knowledge tool
    self._knowledge_provider = KnowledgeProvider(self.knowledge)
    self.tools.add_provider(self._knowledge_provider)

    # Conversation history
    self._history: list[Message] = []

model `property` ¶

model: str

Get the model name.

config `property` ¶

config: AgentConfig

Get agent configuration.

add_provider ¶

add_provider(provider: ToolProvider) -> None

Add a tool provider.

Parameters:

Name	Type	Description	Default
`provider`	`ToolProvider`	ToolProvider instance.	required

Source code in redis_agent_kit/agent.py

def add_provider(self, provider: ToolProvider) -> None:
    """Add a tool provider.

    Args:
        provider: ToolProvider instance.
    """
    self.tools.add_provider(provider)

clear_history ¶

clear_history() -> None

Clear conversation history.

Source code in redis_agent_kit/agent.py

def clear_history(self) -> None:
    """Clear conversation history."""
    self._history.clear()

chat `async` ¶

chat(message: str, use_knowledge: bool | None = None, use_tools: bool | None = None) -> str

Send a message and get a response.

Parameters:

Name	Type	Description	Default
`message`	`str`	User message.	required
`use_knowledge`	`bool \| None`	Override config to enable/disable knowledge context.	`None`
`use_tools`	`bool \| None`	Override config to enable/disable tools.	`None`

Returns:

Type	Description
`str`	Assistant response.

Source code in redis_agent_kit/agent.py

async def chat(
    self,
    message: str,
    use_knowledge: bool | None = None,
    use_tools: bool | None = None,
) -> str:
    """Send a message and get a response.

    Args:
        message: User message.
        use_knowledge: Override config to enable/disable knowledge context.
        use_tools: Override config to enable/disable tools.

    Returns:
        Assistant response.
    """
    use_knowledge = use_knowledge if use_knowledge is not None else self._config.use_knowledge
    use_tools = use_tools if use_tools is not None else self._config.use_tools

    messages = self._build_messages(message, use_knowledge=use_knowledge)

    # Get tools
    tools = None
    if use_tools and self.tools._tools:
        tools = self.tools.get_openai_tools()

    # Call LLM
    response = await self._call_llm(messages, tools)

    # Handle tool calls
    iterations = 0
    while (
        response.choices[0].message.tool_calls and iterations < self._config.max_tool_iterations
    ):
        response = await self._handle_tool_calls(response, messages)
        iterations += 1

    # Extract response
    assistant_content = response.choices[0].message.content or ""

    # Store in history
    self._history.append(Message(role="user", content=message))
    self._history.append(Message(role="assistant", content=assistant_content))

    return assistant_content

AgentConfig¶

AgentConfig `dataclass` ¶

AgentConfig(model: str = 'gpt-4o-mini', system_prompt: str | None = None, temperature: float = 0.7, max_tokens: int | None = None, max_tool_iterations: int = 10, use_knowledge: bool = True, use_tools: bool = True, knowledge_context_limit: int = 5)

Configuration for Agent.

KnowledgeStore¶

KnowledgeStore ¶

KnowledgeStore(client: Redis, vectorizer: Vectorizer | None = None, prefix: str = 'rak', embedding_model: str = 'text-embedding-3-small', dimensions: int | None = None)

High-level knowledge store combining vectorization and search.

Example

from redis_agent_kit import KnowledgeStore

ks = KnowledgeStore.from_url("redis://localhost:6379")

Ingest content¶

await ks.ingest("Redis uses RDB for persistence...", title="Redis Persistence")

Search¶

results = await ks.search("How does Redis save data?") for r in results: print(f"{r.content[:100]}... (score: {r.score:.2f})")

Initialize knowledge store.

Parameters:

Name	Type	Description	Default
`client`	`Redis`	Async Redis client.	required
`vectorizer`	`Vectorizer \| None`	Custom vectorizer (created with model if None).	`None`
`prefix`	`str`	Key prefix for storage.	`'rak'`
`embedding_model`	`str`	Model name for embeddings (used if no vectorizer).	`'text-embedding-3-small'`
`dimensions`	`int \| None`	Vector dimensions (auto-detected if None).	`None`

Source code in redis_agent_kit/knowledge.py

def __init__(
    self,
    client: redis.Redis,
    vectorizer: Vectorizer | None = None,
    prefix: str = "rak",
    embedding_model: str = "text-embedding-3-small",
    dimensions: int | None = None,
):
    """Initialize knowledge store.

    Args:
        client: Async Redis client.
        vectorizer: Custom vectorizer (created with model if None).
        prefix: Key prefix for storage.
        embedding_model: Model name for embeddings (used if no vectorizer).
        dimensions: Vector dimensions (auto-detected if None).
    """
    self._client = client
    self._vectorizer = vectorizer or Vectorizer(model=embedding_model)
    self._vector_store = VectorStore(client, prefix=prefix, dimensions=dimensions)
    self._prefix = prefix

from_url `classmethod` ¶

from_url(redis_url: str = 'redis://localhost:6379', embedding_model: str = 'text-embedding-3-small', prefix: str = 'rak', **kwargs: Any) -> KnowledgeStore

Create KnowledgeStore from Redis URL.

Parameters:

Name	Type	Description	Default
`redis_url`	`str`	Redis connection URL.	`'redis://localhost:6379'`
`embedding_model`	`str`	Model for generating embeddings.	`'text-embedding-3-small'`
`prefix`	`str`	Key prefix for storage.	`'rak'`
`**kwargs`	`Any`	Additional arguments for KnowledgeStore.	`{}`

Returns:

Type	Description
`KnowledgeStore`	Configured KnowledgeStore instance.

Source code in redis_agent_kit/knowledge.py

@classmethod
def from_url(
    cls,
    redis_url: str = "redis://localhost:6379",
    embedding_model: str = "text-embedding-3-small",
    prefix: str = "rak",
    **kwargs: Any,
) -> KnowledgeStore:
    """Create KnowledgeStore from Redis URL.

    Args:
        redis_url: Redis connection URL.
        embedding_model: Model for generating embeddings.
        prefix: Key prefix for storage.
        **kwargs: Additional arguments for KnowledgeStore.

    Returns:
        Configured KnowledgeStore instance.
    """
    client: Any = redis.from_url(  # type: ignore[no-untyped-call]
        redis_url, decode_responses=True
    )
    return cls(client, embedding_model=embedding_model, prefix=prefix, **kwargs)

search `async` ¶

search(query: str, limit: int = 10, distance_threshold: float | None = None) -> list[SearchResult]

Search for relevant content using semantic similarity.

Parameters:

Name	Type	Description	Default
`query`	`str`	Search query text.	required
`limit`	`int`	Maximum number of results.	`10`
`distance_threshold`	`float \| None`	Optional max distance filter.	`None`

Returns:

Type	Description
`list[SearchResult]`	List of SearchResult objects sorted by relevance.

Source code in redis_agent_kit/knowledge.py

async def search(
    self,
    query: str,
    limit: int = 10,
    distance_threshold: float | None = None,
) -> list[SearchResult]:
    """Search for relevant content using semantic similarity.

    Args:
        query: Search query text.
        limit: Maximum number of results.
        distance_threshold: Optional max distance filter.

    Returns:
        List of SearchResult objects sorted by relevance.
    """
    embedding = await self._vectorizer.embed(query)
    return await self._vector_store.search(
        embedding, limit=limit, distance_threshold=distance_threshold
    )

ingest `async` ¶

ingest(content: str, title: str | None = None, doc_id: str | None = None, metadata: dict[str, Any] | None = None) -> str

Ingest a single piece of content.

Parameters:

Name	Type	Description	Default
`content`	`str`	The text content to ingest.	required
`title`	`str \| None`	Optional title for the content.	`None`
`doc_id`	`str \| None`	Optional document ID (auto-generated if None).	`None`
`metadata`	`dict[str, Any] \| None`	Optional metadata dict.	`None`

Returns:

Type	Description
`str`	The document ID.

Source code in redis_agent_kit/knowledge.py

async def ingest(
    self,
    content: str,
    title: str | None = None,
    doc_id: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> str:
    """Ingest a single piece of content.

    Args:
        content: The text content to ingest.
        title: Optional title for the content.
        doc_id: Optional document ID (auto-generated if None).
        metadata: Optional metadata dict.

    Returns:
        The document ID.
    """
    # Generate doc_id from content hash if not provided
    if doc_id is None:
        doc_id = hashlib.sha256(content.encode()).hexdigest()[:16]

    # Generate embedding
    embedding = await self._vectorizer.embed(content)

    # Prepare metadata
    meta = metadata or {}
    if title:
        meta["title"] = title

    # Store
    await self._vector_store.store_chunk(
        chunk_id=f"{doc_id}_0",  # Single chunk
        doc_id=doc_id,
        content=content,
        embedding=embedding,
        metadata=meta,
    )

    return doc_id

ingest_many `async` ¶

ingest_many(documents: list[tuple[str, str | None, dict[str, Any] | None]]) -> list[str]

Ingest multiple documents.

Parameters:

Name	Type	Description	Default
`documents`	`list[tuple[str, str \| None, dict[str, Any] \| None]]`	List of (content, title, metadata) tuples.	required

Returns:

Type	Description
`list[str]`	List of document IDs.

Source code in redis_agent_kit/knowledge.py

async def ingest_many(
    self,
    documents: list[tuple[str, str | None, dict[str, Any] | None]],
) -> list[str]:
    """Ingest multiple documents.

    Args:
        documents: List of (content, title, metadata) tuples.

    Returns:
        List of document IDs.
    """
    doc_ids = []
    for content, title, metadata in documents:
        doc_id = await self.ingest(content, title=title, metadata=metadata)
        doc_ids.append(doc_id)
    return doc_ids

delete `async` ¶

delete(doc_id: str) -> int

Delete a document from the knowledge base.

Parameters:

Name	Type	Description	Default
`doc_id`	`str`	Document ID to delete.	required

Returns:

Type	Description
`int`	Number of chunks deleted.

Source code in redis_agent_kit/knowledge.py

async def delete(self, doc_id: str) -> int:
    """Delete a document from the knowledge base.

    Args:
        doc_id: Document ID to delete.

    Returns:
        Number of chunks deleted.
    """
    return await self._vector_store.delete_by_doc_id(doc_id)

count `async` ¶

count() -> int

Count total documents/chunks in the knowledge base.

Source code in redis_agent_kit/knowledge.py

async def count(self) -> int:
    """Count total documents/chunks in the knowledge base."""
    return await self._vector_store.count()

clear `async` ¶

clear() -> None

Clear all content from the knowledge base.

Source code in redis_agent_kit/knowledge.py

async def clear(self) -> None:
    """Clear all content from the knowledge base."""
    await self._vector_store.clear()

Vectorizer¶

Vectorizer ¶

Vectorizer(model: str = 'text-embedding-3-small', vectorizer: BaseVectorizer | None = None, api_config: dict[str, Any] | None = None, cache: EmbeddingsCache | None = None)

Wrapper around RedisVL vectorizers with a simplified interface.

By default uses OpenAI's text-embedding-3-small model. Can be configured to use any RedisVL vectorizer or a custom LiteLLM-based vectorizer.

Examples:

Default OpenAI vectorizer¶

vectorizer = Vectorizer()

Custom model¶

vectorizer = Vectorizer(model="text-embedding-ada-002")

Use LiteLLM for broader provider support¶

vectorizer = Vectorizer.from_litellm(model="cohere/embed-english-v3.0")

Use any RedisVL vectorizer directly¶

from redisvl.utils.vectorize import CohereTextVectorizer vectorizer = Vectorizer(vectorizer=CohereTextVectorizer())

Initialize vectorizer.

Parameters:

Name	Type	Description	Default
`model`	`str`	OpenAI embedding model name (ignored if vectorizer provided).	`'text-embedding-3-small'`
`vectorizer`	`BaseVectorizer \| None`	Optional RedisVL vectorizer instance to use directly.	`None`
`api_config`	`dict[str, Any] \| None`	Optional API configuration for the vectorizer.	`None`
`cache`	`EmbeddingsCache \| None`	Optional RedisVL EmbeddingsCache for caching.	`None`

Source code in redis_agent_kit/vectorizer.py

def __init__(
    self,
    model: str = "text-embedding-3-small",
    vectorizer: BaseVectorizer | None = None,
    api_config: dict[str, Any] | None = None,
    cache: EmbeddingsCache | None = None,
):
    """Initialize vectorizer.

    Args:
        model: OpenAI embedding model name (ignored if vectorizer provided).
        vectorizer: Optional RedisVL vectorizer instance to use directly.
        api_config: Optional API configuration for the vectorizer.
        cache: Optional RedisVL EmbeddingsCache for caching.
    """
    self._model = model
    if vectorizer is not None:
        self._vectorizer = vectorizer
    else:
        self._vectorizer = OpenAITextVectorizer(
            model=model,
            api_config=api_config,
            cache=cache,
        )

model `property` ¶

model: str

Get the model name.

dims `property` ¶

dims: int | None

Get embedding dimensions from underlying vectorizer.

from_litellm `classmethod` ¶

from_litellm(model: str = 'text-embedding-3-small', dimensions: int | None = None, cache: EmbeddingsCache | None = None) -> Vectorizer

Create a vectorizer using LiteLLM for broader provider support.

LiteLLM supports many providers beyond what RedisVL offers natively: - OpenAI: text-embedding-3-small, text-embedding-ada-002 - Cohere: cohere/embed-english-v3.0 - Azure: azure/text-embedding-ada-002 - Voyage: voyage/voyage-2 - And many more...

Parameters:

Name	Type	Description	Default
`model`	`str`	LiteLLM model name.	`'text-embedding-3-small'`
`dimensions`	`int \| None`	Output dimensions (if supported by model).	`None`
`cache`	`EmbeddingsCache \| None`	Optional RedisVL EmbeddingsCache.	`None`

Returns:

Type	Description
`Vectorizer`	Vectorizer instance using LiteLLM.

Source code in redis_agent_kit/vectorizer.py

@classmethod
def from_litellm(
    cls,
    model: str = "text-embedding-3-small",
    dimensions: int | None = None,
    cache: EmbeddingsCache | None = None,
) -> Vectorizer:
    """Create a vectorizer using LiteLLM for broader provider support.

    LiteLLM supports many providers beyond what RedisVL offers natively:
    - OpenAI: text-embedding-3-small, text-embedding-ada-002
    - Cohere: cohere/embed-english-v3.0
    - Azure: azure/text-embedding-ada-002
    - Voyage: voyage/voyage-2
    - And many more...

    Args:
        model: LiteLLM model name.
        dimensions: Output dimensions (if supported by model).
        cache: Optional RedisVL EmbeddingsCache.

    Returns:
        Vectorizer instance using LiteLLM.
    """
    adapter = LiteLLMVectorizerAdapter(model=model, dimensions=dimensions)
    vectorizer = CustomTextVectorizer(
        embed=adapter.embed,
        embed_many=adapter.embed_many,
        aembed=adapter.aembed,
        aembed_many=adapter.aembed_many,
        cache=cache,
    )
    return cls(vectorizer=vectorizer)

embed `async` ¶

embed(text: str) -> list[float]

Generate embedding for a single text (async).

Source code in redis_agent_kit/vectorizer.py

async def embed(self, text: str) -> list[float]:
    """Generate embedding for a single text (async)."""
    return cast(list[float], await self._vectorizer.aembed(text))

embed_sync ¶

embed_sync(text: str) -> list[float]

Generate embedding for a single text (synchronous).

Source code in redis_agent_kit/vectorizer.py

def embed_sync(self, text: str) -> list[float]:
    """Generate embedding for a single text (synchronous)."""
    return cast(list[float], self._vectorizer.embed(text))

embed_many `async` ¶

embed_many(texts: list[str]) -> list[list[float]]

Generate embeddings for multiple texts (async).

Source code in redis_agent_kit/vectorizer.py

async def embed_many(self, texts: list[str]) -> list[list[float]]:
    """Generate embeddings for multiple texts (async)."""
    if not texts:
        return []
    return cast(list[list[float]], await self._vectorizer.aembed_many(texts))

embed_many_sync ¶

embed_many_sync(texts: list[str]) -> list[list[float]]

Generate embeddings for multiple texts (synchronous).

Source code in redis_agent_kit/vectorizer.py

def embed_many_sync(self, texts: list[str]) -> list[list[float]]:
    """Generate embeddings for multiple texts (synchronous)."""
    if not texts:
        return []
    return cast(list[list[float]], self._vectorizer.embed_many(texts))

LiteLLMVectorizerAdapter¶

LiteLLMVectorizerAdapter ¶

LiteLLMVectorizerAdapter(model: str = 'text-embedding-3-small', dimensions: int | None = None, batch_size: int = 100)

Adapter to use LiteLLM with RedisVL's CustomTextVectorizer.

Provides sync and async embedding methods compatible with RedisVL.

Initialize adapter.

Parameters:

Name	Type	Description	Default
`model`	`str`	LiteLLM model name.	`'text-embedding-3-small'`
`dimensions`	`int \| None`	Output dimensions (if supported).	`None`
`batch_size`	`int`	Maximum texts per API call.	`100`

Source code in redis_agent_kit/vectorizer.py

def __init__(
    self,
    model: str = "text-embedding-3-small",
    dimensions: int | None = None,
    batch_size: int = 100,
):
    """Initialize adapter.

    Args:
        model: LiteLLM model name.
        dimensions: Output dimensions (if supported).
        batch_size: Maximum texts per API call.
    """
    self.model = model
    self.dimensions = dimensions
    self.batch_size = batch_size

embed ¶

embed(text: str) -> list[float]

Embed a single text synchronously.

Source code in redis_agent_kit/vectorizer.py

def embed(self, text: str) -> list[float]:
    """Embed a single text synchronously."""
    return self.embed_many([text])[0]

embed_many ¶

embed_many(texts: list[str]) -> list[list[float]]

Embed multiple texts synchronously.

Source code in redis_agent_kit/vectorizer.py

def embed_many(self, texts: list[str]) -> list[list[float]]:
    """Embed multiple texts synchronously."""
    if not texts:
        return []

    kwargs: dict[str, Any] = {"model": self.model, "input": texts}
    if self.dimensions is not None:
        kwargs["dimensions"] = self.dimensions

    response = litellm.embedding(**kwargs)
    return [item["embedding"] for item in response.data]

aembed `async` ¶

aembed(text: str) -> list[float]

Embed a single text asynchronously.

Source code in redis_agent_kit/vectorizer.py

async def aembed(self, text: str) -> list[float]:
    """Embed a single text asynchronously."""
    result = await self.aembed_many([text])
    return result[0]

aembed_many `async` ¶

aembed_many(texts: list[str]) -> list[list[float]]

Embed multiple texts asynchronously.

Source code in redis_agent_kit/vectorizer.py

async def aembed_many(self, texts: list[str]) -> list[list[float]]:
    """Embed multiple texts asynchronously."""
    if not texts:
        return []

    all_embeddings: list[list[float]] = []

    for i in range(0, len(texts), self.batch_size):
        batch = texts[i : i + self.batch_size]

        kwargs: dict[str, Any] = {"model": self.model, "input": batch}
        if self.dimensions is not None:
            kwargs["dimensions"] = self.dimensions

        response = await litellm.aembedding(**kwargs)
        all_embeddings.extend([item["embedding"] for item in response.data])

    return all_embeddings

Agent & Knowledge¶

Agent¶

Agent ¶

Ingest knowledge¶

Add custom tools¶

Chat¶

model property ¶

config property ¶

add_provider ¶

clear_history ¶

chat async ¶

AgentConfig¶

AgentConfig dataclass ¶

KnowledgeStore¶

KnowledgeStore ¶

Ingest content¶

Search¶

from_url classmethod ¶

search async ¶

ingest async ¶

ingest_many async ¶

delete async ¶

count async ¶

clear async ¶

Vectorizer¶

Vectorizer ¶

Default OpenAI vectorizer¶

Custom model¶

Use LiteLLM for broader provider support¶

Use any RedisVL vectorizer directly¶

model property ¶

dims property ¶

from_litellm classmethod ¶

embed async ¶

embed_sync ¶

embed_many async ¶

embed_many_sync ¶

LiteLLMVectorizerAdapter¶

LiteLLMVectorizerAdapter ¶

embed ¶

embed_many ¶

aembed async ¶

aembed_many async ¶

model `property` ¶

config `property` ¶

chat `async` ¶

AgentConfig `dataclass` ¶

from_url `classmethod` ¶

search `async` ¶

ingest `async` ¶

ingest_many `async` ¶

delete `async` ¶

count `async` ¶

clear `async` ¶

model `property` ¶

dims `property` ¶

from_litellm `classmethod` ¶

embed `async` ¶

embed_many `async` ¶

aembed `async` ¶

aembed_many `async` ¶