Weaviate Cookbooks

from weaviate import connect_to_weaviate_cloud
import os

def retrieve_data(query: str):
    """
    Given a query (free text), return the most relevant documents from the vector database using hybrid search.
    """
    client = connect_to_weaviate_cloud(
        cluster_url=os.getenv("WEAVIATE_URL", ""),
        auth_credentials=os.getenv("WEAVIATE_API_KEY", ""),
    )
    collection = client.collections.use("<collection_name>")
    response = collection.query.hybrid(query=query, limit=5)
    client.close()
    return f"{[obj.properties for obj in response.objects]}"

from pydantic import BaseModel, Field
from typing import Literal, Any

class SearchFilter(BaseModel):
    field: str = Field(description="The field to be filtered on.")
    operator: Literal["=", "!=", ">", "<"] = Field(description="The operator to be used in conjunction with the value.")
    value: Any = Field(description="The value to be used in conjunction with the operator.")

class Search(BaseModel):
    query: str = Field(description="The search query to be used in the vector database.")
    filters: list[SearchFilter] = Field(description="The filters to be used in the vector database.")
    limit: int = Field(description="The number of results to return from the vector database.")

class SearchCreation(dspy.Signature):
    """
    Create a search query for a vector database.
    """
    user_prompt: str = dspy.InputField()
    schema: list[dict] = dspy.InputField(desc="Schema of the collection to be searched.")
    search: Search = dspy.OutputField(
        desc=(
            "Your search query and filters, this should be a valid JSON object. "
            "This should be constructed so that it matches the goal of the user prompt."
        )
    )

from weaviate.classes.query import Filter

def format_filters(search_filters: list[SearchFilter]):
    filters = []
    for search_filter in search_filters:
        base_filter = Filter.by_property(search_filter.field)
        if search_filter.operator == "=":
            filters.append(base_filter.equal(search_filter.value))
        elif search_filter.operator == "!=":
            filters.append(base_filter.not_equal(search_filter.value))
        elif search_filter.operator == ">":
            filters.append(base_filter.greater_than(search_filter.value))
        elif search_filter.operator == "<":
            filters.append(base_filter.less_than(search_filter.value))
    return Filter.all_of(filters) if filters else None

def query_agent_tool(collection_name: str, user_prompt: str):
    """
    Given a query (free text), return the most relevant documents from the vector database using hybrid search with LLM-generated filters.
    """
    client = connect_to_weaviate_cloud(
        cluster_url=os.getenv("WEAVIATE_URL", ""),
        auth_credentials=os.getenv("WEAVIATE_API_KEY", ""),
    )
    collection = client.collections.use(collection_name)
    config = collection.config.get()
    schema = [{"name": p.name, "type": p.data_type[:]} for p in config.properties]

    query_model = dspy.ChainOfThought(SearchCreation)
    query_output = query_model(
        user_prompt=user_prompt,
        schema=schema,
        lm=dspy.LM("<subtask_model_name>")
    )

    response = collection.query.hybrid(
        query=query_output.search.query,
        filters=format_filters(query_output.search.filters),
        limit=query_output.search.limit
    )
    client.close()
    return f"{[obj.properties for obj in response.objects]}"

class MemoryCreation(dspy.Signature):
    user_prompt: str = dspy.InputField()
    assistant_response: str = dspy.InputField()
    memory: str = dspy.OutputField(
        description="A single string representing the most pertinent fact from the user/agent interaction."
    )

class AgentResponse(dspy.Signature):

    # Input Fields
    history: dspy.History = dspy.InputField()
    user_prompt: str = dspy.InputField()
    available_tools: str = dspy.InputField()
    memories: list[str] = dspy.InputField(
        desc="A list of memories from previous conversations, you can use these to inform your response."
    )

    # Output Fields
    response: str = dspy.OutputField(
        description="The response to the user's prompt whilst the tool is running. Update the user on the progress of their request (if a tool is picked), or the final response to the user (if no tool is picked)."
    )
    tool: str | None = dspy.OutputField(
        description="The tool that needs to be used. Return None if no tool is needed."
    )
    tool_inputs: Dict[str, Any] | None = dspy.OutputField(
        description=(
            "The inputs for the tool. Return an empty dictionary (still include the field) if no inputs are needed. "
            "The key is the name of the input, the value is the value of the input."
        )
    )

from weaviate import connect_to_weaviate_cloud
from weaviate.classes.config import Configure

class RouterAgent:
    def __init__(self, model: str, memory_model: str | None = None, tools: List[Callable] = []):
        self.tools: list[Callable] = tools
        self.model = dspy.LM(model)
        self.memory_model = dspy.LM(memory_model) if memory_model else dspy.LM(model)
        self.agent = dspy.ChainOfThought(AgentResponse)
        self.memory_agent = dspy.Predict(MemoryCreation)
        self.conversation_history = dspy.History(messages=[])
        self.weaviate_client = connect_to_weaviate_cloud(
            cluster_url=os.getenv("WEAVIATE_URL", ""),
            auth_credentials=os.getenv("WEAVIATE_API_KEY", ""),
        )

    # ... existing methods from basic_agent.md (add_conversation_history, get_tools_and_descriptions) ...

    def create_memory(self, user_prompt: str, assistant_response: str, tool_result: str):
        if tool_result:
            assistant_response += "\n" + tool_result

        result = self.memory_agent(
            history=self.conversation_history,
            user_prompt=user_prompt,
            assistant_response=assistant_response,
            lm=self.memory_model,
        )
        if not self.weaviate_client.collections.exists("Agent_Memory"):
            self.weaviate_client.collections.create(
                "Agent_Memory",
                vector_config=Configure.Vectors.text2vec_weaviate()
            )

        collection = self.weaviate_client.collections.use("Agent_Memory")
        collection.data.insert({"user_prompt": user_prompt, "memory": result.memory})
        return result.memory

    def retrieve_memories(self, user_prompt: str):
        if not self.weaviate_client.collections.exists("Agent_Memory"):
            return []
        collection = self.weaviate_client.collections.use("Agent_Memory")
        query = collection.query.near_text(query=user_prompt, limit=5)
        return [memory.properties["memory"] for memory in query.objects]

from weaviate.agents.query import QueryAgent

# import client here

qa = QueryAgent(
    client=client, collections=["<collection_name>"]
)
response = qa.search("<user query here>")  # retrieval only
response = qa.ask("<user query here>")     # retrieval + text response via response.final_answer

import elysia
from elysia.tools.text import FakeTextResponse as TextResponseTool

elysia.configure(
    base_model="<model_name>",
    base_provider="<provider>",  # e.g. "anthropic", "openai"
    logging_level="ERROR"
)

tree = elysia.Tree("empty", use_elysia_collections=False)
tree.add_tool(TextResponseTool)

@elysia.tool
async def your_tool(param: str):
    """Tool description."""
    return {"result"}

tree.add_tool(your_tool)
response, _ = tree("user query here")

from elysia import preprocess
preprocess("<collection_name>")

tree = elysia.Tree()
response, _ = tree(
    "user query here",
    collection_names=["<collection_name>"]
)