API Reference

vectara_agentic package.

Agent

Agent class for handling different types of agents and their interactions.

Source code in vectara_agentic/agent.py

class Agent:
    """
    Agent class for handling different types of agents and their interactions.
    """

    def __init__(
        self,
        tools: list[FunctionTool],
        topic: str = "general",
        custom_instructions: str = "",
        verbose: bool = True,
        update_func: Optional[Callable[[AgentStatusType, str], None]] = None,
        agent_progress_callback: Optional[Callable[[AgentStatusType, str], None]] = None,
        query_logging_callback: Optional[Callable[[str, str], None]] = None,
        agent_config: Optional[AgentConfig] = None,
        chat_history: Optional[list[Tuple[str, str]]] = None,
        validate_tools: bool = False,
    ) -> None:
        """
        Initialize the agent with the specified type, tools, topic, and system message.

        Args:

            tools (list[FunctionTool]): A list of tools to be used by the agent.
            topic (str, optional): The topic for the agent. Defaults to 'general'.
            custom_instructions (str, optional): Custom instructions for the agent. Defaults to ''.
            verbose (bool, optional): Whether the agent should print its steps. Defaults to True.
            agent_progress_callback (Callable): A callback function the code calls on any agent updates.
                update_func (Callable): old name for agent_progress_callback. Will be deprecated in future.
            query_logging_callback (Callable): A callback function the code calls upon completion of a query
            agent_config (AgentConfig, optional): The configuration of the agent.
                Defaults to AgentConfig(), which reads from environment variables.
            chat_history (Tuple[str, str], optional): A list of user/agent chat pairs to initialize the agent memory.
            validate_tools (bool, optional): Whether to validate tool inconsistency with instructions.
                Defaults to False.
        """
        self.agent_config = agent_config or AgentConfig()
        self.agent_type = self.agent_config.agent_type
        self.tools = tools
        if not any(tool.metadata.name == 'get_current_date' for tool in self.tools):
            self.tools += [ToolsFactory().create_tool(get_current_date)]
        self.llm = get_llm(LLMRole.MAIN, config=self.agent_config)
        self._custom_instructions = custom_instructions
        self._topic = topic
        self.agent_progress_callback = agent_progress_callback if agent_progress_callback else update_func
        self.query_logging_callback = query_logging_callback

        # Validate tools
        # Check for:
        # 1. multiple copies of the same tool
        # 2. Instructions for using tools that do not exist
        tool_names = [tool.metadata.name for tool in self.tools]
        duplicates = [tool for tool, count in Counter(tool_names).items() if count > 1]
        if duplicates:
            raise ValueError(f"Duplicate tools detected: {', '.join(duplicates)}")

        if validate_tools:
            prompt = f'''
            Given the following instructions, and a list of tool names,
            Please identify tools mentioned in the instructions that do not exist in the list.
            Instructions:
            {self._custom_instructions}
            Tool names: {', '.join(tool_names)}
            Your response should include a comma separated list of tool names that do not exist in the list.
            Your response should be an empty string if all tools mentioned in the instructions are in the list.
            '''
            llm = get_llm(LLMRole.MAIN, config=self.agent_config)
            bad_tools = llm.complete(prompt).text.split(", ")
            if bad_tools:
                raise ValueError(f"The Agent custom instructions mention these invalid tools: {', '.join(bad_tools)}")

        # Create token counters for the main and tool LLMs
        main_tok = get_tokenizer_for_model(role=LLMRole.MAIN)
        self.main_token_counter = TokenCountingHandler(tokenizer=main_tok) if main_tok else None
        tool_tok = get_tokenizer_for_model(role=LLMRole.TOOL)
        self.tool_token_counter = TokenCountingHandler(tokenizer=tool_tok) if tool_tok else None

        # Setup callback manager
        callbacks: list[BaseCallbackHandler] = [AgentCallbackHandler(self.agent_progress_callback)]
        if self.main_token_counter:
            callbacks.append(self.main_token_counter)
        if self.tool_token_counter:
            callbacks.append(self.tool_token_counter)
        callback_manager = CallbackManager(callbacks)  # type: ignore
        self.llm.callback_manager = callback_manager
        self.verbose = verbose

        if chat_history:
            msg_history = []
            for text_pairs in chat_history:
                msg_history.append(ChatMessage.from_str(content=text_pairs[0], role=MessageRole.USER))
                msg_history.append(ChatMessage.from_str(content=text_pairs[1], role=MessageRole.ASSISTANT))
            self.memory = ChatMemoryBuffer.from_defaults(token_limit=128000, chat_history=msg_history)
        else:
            self.memory = ChatMemoryBuffer.from_defaults(token_limit=128000)
        if self.agent_type == AgentType.REACT:
            prompt = _get_prompt(REACT_PROMPT_TEMPLATE, topic, custom_instructions)
            self.agent = ReActAgent.from_tools(
                tools=self.tools,
                llm=self.llm,
                memory=self.memory,
                verbose=verbose,
                react_chat_formatter=ReActChatFormatter(system_header=prompt),
                max_iterations=self.agent_config.max_reasoning_steps,
                callable_manager=callback_manager,
            )
        elif self.agent_type == AgentType.OPENAI:
            prompt = _get_prompt(GENERAL_PROMPT_TEMPLATE, topic, custom_instructions)
            self.agent = OpenAIAgent.from_tools(
                tools=self.tools,
                llm=self.llm,
                memory=self.memory,
                verbose=verbose,
                callable_manager=callback_manager,
                max_function_calls=self.agent_config.max_reasoning_steps,
                system_prompt=prompt,
            )
        elif self.agent_type == AgentType.LLMCOMPILER:
            agent_worker = LLMCompilerAgentWorker.from_tools(
                tools=self.tools,
                llm=self.llm,
                verbose=verbose,
                callable_manager=callback_manager,
            )
            agent_worker.system_prompt = _get_prompt(
                _get_llm_compiler_prompt(agent_worker.system_prompt, topic, custom_instructions),
                topic, custom_instructions
            )
            agent_worker.system_prompt_replan = _get_prompt(
                _get_llm_compiler_prompt(agent_worker.system_prompt_replan, topic, custom_instructions),
                topic, custom_instructions
            )
            self.agent = agent_worker.as_agent()
        elif self.agent_type == AgentType.LATS:
            agent_worker = LATSAgentWorker.from_tools(
                tools=self.tools,
                llm=self.llm,
                num_expansions=3,
                max_rollouts=-1,
                verbose=verbose,
                callable_manager=callback_manager,
            )
            prompt = _get_prompt(REACT_PROMPT_TEMPLATE, topic, custom_instructions)
            agent_worker.chat_formatter = ReActChatFormatter(system_header=prompt)
            self.agent = agent_worker.as_agent()
        else:
            raise ValueError(f"Unknown agent type: {self.agent_type}")

        try:
            self.observability_enabled = setup_observer(self.agent_config)
        except Exception as e:
            print(f"Failed to set up observer ({e}), ignoring")
            self.observability_enabled = False

    def clear_memory(self) -> None:
        """
        Clear the agent's memory.
        """
        self.agent.memory.reset()

    def __eq__(self, other):
        if not isinstance(other, Agent):
            print(f"Comparison failed: other is not an instance of Agent. (self: {type(self)}, other: {type(other)})")
            return False

        # Compare agent_type
        if self.agent_type != other.agent_type:
            print(
                f"Comparison failed: agent_type differs. (self.agent_type: {self.agent_type}, "
                f"other.agent_type: {other.agent_type})"
            )
            return False

        # Compare tools
        if self.tools != other.tools:
            print(
                "Comparison failed: tools differ."
                f"(self.tools: {[t.metadata.name for t in self.tools]}, "
                f"other.tools: {[t.metadata.name for t in other.tools]})")
            return False

        # Compare topic
        if self._topic != other._topic:
            print(f"Comparison failed: topic differs. (self.topic: {self._topic}, other.topic: {other._topic})")
            return False

        # Compare custom_instructions
        if self._custom_instructions != other._custom_instructions:
            print(
                "Comparison failed: custom_instructions differ. (self.custom_instructions: "
                f"{self._custom_instructions}, other.custom_instructions: {other._custom_instructions})"
            )
            return False

        # Compare verbose
        if self.verbose != other.verbose:
            print(f"Comparison failed: verbose differs. (self.verbose: {self.verbose}, other.verbose: {other.verbose})")
            return False

        # Compare agent
        if self.agent.memory.chat_store != other.agent.memory.chat_store:
            print(
                f"Comparison failed: agent memory differs. (self.agent: {repr(self.agent.memory.chat_store)}, "
                f"other.agent: {repr(other.agent.memory.chat_store)})"
            )
            return False

        # If all comparisons pass
        print("All comparisons passed. Objects are equal.")
        return True

    @classmethod
    def from_tools(
        cls,
        tools: List[FunctionTool],
        topic: str = "general",
        custom_instructions: str = "",
        verbose: bool = True,
        update_func: Optional[Callable[[AgentStatusType, str], None]] = None,
        agent_progress_callback: Optional[Callable[[AgentStatusType, str], None]] = None,
        query_logging_callback: Optional[Callable[[str, str], None]] = None,
        agent_config: AgentConfig = AgentConfig(),
        chat_history: Optional[list[Tuple[str, str]]] = None,
    ) -> "Agent":
        """
        Create an agent from tools, agent type, and language model.

        Args:

            tools (list[FunctionTool]): A list of tools to be used by the agent.
            topic (str, optional): The topic for the agent. Defaults to 'general'.
            custom_instructions (str, optional): custom instructions for the agent. Defaults to ''.
            verbose (bool, optional): Whether the agent should print its steps. Defaults to True.
            agent_progress_callback (Callable): A callback function the code calls on any agent updates.
                update_func (Callable): old name for agent_progress_callback. Will be deprecated in future.
            query_logging_callback (Callable): A callback function the code calls upon completion of a query
            agent_config (AgentConfig, optional): The configuration of the agent.
            chat_history (Tuple[str, str], optional): A list of user/agent chat pairs to initialize the agent memory.

        Returns:
            Agent: An instance of the Agent class.
        """
        return cls(
            tools=tools, topic=topic, custom_instructions=custom_instructions,
            verbose=verbose, agent_progress_callback=agent_progress_callback,
            query_logging_callback=query_logging_callback,
            update_func=update_func, agent_config=agent_config,
            chat_history=chat_history,
        )

    @classmethod
    def from_corpus(
        cls,
        tool_name: str,
        data_description: str,
        assistant_specialty: str,
        vectara_corpus_key: str = str(os.environ.get("VECTARA_CORPUS_KEY", "")),
        vectara_api_key: str = str(os.environ.get("VECTARA_API_KEY", "")),
        agent_progress_callback: Optional[Callable[[AgentStatusType, str], None]] = None,
        query_logging_callback: Optional[Callable[[str, str], None]] = None,
        verbose: bool = False,
        vectara_filter_fields: list[dict] = [],
        vectara_offset: int = 0,
        vectara_lambda_val: float = 0.005,
        vectara_semantics: str = "default",
        vectara_custom_dimensions: Dict = {},
        vectara_reranker: str = "slingshot",
        vectara_rerank_k: int = 50,
        vectara_rerank_limit: Optional[int] = None,
        vectara_rerank_cutoff: Optional[float] = None,
        vectara_diversity_bias: float = 0.2,
        vectara_udf_expression: str = None,
        vectara_rerank_chain: List[Dict] = None,
        vectara_n_sentences_before: int = 2,
        vectara_n_sentences_after: int = 2,
        vectara_summary_num_results: int = 10,
        vectara_summarizer: str = "vectara-summary-ext-24-05-med-omni",
        vectara_summary_response_language: str = "eng",
        vectara_summary_prompt_text: Optional[str] = None,
        vectara_max_response_chars: Optional[int] = None,
        vectara_max_tokens: Optional[int] = None,
        vectara_temperature: Optional[float] = None,
        vectara_frequency_penalty: Optional[float] = None,
        vectara_presence_penalty: Optional[float] = None,
        vectara_save_history: bool = True,
    ) -> "Agent":
        """
        Create an agent from a single Vectara corpus

        Args:
            tool_name (str): The name of Vectara tool used by the agent
            vectara_corpus_key (str): The Vectara corpus key (or comma separated list of keys).
            vectara_api_key (str): The Vectara API key.
            agent_progress_callback (Callable): A callback function the code calls on any agent updates.
            query_logging_callback (Callable): A callback function the code calls upon completion of a query
            data_description (str): The description of the data.
            assistant_specialty (str): The specialty of the assistant.
            verbose (bool, optional): Whether to print verbose output.
            vectara_filter_fields (List[dict], optional): The filterable attributes
                (each dict maps field name to Tuple[type, description]).
            vectara_offset (int, optional): Number of results to skip.
            vectara_lambda_val (float, optional): Lambda value for Vectara hybrid search.
            vectara_semantics: (str, optional): Indicates whether the query is intended as a query or response.
            vectara_custom_dimensions: (Dict, optional): Custom dimensions for the query.
            vectara_reranker (str, optional): The Vectara reranker name (default "slingshot")
            vectara_rerank_k (int, optional): The number of results to use with reranking.
            vectara_rerank_limit: (int, optional): The maximum number of results to return after reranking.
            vectara_rerank_cutoff: (float, optional): The minimum score threshold for results to include after
                reranking.
            vectara_diversity_bias (float, optional): The MMR diversity bias.
            vectara_udf_expression (str, optional): The user defined expression for reranking results.
            vectara_rerank_chain (List[Dict], optional): A list of Vectara rerankers to be applied sequentially.
            vectara_n_sentences_before (int, optional): The number of sentences before the matching text
            vectara_n_sentences_after (int, optional): The number of sentences after the matching text.
            vectara_summary_num_results (int, optional): The number of results to use in summarization.
            vectara_summarizer (str, optional): The Vectara summarizer name.
            vectara_summary_response_language (str, optional): The response language for the Vectara summary.
            vectara_summary_prompt_text (str, optional): The custom prompt, using appropriate prompt variables and
                functions.
            vectara_max_response_chars (int, optional): The desired maximum number of characters for the generated
                summary.
            vectara_max_tokens (int, optional): The maximum number of tokens to be returned by the LLM.
            vectara_temperature (float, optional): The sampling temperature; higher values lead to more randomness.
            vectara_frequency_penalty (float, optional): How much to penalize repeating tokens in the response,
                higher values reducing likelihood of repeating the same line.
            vectara_presence_penalty (float, optional): How much to penalize repeating tokens in the response,
                higher values increasing the diversity of topics.
            vectara_save_history (bool, optional): Whether to save the query in history.

        Returns:
            Agent: An instance of the Agent class.
        """
        vec_factory = VectaraToolFactory(
            vectara_api_key=vectara_api_key,
            vectara_corpus_key=vectara_corpus_key,
        )
        field_definitions = {}
        field_definitions["query"] = (str, Field(description="The user query"))  # type: ignore
        for field in vectara_filter_fields:
            field_definitions[field["name"]] = (
                eval(field["type"]),
                Field(description=field["description"]),
            )  # type: ignore
        query_args = create_model("QueryArgs", **field_definitions)  # type: ignore

        # tool name must be valid Python function name
        if tool_name:
            tool_name = re.sub(r"[^A-Za-z0-9_]", "_", tool_name)

        vectara_tool = vec_factory.create_rag_tool(
            tool_name=tool_name or f"vectara_{vectara_corpus_key}",
            tool_description=f"""
            Given a user query,
            returns a response (str) to a user question about {data_description}.
            """,
            tool_args_schema=query_args,
            reranker=vectara_reranker,
            rerank_k=vectara_rerank_k,
            rerank_limit=vectara_rerank_limit,
            rerank_cutoff=vectara_rerank_cutoff,
            mmr_diversity_bias=vectara_diversity_bias,
            udf_expression=vectara_udf_expression,
            rerank_chain=vectara_rerank_chain,
            n_sentences_before=vectara_n_sentences_before,
            n_sentences_after=vectara_n_sentences_after,
            offset=vectara_offset,
            lambda_val=vectara_lambda_val,
            semantics=vectara_semantics,
            custom_dimensions=vectara_custom_dimensions,
            summary_num_results=vectara_summary_num_results,
            vectara_summarizer=vectara_summarizer,
            summary_response_lang=vectara_summary_response_language,
            vectara_prompt_text=vectara_summary_prompt_text,
            max_response_chars=vectara_max_response_chars,
            max_tokens=vectara_max_tokens,
            temperature=vectara_temperature,
            frequency_penalty=vectara_frequency_penalty,
            presence_penalty=vectara_presence_penalty,
            save_history=vectara_save_history,
            include_citations=True,
            verbose=verbose,
        )

        assistant_instructions = f"""
        - You are a helpful {assistant_specialty} assistant.
        - You can answer questions about {data_description}.
        - Never discuss politics, and always respond politely.
        """

        return cls(
            tools=[vectara_tool],
            topic=assistant_specialty,
            custom_instructions=assistant_instructions,
            verbose=verbose,
            agent_progress_callback=agent_progress_callback,
            query_logging_callback=query_logging_callback,
        )

    def report(self) -> None:
        """
        Get a report from the agent.

        Returns:
            str: The report from the agent.
        """
        print("Vectara agentic Report:")
        print(f"Agent Type = {self.agent_type}")
        print(f"Topic = {self._topic}")
        print("Tools:")
        for tool in self.tools:
            if hasattr(tool, 'metadata'):
                print(f"- {tool.metadata.name}")
            else:
                print("- tool without metadata")
        print(f"Agent LLM = {get_llm(LLMRole.MAIN, config=self.agent_config).metadata.model_name}")
        print(f"Tool LLM = {get_llm(LLMRole.TOOL, config=self.agent_config).metadata.model_name}")

    def token_counts(self) -> dict:
        """
        Get the token counts for the agent and tools.

        Returns:
            dict: The token counts for the agent and tools.
        """
        return {
            "main token count": self.main_token_counter.total_llm_token_count if self.main_token_counter else -1,
            "tool token count": self.tool_token_counter.total_llm_token_count if self.tool_token_counter else -1,
        }

    async def _aformat_for_lats(self, prompt, agent_response):
        llm_prompt = f"""
        Given the question '{prompt}', and agent response '{agent_response.response}',
        Please provide a well formatted final response to the query.
        final response:
        """
        agent_response.response = str(self.llm.acomplete(llm_prompt))

    def chat(self, prompt: str) -> AgentResponse:           # type: ignore
        """
        Interact with the agent using a chat prompt.

        Args:
            prompt (str): The chat prompt.

        Returns:
            AgentResponse: The response from the agent.
        """
        return asyncio.run(self.achat(prompt))

    @retry(
        retry_on_exception=_retry_if_exception,
        stop_max_attempt_number=3,
        wait_fixed=2000,
    )
    async def achat(self, prompt: str) -> AgentResponse:    # type: ignore
        """
        Interact with the agent using a chat prompt.

        Args:
            prompt (str): The chat prompt.

        Returns:
            AgentResponse: The response from the agent.
        """

        try:
            st = time.time()
            agent_response = await self.agent.achat(prompt)
            if self.agent_type == AgentType.LATS:
                await self._aformat_for_lats(prompt, agent_response)
            if self.verbose:
                print(f"Time taken: {time.time() - st}")
            if self.observability_enabled:
                eval_fcs()
            if self.query_logging_callback:
                self.query_logging_callback(prompt, agent_response.response)
            return agent_response
        except Exception as e:
            return AgentResponse(
                response = (
                    f"Vectara Agentic: encountered an exception ({e}) at ({traceback.format_exc()})"
                    ", and can't respond."
                )
            )

    def stream_chat(self, prompt: str) -> AgentStreamingResponse:    # type: ignore
        """
        Interact with the agent using a chat prompt with streaming.
        Args:
            prompt (str): The chat prompt.
        Returns:
            AgentStreamingResponse: The streaming response from the agent.
        """
        return asyncio.run(self.astream_chat(prompt))

    @retry(
        retry_on_exception=_retry_if_exception,
        stop_max_attempt_number=3,
        wait_fixed=2000,
    )
    async def astream_chat(self, prompt: str) -> AgentStreamingResponse:    # type: ignore
        """
        Interact with the agent using a chat prompt asynchronously with streaming.
        Args:
            prompt (str): The chat prompt.
        Returns:
            AgentStreamingResponse: The streaming response from the agent.
        """
        try:
            agent_response = await self.agent.astream_chat(prompt)
            original_async_response_gen = agent_response.async_response_gen

            # Wrap async_response_gen
            async def _stream_response_wrapper():
                async for token in original_async_response_gen():
                    yield token  # Yield async token to keep streaming behavior

                # After streaming completes, execute additional logic
                if self.agent_type == AgentType.LATS:
                    await self._aformat_for_lats(prompt, agent_response)
                if self.query_logging_callback:
                    self.query_logging_callback(prompt, agent_response.response)
                if self.observability_enabled:
                    eval_fcs()

            agent_response.async_response_gen = _stream_response_wrapper  # Override method
            return agent_response
        except Exception as e:
            raise ValueError(
                f"Vectara Agentic: encountered an exception ({e}) at ({traceback.format_exc()}), and can't respond."
            ) from e

    #
    # Serialization methods
    #
    def dumps(self) -> str:
        """Serialize the Agent instance to a JSON string."""
        return json.dumps(self.to_dict())

    @classmethod
    def loads(cls, data: str) -> "Agent":
        """Create an Agent instance from a JSON string."""
        return cls.from_dict(json.loads(data))

    def to_dict(self) -> Dict[str, Any]:
        """Serialize the Agent instance to a dictionary."""
        tool_info = []

        for tool in self.tools:
            # Serialize each tool's metadata, function, and dynamic model schema (QueryArgs)
            tool_dict = {
                "tool_type": tool.metadata.tool_type.value,
                "name": tool.metadata.name,
                "description": tool.metadata.description,
                "fn": pickle.dumps(tool.fn).decode("latin-1") if tool.fn else None,  # Serialize fn
                "async_fn": pickle.dumps(tool.async_fn).decode("latin-1")
                if tool.async_fn
                else None,  # Serialize async_fn
                "fn_schema": tool.metadata.fn_schema.model_json_schema()
                if hasattr(tool.metadata, "fn_schema")
                else None,  # Serialize schema if available
            }
            tool_info.append(tool_dict)

        return {
            "agent_type": self.agent_type.value,
            "memory": pickle.dumps(self.agent.memory).decode("latin-1"),
            "tools": tool_info,
            "topic": self._topic,
            "custom_instructions": self._custom_instructions,
            "verbose": self.verbose,
            "agent_config": self.agent_config.to_dict(),
        }

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "Agent":
        """Create an Agent instance from a dictionary."""
        agent_config = AgentConfig.from_dict(data["agent_config"])
        tools = []

        for tool_data in data["tools"]:
            # Recreate the dynamic model using the schema info
            if tool_data.get("fn_schema"):
                field_definitions = {}
                for field, values in tool_data["fn_schema"]["properties"].items():
                    # Instead of checking for 'type', use the helper:
                    field_type = get_field_type(values)
                    # If there's a default value, include it.
                    if "default" in values:
                        field_definitions[field] = (
                            field_type,
                            Field(description=values.get("description", ""), default=values["default"]),
                        )
                    else:
                        field_definitions[field] = (
                            field_type,
                            Field(description=values.get("description", "")),
                        )
                query_args_model = create_model("QueryArgs", **field_definitions)  # type: ignore
            else:
                query_args_model = create_model("QueryArgs")

            fn = pickle.loads(tool_data["fn"].encode("latin-1")) if tool_data["fn"] else None
            async_fn = pickle.loads(tool_data["async_fn"].encode("latin-1")) if tool_data["async_fn"] else None

            tool = VectaraTool.from_defaults(
                name=tool_data["name"],
                description=tool_data["description"],
                fn=fn,
                async_fn=async_fn,
                fn_schema=query_args_model,  # Re-assign the recreated dynamic model
                tool_type=ToolType(tool_data["tool_type"]),
            )
            tools.append(tool)

        agent = cls(
            tools=tools,
            agent_config=agent_config,
            topic=data["topic"],
            custom_instructions=data["custom_instructions"],
            verbose=data["verbose"],
        )
        memory = pickle.loads(data["memory"].encode("latin-1")) if data.get("memory") else None
        if memory:
            agent.agent.memory = memory
        return agent

__init__(tools, topic='general', custom_instructions='', verbose=True, update_func=None, agent_progress_callback=None, query_logging_callback=None, agent_config=None, chat_history=None, validate_tools=False)

Initialize the agent with the specified type, tools, topic, and system message.

Args:

tools (list[FunctionTool]): A list of tools to be used by the agent.
topic (str, optional): The topic for the agent. Defaults to 'general'.
custom_instructions (str, optional): Custom instructions for the agent. Defaults to ''.
verbose (bool, optional): Whether the agent should print its steps. Defaults to True.
agent_progress_callback (Callable): A callback function the code calls on any agent updates.
    update_func (Callable): old name for agent_progress_callback. Will be deprecated in future.
query_logging_callback (Callable): A callback function the code calls upon completion of a query
agent_config (AgentConfig, optional): The configuration of the agent.
    Defaults to AgentConfig(), which reads from environment variables.
chat_history (Tuple[str, str], optional): A list of user/agent chat pairs to initialize the agent memory.
validate_tools (bool, optional): Whether to validate tool inconsistency with instructions.
    Defaults to False.

Source code in vectara_agentic/agent.py

def __init__(
    self,
    tools: list[FunctionTool],
    topic: str = "general",
    custom_instructions: str = "",
    verbose: bool = True,
    update_func: Optional[Callable[[AgentStatusType, str], None]] = None,
    agent_progress_callback: Optional[Callable[[AgentStatusType, str], None]] = None,
    query_logging_callback: Optional[Callable[[str, str], None]] = None,
    agent_config: Optional[AgentConfig] = None,
    chat_history: Optional[list[Tuple[str, str]]] = None,
    validate_tools: bool = False,
) -> None:
    """
    Initialize the agent with the specified type, tools, topic, and system message.

    Args:

        tools (list[FunctionTool]): A list of tools to be used by the agent.
        topic (str, optional): The topic for the agent. Defaults to 'general'.
        custom_instructions (str, optional): Custom instructions for the agent. Defaults to ''.
        verbose (bool, optional): Whether the agent should print its steps. Defaults to True.
        agent_progress_callback (Callable): A callback function the code calls on any agent updates.
            update_func (Callable): old name for agent_progress_callback. Will be deprecated in future.
        query_logging_callback (Callable): A callback function the code calls upon completion of a query
        agent_config (AgentConfig, optional): The configuration of the agent.
            Defaults to AgentConfig(), which reads from environment variables.
        chat_history (Tuple[str, str], optional): A list of user/agent chat pairs to initialize the agent memory.
        validate_tools (bool, optional): Whether to validate tool inconsistency with instructions.
            Defaults to False.
    """
    self.agent_config = agent_config or AgentConfig()
    self.agent_type = self.agent_config.agent_type
    self.tools = tools
    if not any(tool.metadata.name == 'get_current_date' for tool in self.tools):
        self.tools += [ToolsFactory().create_tool(get_current_date)]
    self.llm = get_llm(LLMRole.MAIN, config=self.agent_config)
    self._custom_instructions = custom_instructions
    self._topic = topic
    self.agent_progress_callback = agent_progress_callback if agent_progress_callback else update_func
    self.query_logging_callback = query_logging_callback

    # Validate tools
    # Check for:
    # 1. multiple copies of the same tool
    # 2. Instructions for using tools that do not exist
    tool_names = [tool.metadata.name for tool in self.tools]
    duplicates = [tool for tool, count in Counter(tool_names).items() if count > 1]
    if duplicates:
        raise ValueError(f"Duplicate tools detected: {', '.join(duplicates)}")

    if validate_tools:
        prompt = f'''
        Given the following instructions, and a list of tool names,
        Please identify tools mentioned in the instructions that do not exist in the list.
        Instructions:
        {self._custom_instructions}
        Tool names: {', '.join(tool_names)}
        Your response should include a comma separated list of tool names that do not exist in the list.
        Your response should be an empty string if all tools mentioned in the instructions are in the list.
        '''
        llm = get_llm(LLMRole.MAIN, config=self.agent_config)
        bad_tools = llm.complete(prompt).text.split(", ")
        if bad_tools:
            raise ValueError(f"The Agent custom instructions mention these invalid tools: {', '.join(bad_tools)}")

    # Create token counters for the main and tool LLMs
    main_tok = get_tokenizer_for_model(role=LLMRole.MAIN)
    self.main_token_counter = TokenCountingHandler(tokenizer=main_tok) if main_tok else None
    tool_tok = get_tokenizer_for_model(role=LLMRole.TOOL)
    self.tool_token_counter = TokenCountingHandler(tokenizer=tool_tok) if tool_tok else None

    # Setup callback manager
    callbacks: list[BaseCallbackHandler] = [AgentCallbackHandler(self.agent_progress_callback)]
    if self.main_token_counter:
        callbacks.append(self.main_token_counter)
    if self.tool_token_counter:
        callbacks.append(self.tool_token_counter)
    callback_manager = CallbackManager(callbacks)  # type: ignore
    self.llm.callback_manager = callback_manager
    self.verbose = verbose

    if chat_history:
        msg_history = []
        for text_pairs in chat_history:
            msg_history.append(ChatMessage.from_str(content=text_pairs[0], role=MessageRole.USER))
            msg_history.append(ChatMessage.from_str(content=text_pairs[1], role=MessageRole.ASSISTANT))
        self.memory = ChatMemoryBuffer.from_defaults(token_limit=128000, chat_history=msg_history)
    else:
        self.memory = ChatMemoryBuffer.from_defaults(token_limit=128000)
    if self.agent_type == AgentType.REACT:
        prompt = _get_prompt(REACT_PROMPT_TEMPLATE, topic, custom_instructions)
        self.agent = ReActAgent.from_tools(
            tools=self.tools,
            llm=self.llm,
            memory=self.memory,
            verbose=verbose,
            react_chat_formatter=ReActChatFormatter(system_header=prompt),
            max_iterations=self.agent_config.max_reasoning_steps,
            callable_manager=callback_manager,
        )
    elif self.agent_type == AgentType.OPENAI:
        prompt = _get_prompt(GENERAL_PROMPT_TEMPLATE, topic, custom_instructions)
        self.agent = OpenAIAgent.from_tools(
            tools=self.tools,
            llm=self.llm,
            memory=self.memory,
            verbose=verbose,
            callable_manager=callback_manager,
            max_function_calls=self.agent_config.max_reasoning_steps,
            system_prompt=prompt,
        )
    elif self.agent_type == AgentType.LLMCOMPILER:
        agent_worker = LLMCompilerAgentWorker.from_tools(
            tools=self.tools,
            llm=self.llm,
            verbose=verbose,
            callable_manager=callback_manager,
        )
        agent_worker.system_prompt = _get_prompt(
            _get_llm_compiler_prompt(agent_worker.system_prompt, topic, custom_instructions),
            topic, custom_instructions
        )
        agent_worker.system_prompt_replan = _get_prompt(
            _get_llm_compiler_prompt(agent_worker.system_prompt_replan, topic, custom_instructions),
            topic, custom_instructions
        )
        self.agent = agent_worker.as_agent()
    elif self.agent_type == AgentType.LATS:
        agent_worker = LATSAgentWorker.from_tools(
            tools=self.tools,
            llm=self.llm,
            num_expansions=3,
            max_rollouts=-1,
            verbose=verbose,
            callable_manager=callback_manager,
        )
        prompt = _get_prompt(REACT_PROMPT_TEMPLATE, topic, custom_instructions)
        agent_worker.chat_formatter = ReActChatFormatter(system_header=prompt)
        self.agent = agent_worker.as_agent()
    else:
        raise ValueError(f"Unknown agent type: {self.agent_type}")

    try:
        self.observability_enabled = setup_observer(self.agent_config)
    except Exception as e:
        print(f"Failed to set up observer ({e}), ignoring")
        self.observability_enabled = False

achat(prompt) async

Interact with the agent using a chat prompt.

Parameters:

Name	Type	Description	Default
prompt	str	The chat prompt.	required

Returns:

Name	Type	Description
AgentResponse	AgentResponse	The response from the agent.

Source code in vectara_agentic/agent.py

@retry(
    retry_on_exception=_retry_if_exception,
    stop_max_attempt_number=3,
    wait_fixed=2000,
)
async def achat(self, prompt: str) -> AgentResponse:    # type: ignore
    """
    Interact with the agent using a chat prompt.

    Args:
        prompt (str): The chat prompt.

    Returns:
        AgentResponse: The response from the agent.
    """

    try:
        st = time.time()
        agent_response = await self.agent.achat(prompt)
        if self.agent_type == AgentType.LATS:
            await self._aformat_for_lats(prompt, agent_response)
        if self.verbose:
            print(f"Time taken: {time.time() - st}")
        if self.observability_enabled:
            eval_fcs()
        if self.query_logging_callback:
            self.query_logging_callback(prompt, agent_response.response)
        return agent_response
    except Exception as e:
        return AgentResponse(
            response = (
                f"Vectara Agentic: encountered an exception ({e}) at ({traceback.format_exc()})"
                ", and can't respond."
            )
        )

astream_chat(prompt) async

Interact with the agent using a chat prompt asynchronously with streaming. Args: prompt (str): The chat prompt. Returns: AgentStreamingResponse: The streaming response from the agent.

Source code in vectara_agentic/agent.py

@retry(
    retry_on_exception=_retry_if_exception,
    stop_max_attempt_number=3,
    wait_fixed=2000,
)
async def astream_chat(self, prompt: str) -> AgentStreamingResponse:    # type: ignore
    """
    Interact with the agent using a chat prompt asynchronously with streaming.
    Args:
        prompt (str): The chat prompt.
    Returns:
        AgentStreamingResponse: The streaming response from the agent.
    """
    try:
        agent_response = await self.agent.astream_chat(prompt)
        original_async_response_gen = agent_response.async_response_gen

        # Wrap async_response_gen
        async def _stream_response_wrapper():
            async for token in original_async_response_gen():
                yield token  # Yield async token to keep streaming behavior

            # After streaming completes, execute additional logic
            if self.agent_type == AgentType.LATS:
                await self._aformat_for_lats(prompt, agent_response)
            if self.query_logging_callback:
                self.query_logging_callback(prompt, agent_response.response)
            if self.observability_enabled:
                eval_fcs()

        agent_response.async_response_gen = _stream_response_wrapper  # Override method
        return agent_response
    except Exception as e:
        raise ValueError(
            f"Vectara Agentic: encountered an exception ({e}) at ({traceback.format_exc()}), and can't respond."
        ) from e

chat(prompt)

Interact with the agent using a chat prompt.

Parameters:

Name	Type	Description	Default
prompt	str	The chat prompt.	required

Returns:

Name	Type	Description
AgentResponse	AgentResponse	The response from the agent.

Source code in vectara_agentic/agent.py

def chat(self, prompt: str) -> AgentResponse:           # type: ignore
    """
    Interact with the agent using a chat prompt.

    Args:
        prompt (str): The chat prompt.

    Returns:
        AgentResponse: The response from the agent.
    """
    return asyncio.run(self.achat(prompt))

clear_memory()

Clear the agent's memory.

Source code in vectara_agentic/agent.py

def clear_memory(self) -> None:
    """
    Clear the agent's memory.
    """
    self.agent.memory.reset()

dumps()

Serialize the Agent instance to a JSON string.

Source code in vectara_agentic/agent.py

def dumps(self) -> str:
    """Serialize the Agent instance to a JSON string."""
    return json.dumps(self.to_dict())

from_corpus(tool_name, data_description, assistant_specialty, vectara_corpus_key=str(os.environ.get('VECTARA_CORPUS_KEY', '')), vectara_api_key=str(os.environ.get('VECTARA_API_KEY', '')), agent_progress_callback=None, query_logging_callback=None, verbose=False, vectara_filter_fields=[], vectara_offset=0, vectara_lambda_val=0.005, vectara_semantics='default', vectara_custom_dimensions={}, vectara_reranker='slingshot', vectara_rerank_k=50, vectara_rerank_limit=None, vectara_rerank_cutoff=None, vectara_diversity_bias=0.2, vectara_udf_expression=None, vectara_rerank_chain=None, vectara_n_sentences_before=2, vectara_n_sentences_after=2, vectara_summary_num_results=10, vectara_summarizer='vectara-summary-ext-24-05-med-omni', vectara_summary_response_language='eng', vectara_summary_prompt_text=None, vectara_max_response_chars=None, vectara_max_tokens=None, vectara_temperature=None, vectara_frequency_penalty=None, vectara_presence_penalty=None, vectara_save_history=True) classmethod

Create an agent from a single Vectara corpus

Parameters:

Name	Type	Description	Default
tool_name	str	The name of Vectara tool used by the agent	required
vectara_corpus_key	str	The Vectara corpus key (or comma separated list of keys).	str(get('VECTARA_CORPUS_KEY', ''))
vectara_api_key	str	The Vectara API key.	str(get('VECTARA_API_KEY', ''))
agent_progress_callback	Callable	A callback function the code calls on any agent updates.	None
query_logging_callback	Callable	A callback function the code calls upon completion of a query	None
data_description	str	The description of the data.	required
assistant_specialty	str	The specialty of the assistant.	required
verbose	bool	Whether to print verbose output.	False
vectara_filter_fields	List[dict]	The filterable attributes (each dict maps field name to Tuple[type, description]).	[]
vectara_offset	int	Number of results to skip.	0
vectara_lambda_val	float	Lambda value for Vectara hybrid search.	0.005
vectara_semantics	str	(str, optional): Indicates whether the query is intended as a query or response.	'default'
vectara_custom_dimensions	Dict	(Dict, optional): Custom dimensions for the query.	{}
vectara_reranker	str	The Vectara reranker name (default "slingshot")	'slingshot'
vectara_rerank_k	int	The number of results to use with reranking.	50
vectara_rerank_limit	Optional[int]	(int, optional): The maximum number of results to return after reranking.	None
vectara_rerank_cutoff	Optional[float]	(float, optional): The minimum score threshold for results to include after reranking.	None
vectara_diversity_bias	float	The MMR diversity bias.	0.2
vectara_udf_expression	str	The user defined expression for reranking results.	None
vectara_rerank_chain	List[Dict]	A list of Vectara rerankers to be applied sequentially.	None
vectara_n_sentences_before	int	The number of sentences before the matching text	2
vectara_n_sentences_after	int	The number of sentences after the matching text.	2
vectara_summary_num_results	int	The number of results to use in summarization.	10
vectara_summarizer	str	The Vectara summarizer name.	'vectara-summary-ext-24-05-med-omni'
vectara_summary_response_language	str	The response language for the Vectara summary.	'eng'
vectara_summary_prompt_text	str	The custom prompt, using appropriate prompt variables and functions.	None
vectara_max_response_chars	int	The desired maximum number of characters for the generated summary.	None
vectara_max_tokens	int	The maximum number of tokens to be returned by the LLM.	None
vectara_temperature	float	The sampling temperature; higher values lead to more randomness.	None
vectara_frequency_penalty	float	How much to penalize repeating tokens in the response, higher values reducing likelihood of repeating the same line.	None
vectara_presence_penalty	float	How much to penalize repeating tokens in the response, higher values increasing the diversity of topics.	None
vectara_save_history	bool	Whether to save the query in history.	True

Returns:

Name	Type	Description
Agent	Agent	An instance of the Agent class.

Source code in vectara_agentic/agent.py

@classmethod
def from_corpus(
    cls,
    tool_name: str,
    data_description: str,
    assistant_specialty: str,
    vectara_corpus_key: str = str(os.environ.get("VECTARA_CORPUS_KEY", "")),
    vectara_api_key: str = str(os.environ.get("VECTARA_API_KEY", "")),
    agent_progress_callback: Optional[Callable[[AgentStatusType, str], None]] = None,
    query_logging_callback: Optional[Callable[[str, str], None]] = None,
    verbose: bool = False,
    vectara_filter_fields: list[dict] = [],
    vectara_offset: int = 0,
    vectara_lambda_val: float = 0.005,
    vectara_semantics: str = "default",
    vectara_custom_dimensions: Dict = {},
    vectara_reranker: str = "slingshot",
    vectara_rerank_k: int = 50,
    vectara_rerank_limit: Optional[int] = None,
    vectara_rerank_cutoff: Optional[float] = None,
    vectara_diversity_bias: float = 0.2,
    vectara_udf_expression: str = None,
    vectara_rerank_chain: List[Dict] = None,
    vectara_n_sentences_before: int = 2,
    vectara_n_sentences_after: int = 2,
    vectara_summary_num_results: int = 10,
    vectara_summarizer: str = "vectara-summary-ext-24-05-med-omni",
    vectara_summary_response_language: str = "eng",
    vectara_summary_prompt_text: Optional[str] = None,
    vectara_max_response_chars: Optional[int] = None,
    vectara_max_tokens: Optional[int] = None,
    vectara_temperature: Optional[float] = None,
    vectara_frequency_penalty: Optional[float] = None,
    vectara_presence_penalty: Optional[float] = None,
    vectara_save_history: bool = True,
) -> "Agent":
    """
    Create an agent from a single Vectara corpus

    Args:
        tool_name (str): The name of Vectara tool used by the agent
        vectara_corpus_key (str): The Vectara corpus key (or comma separated list of keys).
        vectara_api_key (str): The Vectara API key.
        agent_progress_callback (Callable): A callback function the code calls on any agent updates.
        query_logging_callback (Callable): A callback function the code calls upon completion of a query
        data_description (str): The description of the data.
        assistant_specialty (str): The specialty of the assistant.
        verbose (bool, optional): Whether to print verbose output.
        vectara_filter_fields (List[dict], optional): The filterable attributes
            (each dict maps field name to Tuple[type, description]).
        vectara_offset (int, optional): Number of results to skip.
        vectara_lambda_val (float, optional): Lambda value for Vectara hybrid search.
        vectara_semantics: (str, optional): Indicates whether the query is intended as a query or response.
        vectara_custom_dimensions: (Dict, optional): Custom dimensions for the query.
        vectara_reranker (str, optional): The Vectara reranker name (default "slingshot")
        vectara_rerank_k (int, optional): The number of results to use with reranking.
        vectara_rerank_limit: (int, optional): The maximum number of results to return after reranking.
        vectara_rerank_cutoff: (float, optional): The minimum score threshold for results to include after
            reranking.
        vectara_diversity_bias (float, optional): The MMR diversity bias.
        vectara_udf_expression (str, optional): The user defined expression for reranking results.
        vectara_rerank_chain (List[Dict], optional): A list of Vectara rerankers to be applied sequentially.
        vectara_n_sentences_before (int, optional): The number of sentences before the matching text
        vectara_n_sentences_after (int, optional): The number of sentences after the matching text.
        vectara_summary_num_results (int, optional): The number of results to use in summarization.
        vectara_summarizer (str, optional): The Vectara summarizer name.
        vectara_summary_response_language (str, optional): The response language for the Vectara summary.
        vectara_summary_prompt_text (str, optional): The custom prompt, using appropriate prompt variables and
            functions.
        vectara_max_response_chars (int, optional): The desired maximum number of characters for the generated
            summary.
        vectara_max_tokens (int, optional): The maximum number of tokens to be returned by the LLM.
        vectara_temperature (float, optional): The sampling temperature; higher values lead to more randomness.
        vectara_frequency_penalty (float, optional): How much to penalize repeating tokens in the response,
            higher values reducing likelihood of repeating the same line.
        vectara_presence_penalty (float, optional): How much to penalize repeating tokens in the response,
            higher values increasing the diversity of topics.
        vectara_save_history (bool, optional): Whether to save the query in history.

    Returns:
        Agent: An instance of the Agent class.
    """
    vec_factory = VectaraToolFactory(
        vectara_api_key=vectara_api_key,
        vectara_corpus_key=vectara_corpus_key,
    )
    field_definitions = {}
    field_definitions["query"] = (str, Field(description="The user query"))  # type: ignore
    for field in vectara_filter_fields:
        field_definitions[field["name"]] = (
            eval(field["type"]),
            Field(description=field["description"]),
        )  # type: ignore
    query_args = create_model("QueryArgs", **field_definitions)  # type: ignore

    # tool name must be valid Python function name
    if tool_name:
        tool_name = re.sub(r"[^A-Za-z0-9_]", "_", tool_name)

    vectara_tool = vec_factory.create_rag_tool(
        tool_name=tool_name or f"vectara_{vectara_corpus_key}",
        tool_description=f"""
        Given a user query,
        returns a response (str) to a user question about {data_description}.
        """,
        tool_args_schema=query_args,
        reranker=vectara_reranker,
        rerank_k=vectara_rerank_k,
        rerank_limit=vectara_rerank_limit,
        rerank_cutoff=vectara_rerank_cutoff,
        mmr_diversity_bias=vectara_diversity_bias,
        udf_expression=vectara_udf_expression,
        rerank_chain=vectara_rerank_chain,
        n_sentences_before=vectara_n_sentences_before,
        n_sentences_after=vectara_n_sentences_after,
        offset=vectara_offset,
        lambda_val=vectara_lambda_val,
        semantics=vectara_semantics,
        custom_dimensions=vectara_custom_dimensions,
        summary_num_results=vectara_summary_num_results,
        vectara_summarizer=vectara_summarizer,
        summary_response_lang=vectara_summary_response_language,
        vectara_prompt_text=vectara_summary_prompt_text,
        max_response_chars=vectara_max_response_chars,
        max_tokens=vectara_max_tokens,
        temperature=vectara_temperature,
        frequency_penalty=vectara_frequency_penalty,
        presence_penalty=vectara_presence_penalty,
        save_history=vectara_save_history,
        include_citations=True,
        verbose=verbose,
    )

    assistant_instructions = f"""
    - You are a helpful {assistant_specialty} assistant.
    - You can answer questions about {data_description}.
    - Never discuss politics, and always respond politely.
    """

    return cls(
        tools=[vectara_tool],
        topic=assistant_specialty,
        custom_instructions=assistant_instructions,
        verbose=verbose,
        agent_progress_callback=agent_progress_callback,
        query_logging_callback=query_logging_callback,
    )

from_dict(data) classmethod

Create an Agent instance from a dictionary.

Source code in vectara_agentic/agent.py

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "Agent":
    """Create an Agent instance from a dictionary."""
    agent_config = AgentConfig.from_dict(data["agent_config"])
    tools = []

    for tool_data in data["tools"]:
        # Recreate the dynamic model using the schema info
        if tool_data.get("fn_schema"):
            field_definitions = {}
            for field, values in tool_data["fn_schema"]["properties"].items():
                # Instead of checking for 'type', use the helper:
                field_type = get_field_type(values)
                # If there's a default value, include it.
                if "default" in values:
                    field_definitions[field] = (
                        field_type,
                        Field(description=values.get("description", ""), default=values["default"]),
                    )
                else:
                    field_definitions[field] = (
                        field_type,
                        Field(description=values.get("description", "")),
                    )
            query_args_model = create_model("QueryArgs", **field_definitions)  # type: ignore
        else:
            query_args_model = create_model("QueryArgs")

        fn = pickle.loads(tool_data["fn"].encode("latin-1")) if tool_data["fn"] else None
        async_fn = pickle.loads(tool_data["async_fn"].encode("latin-1")) if tool_data["async_fn"] else None

        tool = VectaraTool.from_defaults(
            name=tool_data["name"],
            description=tool_data["description"],
            fn=fn,
            async_fn=async_fn,
            fn_schema=query_args_model,  # Re-assign the recreated dynamic model
            tool_type=ToolType(tool_data["tool_type"]),
        )
        tools.append(tool)

    agent = cls(
        tools=tools,
        agent_config=agent_config,
        topic=data["topic"],
        custom_instructions=data["custom_instructions"],
        verbose=data["verbose"],
    )
    memory = pickle.loads(data["memory"].encode("latin-1")) if data.get("memory") else None
    if memory:
        agent.agent.memory = memory
    return agent

from_tools(tools, topic='general', custom_instructions='', verbose=True, update_func=None, agent_progress_callback=None, query_logging_callback=None, agent_config=AgentConfig(), chat_history=None) classmethod

Create an agent from tools, agent type, and language model.

Args:

tools (list[FunctionTool]): A list of tools to be used by the agent.
topic (str, optional): The topic for the agent. Defaults to 'general'.
custom_instructions (str, optional): custom instructions for the agent. Defaults to ''.
verbose (bool, optional): Whether the agent should print its steps. Defaults to True.
agent_progress_callback (Callable): A callback function the code calls on any agent updates.
    update_func (Callable): old name for agent_progress_callback. Will be deprecated in future.
query_logging_callback (Callable): A callback function the code calls upon completion of a query
agent_config (AgentConfig, optional): The configuration of the agent.
chat_history (Tuple[str, str], optional): A list of user/agent chat pairs to initialize the agent memory.

Returns:

Name	Type	Description
Agent	Agent	An instance of the Agent class.

Source code in vectara_agentic/agent.py

@classmethod
def from_tools(
    cls,
    tools: List[FunctionTool],
    topic: str = "general",
    custom_instructions: str = "",
    verbose: bool = True,
    update_func: Optional[Callable[[AgentStatusType, str], None]] = None,
    agent_progress_callback: Optional[Callable[[AgentStatusType, str], None]] = None,
    query_logging_callback: Optional[Callable[[str, str], None]] = None,
    agent_config: AgentConfig = AgentConfig(),
    chat_history: Optional[list[Tuple[str, str]]] = None,
) -> "Agent":
    """
    Create an agent from tools, agent type, and language model.

    Args:

        tools (list[FunctionTool]): A list of tools to be used by the agent.
        topic (str, optional): The topic for the agent. Defaults to 'general'.
        custom_instructions (str, optional): custom instructions for the agent. Defaults to ''.
        verbose (bool, optional): Whether the agent should print its steps. Defaults to True.
        agent_progress_callback (Callable): A callback function the code calls on any agent updates.
            update_func (Callable): old name for agent_progress_callback. Will be deprecated in future.
        query_logging_callback (Callable): A callback function the code calls upon completion of a query
        agent_config (AgentConfig, optional): The configuration of the agent.
        chat_history (Tuple[str, str], optional): A list of user/agent chat pairs to initialize the agent memory.

    Returns:
        Agent: An instance of the Agent class.
    """
    return cls(
        tools=tools, topic=topic, custom_instructions=custom_instructions,
        verbose=verbose, agent_progress_callback=agent_progress_callback,
        query_logging_callback=query_logging_callback,
        update_func=update_func, agent_config=agent_config,
        chat_history=chat_history,
    )

loads(data) classmethod

Create an Agent instance from a JSON string.

Source code in vectara_agentic/agent.py

@classmethod
def loads(cls, data: str) -> "Agent":
    """Create an Agent instance from a JSON string."""
    return cls.from_dict(json.loads(data))

report()

Get a report from the agent.

Returns:

Name	Type	Description
str	None	The report from the agent.

Source code in vectara_agentic/agent.py

def report(self) -> None:
    """
    Get a report from the agent.

    Returns:
        str: The report from the agent.
    """
    print("Vectara agentic Report:")
    print(f"Agent Type = {self.agent_type}")
    print(f"Topic = {self._topic}")
    print("Tools:")
    for tool in self.tools:
        if hasattr(tool, 'metadata'):
            print(f"- {tool.metadata.name}")
        else:
            print("- tool without metadata")
    print(f"Agent LLM = {get_llm(LLMRole.MAIN, config=self.agent_config).metadata.model_name}")
    print(f"Tool LLM = {get_llm(LLMRole.TOOL, config=self.agent_config).metadata.model_name}")

stream_chat(prompt)

Interact with the agent using a chat prompt with streaming. Args: prompt (str): The chat prompt. Returns: AgentStreamingResponse: The streaming response from the agent.

Source code in vectara_agentic/agent.py

def stream_chat(self, prompt: str) -> AgentStreamingResponse:    # type: ignore
    """
    Interact with the agent using a chat prompt with streaming.
    Args:
        prompt (str): The chat prompt.
    Returns:
        AgentStreamingResponse: The streaming response from the agent.
    """
    return asyncio.run(self.astream_chat(prompt))

to_dict()

Serialize the Agent instance to a dictionary.

Source code in vectara_agentic/agent.py

def to_dict(self) -> Dict[str, Any]:
    """Serialize the Agent instance to a dictionary."""
    tool_info = []

    for tool in self.tools:
        # Serialize each tool's metadata, function, and dynamic model schema (QueryArgs)
        tool_dict = {
            "tool_type": tool.metadata.tool_type.value,
            "name": tool.metadata.name,
            "description": tool.metadata.description,
            "fn": pickle.dumps(tool.fn).decode("latin-1") if tool.fn else None,  # Serialize fn
            "async_fn": pickle.dumps(tool.async_fn).decode("latin-1")
            if tool.async_fn
            else None,  # Serialize async_fn
            "fn_schema": tool.metadata.fn_schema.model_json_schema()
            if hasattr(tool.metadata, "fn_schema")
            else None,  # Serialize schema if available
        }
        tool_info.append(tool_dict)

    return {
        "agent_type": self.agent_type.value,
        "memory": pickle.dumps(self.agent.memory).decode("latin-1"),
        "tools": tool_info,
        "topic": self._topic,
        "custom_instructions": self._custom_instructions,
        "verbose": self.verbose,
        "agent_config": self.agent_config.to_dict(),
    }

token_counts()

Get the token counts for the agent and tools.

Returns:

Name	Type	Description
dict	dict	The token counts for the agent and tools.

Source code in vectara_agentic/agent.py

def token_counts(self) -> dict:
    """
    Get the token counts for the agent and tools.

    Returns:
        dict: The token counts for the agent and tools.
    """
    return {
        "main token count": self.main_token_counter.total_llm_token_count if self.main_token_counter else -1,
        "tool token count": self.tool_token_counter.total_llm_token_count if self.tool_token_counter else -1,
    }

VectaraTool

Bases: FunctionTool

A subclass of FunctionTool adding the tool_type attribute.

Source code in vectara_agentic/tools.py

class VectaraTool(FunctionTool):
    """
    A subclass of FunctionTool adding the tool_type attribute.
    """

    def __init__(
        self,
        tool_type: ToolType,
        metadata: ToolMetadata,
        fn: Optional[Callable[..., Any]] = None,
        async_fn: Optional[AsyncCallable] = None,
    ) -> None:
        metadata_dict = metadata.dict() if hasattr(metadata, 'dict') else metadata.__dict__
        vm = VectaraToolMetadata(tool_type=tool_type, **metadata_dict)
        super().__init__(fn, vm, async_fn)

    @classmethod
    def from_defaults(
        cls,
        fn: Optional[Callable[..., Any]] = None,
        name: Optional[str] = None,
        description: Optional[str] = None,
        return_direct: bool = False,
        fn_schema: Optional[Type[BaseModel]] = None,
        async_fn: Optional[AsyncCallable] = None,
        tool_metadata: Optional[ToolMetadata] = None,
        callback: Optional[Callable[[Any], Any]] = None,
        async_callback: Optional[AsyncCallable] = None,
        tool_type: ToolType = ToolType.QUERY,
    ) -> "VectaraTool":
        tool = FunctionTool.from_defaults(
            fn, name, description, return_direct, fn_schema, async_fn, tool_metadata,
            callback, async_callback
        )
        vectara_tool = cls(tool_type=tool_type, fn=tool.fn, metadata=tool.metadata, async_fn=tool.async_fn)
        return vectara_tool

    def __eq__(self, other):
        if self.metadata.tool_type != other.metadata.tool_type:
            return False

        if self.metadata.name != other.metadata.name or self.metadata.description != other.metadata.description:
            return False

        # Check if fn_schema is an instance of a BaseModel or a class itself (metaclass)
        self_schema_dict = self.metadata.fn_schema.model_fields
        other_schema_dict = other.metadata.fn_schema.model_fields
        is_equal = True
        for key in self_schema_dict.keys():
            if key not in other_schema_dict:
                is_equal = False
                break
            if (
                self_schema_dict[key].annotation != other_schema_dict[key].annotation
                or self_schema_dict[key].description != other_schema_dict[key].description
                or self_schema_dict[key].is_required() != other_schema_dict[key].is_required()
            ):
                is_equal = False
                break
        return is_equal

    def call(
        self, *args: Any, ctx: Optional[Context] = None, **kwargs: Any
    ) -> ToolOutput:
        try:
            return super().call(*args, ctx=ctx, **kwargs)
        except Exception as e:
            err_output = ToolOutput(
                tool_name=self.metadata.name,
                content=f"Tool Malfunction: {str(e)}",
                raw_input={"args": args, "kwargs": kwargs},
                raw_output={"response": str(e)},
            )
            return err_output

    async def acall(
        self, *args: Any, ctx: Optional[Context] = None, **kwargs: Any
    ) -> ToolOutput:
        try:
            return super().call(*args, ctx=ctx, **kwargs)
        except Exception as e:
            err_output = ToolOutput(
                tool_name=self.metadata.name,
                content=f"Tool Malfunction: {str(e)}",
                raw_input={"args": args, "kwargs": kwargs},
                raw_output={"response": str(e)},
            )
            return err_output

VectaraToolFactory

A factory class for creating Vectara RAG tools.

Source code in vectara_agentic/tools.py

class VectaraToolFactory:
    """
    A factory class for creating Vectara RAG tools.
    """

    def __init__(
        self,
        vectara_corpus_key: str = str(os.environ.get("VECTARA_CORPUS_KEY", "")),
        vectara_api_key: str = str(os.environ.get("VECTARA_API_KEY", "")),
    ) -> None:
        """
        Initialize the VectaraToolFactory
        Args:
            vectara_corpus_key (str): The Vectara corpus key (or comma separated list of keys).
            vectara_api_key (str): The Vectara API key.
        """
        self.vectara_corpus_key = vectara_corpus_key
        self.vectara_api_key = vectara_api_key
        self.num_corpora = len(vectara_corpus_key.split(","))

    def create_search_tool(
        self,
        tool_name: str,
        tool_description: str,
        tool_args_schema: type[BaseModel],
        tool_args_type: Dict[str, str] = {},
        fixed_filter: str = "",
        lambda_val: Union[List[float], float] = 0.005,
        semantics: Union[List[str] | str] = "default",
        custom_dimensions: Union[List[Dict], Dict] = {},
        offset: int = 0,
        n_sentences_before: int = 2,
        n_sentences_after: int = 2,
        reranker: str = "slingshot",
        rerank_k: int = 50,
        rerank_limit: Optional[int] = None,
        rerank_cutoff: Optional[float] = None,
        mmr_diversity_bias: float = 0.2,
        udf_expression: str = None,
        rerank_chain: List[Dict] = None,
        save_history: bool = True,
        verbose: bool = False,
        vectara_base_url: str = "https://api.vectara.io",
        vectara_verify_ssl: bool = True,
    ) -> VectaraTool:
        """
        Creates a Vectara search/retrieval tool

        Args:
            tool_name (str): The name of the tool.
            tool_description (str): The description of the tool.
            tool_args_schema (BaseModel): The schema for the tool arguments.
            tool_args_type (Dict[str, str], optional): The type of each argument (doc or part).
            fixed_filter (str, optional): A fixed Vectara filter condition to apply to all queries.
            lambda_val (Union[List[float] | float], optional): Lambda value (or list of values for each corpora)
                for the Vectara query, when using hybrid search.
            semantics (Union[List[str], str], optional): Indicates whether the query is intended as a query or response.
                Include list if using multiple corpora specifying the query type for each corpus.
            custom_dimensions (Union[List[Dict] | Dict], optional): Custom dimensions for the query (for each corpora).
            offset (int, optional): Number of results to skip.
            n_sentences_before (int, optional): Number of sentences before the matching document part.
            n_sentences_after (int, optional): Number of sentences after the matching document part.
            reranker (str, optional): The reranker mode.
            rerank_k (int, optional): Number of top-k documents for reranking.
            rerank_limit (int, optional): Maximum number of results to return after reranking.
            rerank_cutoff (float, optional): Minimum score threshold for results to include after reranking.
            mmr_diversity_bias (float, optional): MMR diversity bias.
            udf_expression (str, optional): the user defined expression for reranking results.
            rerank_chain (List[Dict], optional): A list of rerankers to be applied sequentially.
                Each dictionary should specify the "type" of reranker (mmr, slingshot, udf)
                and any other parameters (e.g. "limit" or "cutoff" for any type,
                "diversity_bias" for mmr, and "user_function" for udf).
                If using slingshot/multilingual_reranker_v1, it must be first in the list.
            save_history (bool, optional): Whether to save the query in history.
            verbose (bool, optional): Whether to print verbose output.
            vectara_base_url (str, optional): The base URL for the Vectara API.
            vectara_verify_ssl (bool, optional): Whether to verify SSL certificates for the Vectara API.

        Returns:
            VectaraTool: A VectaraTool object.
        """

        vectara = VectaraIndex(
            vectara_api_key=self.vectara_api_key,
            vectara_corpus_key=self.vectara_corpus_key,
            x_source_str="vectara-agentic",
            base_url=vectara_base_url,
            verify_ssl=vectara_verify_ssl,
        )

        # Dynamically generate the search function
        def search_function(*args, **kwargs) -> ToolOutput:
            """
            Dynamically generated function for semantic search Vectara.
            """
            # Convert args to kwargs using the function signature
            sig = inspect.signature(search_function)
            bound_args = sig.bind_partial(*args, **kwargs)
            bound_args.apply_defaults()
            kwargs = bound_args.arguments

            query = kwargs.pop("query")
            top_k = kwargs.pop("top_k", 10)
            try:
                filter_string = _build_filter_string(kwargs, tool_args_type, fixed_filter)
            except ValueError as e:
                return ToolOutput(
                    tool_name=search_function.__name__,
                    content=str(e),
                    raw_input={"args": args, "kwargs": kwargs},
                    raw_output={"response": str(e)},
                )

            vectara_retriever = vectara.as_retriever(
                summary_enabled=False,
                similarity_top_k=top_k,
                reranker=reranker,
                rerank_k=rerank_k if rerank_k * self.num_corpora <= 100 else int(100 / self.num_corpora),
                rerank_limit=rerank_limit,
                rerank_cutoff=rerank_cutoff,
                mmr_diversity_bias=mmr_diversity_bias,
                udf_expression=udf_expression,
                rerank_chain=rerank_chain,
                lambda_val=lambda_val,
                semantics=semantics,
                custom_dimensions=custom_dimensions,
                offset=offset,
                filter=filter_string,
                n_sentences_before=n_sentences_before,
                n_sentences_after=n_sentences_after,
                save_history=save_history,
                x_source_str="vectara-agentic",
                verbose=verbose,
            )
            response = vectara_retriever.retrieve(query)

            if len(response) == 0:
                msg = "Vectara Tool failed to retreive any results for the query."
                return ToolOutput(
                    tool_name=search_function.__name__,
                    content=msg,
                    raw_input={"args": args, "kwargs": kwargs},
                    raw_output={"response": msg},
                )
            tool_output = "Matching documents:\n"
            unique_ids = set()
            for doc in response:
                if doc.id_ in unique_ids:
                    continue
                unique_ids.add(doc.id_)
                tool_output += f"document '{doc.id_}' metadata: {doc.metadata}\n"
            out = ToolOutput(
                tool_name=search_function.__name__,
                content=tool_output,
                raw_input={"args": args, "kwargs": kwargs},
                raw_output=response,
            )
            return out

        fields = tool_args_schema.model_fields
        params = [
            inspect.Parameter(
                name=field_name,
                kind=inspect.Parameter.POSITIONAL_OR_KEYWORD,
                default=field_info.default,
                annotation=field_info,
            )
            for field_name, field_info in fields.items()
        ]

        # Create a new signature using the extracted parameters
        sig = inspect.Signature(params)
        search_function.__signature__ = sig
        search_function.__annotations__["return"] = dict[str, Any]
        search_function.__name__ = "_" + re.sub(r"[^A-Za-z0-9_]", "_", tool_name)

        # Create the tool function signature string
        fields = []
        for name, field in tool_args_schema.model_fields.items():
            annotation = field.annotation
            type_name = annotation.__name__ if hasattr(annotation, '__name__') else str(annotation)
            fields.append(f"{name}: {type_name}")
        args_str = ", ".join(fields)
        function_str = f"{tool_name}({args_str}) -> str"

        # Create the tool
        tool = VectaraTool.from_defaults(
            fn=search_function,
            name=tool_name,
            description=function_str + ". " + tool_description,
            fn_schema=tool_args_schema,
            tool_type=ToolType.QUERY,
        )
        return tool

    def create_rag_tool(
        self,
        tool_name: str,
        tool_description: str,
        tool_args_schema: type[BaseModel],
        tool_args_type: Dict[str, dict] = {},
        fixed_filter: str = "",
        vectara_summarizer: str = "vectara-summary-ext-24-05-med-omni",
        vectara_prompt_text: str = None,
        summary_num_results: int = 5,
        summary_response_lang: str = "eng",
        n_sentences_before: int = 2,
        n_sentences_after: int = 2,
        offset: int = 0,
        lambda_val: Union[List[float], float] = 0.005,
        semantics: Union[List[str] | str] = "default",
        custom_dimensions: Union[List[Dict], Dict] = {},
        reranker: str = "slingshot",
        rerank_k: int = 50,
        rerank_limit: Optional[int] = None,
        rerank_cutoff: Optional[float] = None,
        mmr_diversity_bias: float = 0.2,
        udf_expression: str = None,
        rerank_chain: List[Dict] = None,
        max_response_chars: Optional[int] = None,
        max_tokens: Optional[int] = None,
        temperature: Optional[float] = None,
        frequency_penalty: Optional[float] = None,
        presence_penalty: Optional[float] = None,
        include_citations: bool = True,
        save_history: bool = False,
        fcs_threshold: float = 0.0,
        verbose: bool = False,
        vectara_base_url: str = "https://api.vectara.io",
        vectara_verify_ssl: bool = True,
    ) -> VectaraTool:
        """
        Creates a RAG (Retrieve and Generate) tool.

        Args:
            tool_name (str): The name of the tool.
            tool_description (str): The description of the tool.
            tool_args_schema (BaseModel): The schema for the tool arguments.
            tool_args_type (Dict[str, dict], optional): attributes for each argument where they key is the field name
                and the value is a dictionary with the following keys:
                - 'type': the type of each filter attribute in Vectara (doc or part).
                - 'is_list': whether the filterable attribute is a list.
            fixed_filter (str, optional): A fixed Vectara filter condition to apply to all queries.
            vectara_summarizer (str, optional): The Vectara summarizer to use.
            vectara_prompt_text (str, optional): The prompt text for the Vectara summarizer.
            summary_num_results (int, optional): The number of summary results.
            summary_response_lang (str, optional): The response language for the summary.
            n_sentences_before (int, optional): Number of sentences before the summary.
            n_sentences_after (int, optional): Number of sentences after the summary.
            offset (int, optional): Number of results to skip.
            lambda_val (Union[List[float] | float], optional): Lambda value (or list of values for each corpora)
                for the Vectara query, when using hybrid search.
            semantics (Union[List[str], str], optional): Indicates whether the query is intended as a query or response.
                Include list if using multiple corpora specifying the query type for each corpus.
            custom_dimensions (Union[List[Dict] | Dict], optional): Custom dimensions for the query (for each corpora).
            reranker (str, optional): The reranker mode.
            rerank_k (int, optional): Number of top-k documents for reranking.
            rerank_limit (int, optional): Maximum number of results to return after reranking.
            rerank_cutoff (float, optional): Minimum score threshold for results to include after reranking.
            mmr_diversity_bias (float, optional): MMR diversity bias.
            udf_expression (str, optional): The user defined expression for reranking results.
            rerank_chain (List[Dict], optional): A list of rerankers to be applied sequentially.
                Each dictionary should specify the "type" of reranker (mmr, slingshot, udf)
                and any other parameters (e.g. "limit" or "cutoff" for any type,
                "diversity_bias" for mmr, and "user_function" for udf).
                If using slingshot/multilingual_reranker_v1, it must be first in the list.
            max_response_chars (int, optional): The desired maximum number of characters for the generated summary.
            max_tokens (int, optional): The maximum number of tokens to be returned by the LLM.
            temperature (float, optional): The sampling temperature; higher values lead to more randomness.
            frequency_penalty (float, optional): How much to penalize repeating tokens in the response,
                higher values reducing likelihood of repeating the same line.
            presence_penalty (float, optional): How much to penalize repeating tokens in the response,
                higher values increasing the diversity of topics.
            include_citations (bool, optional): Whether to include citations in the response.
                If True, uses markdown vectara citations that requires the Vectara scale plan.
            save_history (bool, optional): Whether to save the query in history.
            fcs_threshold (float, optional): A threshold for factual consistency.
                If set above 0, the tool notifies the calling agent that it "cannot respond" if FCS is too low.
            verbose (bool, optional): Whether to print verbose output.
            vectara_base_url (str, optional): The base URL for the Vectara API.
            vectara_verify_ssl (bool, optional): Whether to verify SSL certificates for the Vectara API.

        Returns:
            VectaraTool: A VectaraTool object.
        """

        vectara = VectaraIndex(
            vectara_api_key=self.vectara_api_key,
            vectara_corpus_key=self.vectara_corpus_key,
            x_source_str="vectara-agentic",
            base_url=vectara_base_url,
            verify_ssl=vectara_verify_ssl,
        )

        # Dynamically generate the RAG function
        def rag_function(*args, **kwargs) -> ToolOutput:
            """
            Dynamically generated function for RAG query with Vectara.
            """
            # Convert args to kwargs using the function signature
            sig = inspect.signature(rag_function)
            bound_args = sig.bind_partial(*args, **kwargs)
            bound_args.apply_defaults()
            kwargs = bound_args.arguments

            query = kwargs.pop("query")
            try:
                filter_string = _build_filter_string(kwargs, tool_args_type, fixed_filter)
            except ValueError as e:
                return ToolOutput(
                    tool_name=rag_function.__name__,
                    content=str(e),
                    raw_input={"args": args, "kwargs": kwargs},
                    raw_output={"response": str(e)},
                )

            vectara_query_engine = vectara.as_query_engine(
                summary_enabled=True,
                similarity_top_k=summary_num_results,
                summary_num_results=summary_num_results,
                summary_response_lang=summary_response_lang,
                summary_prompt_name=vectara_summarizer,
                prompt_text=vectara_prompt_text,
                reranker=reranker,
                rerank_k=rerank_k if rerank_k * self.num_corpora <= 100 else int(100 / self.num_corpora),
                rerank_limit=rerank_limit,
                rerank_cutoff=rerank_cutoff,
                mmr_diversity_bias=mmr_diversity_bias,
                udf_expression=udf_expression,
                rerank_chain=rerank_chain,
                n_sentence_before=n_sentences_before,
                n_sentence_after=n_sentences_after,
                offset=offset,
                lambda_val=lambda_val,
                semantics=semantics,
                custom_dimensions=custom_dimensions,
                filter=filter_string,
                max_response_chars=max_response_chars,
                max_tokens=max_tokens,
                temperature=temperature,
                frequency_penalty=frequency_penalty,
                presence_penalty=presence_penalty,
                citations_style="markdown" if include_citations else None,
                citations_url_pattern="{doc.url}" if include_citations else None,
                save_history=save_history,
                x_source_str="vectara-agentic",
                verbose=verbose,
            )
            response = vectara_query_engine.query(query)

            if len(response.source_nodes) == 0:
                msg = "Tool failed to generate a response since no matches were found."
                return ToolOutput(
                    tool_name=rag_function.__name__,
                    content=msg,
                    raw_input={"args": args, "kwargs": kwargs},
                    raw_output={"response": msg},
                )
            if str(response) == "None":
                msg = "Tool failed to generate a response."
                return ToolOutput(
                    tool_name=rag_function.__name__,
                    content=msg,
                    raw_input={"args": args, "kwargs": kwargs},
                    raw_output={"response": msg},
                )

            # Extract citation metadata
            pattern = r"\[(\d+)\]"
            matches = re.findall(pattern, response.response)
            citation_numbers = sorted(set(int(match) for match in matches))
            citation_metadata = ""
            keys_to_ignore = ["lang", "offset", "len"]
            for citation_number in citation_numbers:
                metadata = response.source_nodes[citation_number - 1].metadata
                citation_metadata += (
                    f"[{citation_number}]: "
                    + "; ".join(
                        [
                            f"{k}='{v}'"
                            for k, v in metadata.items()
                            if k not in keys_to_ignore
                        ]
                    )
                    + ".\n"
                )
            fcs = response.metadata["fcs"] if "fcs" in response.metadata else 0.0
            if fcs and fcs < fcs_threshold:
                msg = f"Could not answer the query due to suspected hallucination (fcs={fcs})."
                return ToolOutput(
                    tool_name=rag_function.__name__,
                    content=msg,
                    raw_input={"args": args, "kwargs": kwargs},
                    raw_output={"response": msg},
                )
            res = {
                "response": response.response,
                "references_metadata": citation_metadata,
            }
            if len(citation_metadata) > 0:
                tool_output = f"""
                    Response: '''{res['response']}'''
                    References:
                    {res['references_metadata']}
                """
            else:
                tool_output = f"Response: '''{res['response']}'''"
            out = ToolOutput(
                tool_name=rag_function.__name__,
                content=tool_output,
                raw_input={"args": args, "kwargs": kwargs},
                raw_output=res,
            )
            return out

        fields = tool_args_schema.model_fields
        params = [
            inspect.Parameter(
                name=field_name,
                kind=inspect.Parameter.POSITIONAL_OR_KEYWORD,
                default=field_info.default,
                annotation=field_info,
            )
            for field_name, field_info in fields.items()
        ]

        # Create a new signature using the extracted parameters
        sig = inspect.Signature(params)
        rag_function.__signature__ = sig
        rag_function.__annotations__["return"] = dict[str, Any]
        rag_function.__name__ = "_" + re.sub(r"[^A-Za-z0-9_]", "_", tool_name)

        # Create the tool function signature string
        fields = []
        for name, field in tool_args_schema.model_fields.items():
            annotation = field.annotation
            type_name = annotation.__name__ if hasattr(annotation, '__name__') else str(annotation)
            fields.append(f"{name}: {type_name}")
        args_str = ", ".join(fields)
        function_str = f"{tool_name}({args_str}) -> str"

        # Create the tool
        tool = VectaraTool.from_defaults(
            fn=rag_function,
            name=tool_name,
            description=function_str + ". " + tool_description,
            fn_schema=tool_args_schema,
            tool_type=ToolType.QUERY,
        )
        return tool

__init__(vectara_corpus_key=str(os.environ.get('VECTARA_CORPUS_KEY', '')), vectara_api_key=str(os.environ.get('VECTARA_API_KEY', '')))

Initialize the VectaraToolFactory Args: vectara_corpus_key (str): The Vectara corpus key (or comma separated list of keys). vectara_api_key (str): The Vectara API key.

Source code in vectara_agentic/tools.py

def __init__(
    self,
    vectara_corpus_key: str = str(os.environ.get("VECTARA_CORPUS_KEY", "")),
    vectara_api_key: str = str(os.environ.get("VECTARA_API_KEY", "")),
) -> None:
    """
    Initialize the VectaraToolFactory
    Args:
        vectara_corpus_key (str): The Vectara corpus key (or comma separated list of keys).
        vectara_api_key (str): The Vectara API key.
    """
    self.vectara_corpus_key = vectara_corpus_key
    self.vectara_api_key = vectara_api_key
    self.num_corpora = len(vectara_corpus_key.split(","))

create_rag_tool(tool_name, tool_description, tool_args_schema, tool_args_type={}, fixed_filter='', vectara_summarizer='vectara-summary-ext-24-05-med-omni', vectara_prompt_text=None, summary_num_results=5, summary_response_lang='eng', n_sentences_before=2, n_sentences_after=2, offset=0, lambda_val=0.005, semantics='default', custom_dimensions={}, reranker='slingshot', rerank_k=50, rerank_limit=None, rerank_cutoff=None, mmr_diversity_bias=0.2, udf_expression=None, rerank_chain=None, max_response_chars=None, max_tokens=None, temperature=None, frequency_penalty=None, presence_penalty=None, include_citations=True, save_history=False, fcs_threshold=0.0, verbose=False, vectara_base_url='https://api.vectara.io', vectara_verify_ssl=True)

Creates a RAG (Retrieve and Generate) tool.

Parameters:

Name	Type	Description	Default
tool_name	str	The name of the tool.	required
tool_description	str	The description of the tool.	required
tool_args_schema	BaseModel	The schema for the tool arguments.	required
tool_args_type	Dict[str, dict]	attributes for each argument where they key is the field name and the value is a dictionary with the following keys: - 'type': the type of each filter attribute in Vectara (doc or part). - 'is_list': whether the filterable attribute is a list.	{}
fixed_filter	str	A fixed Vectara filter condition to apply to all queries.	''
vectara_summarizer	str	The Vectara summarizer to use.	'vectara-summary-ext-24-05-med-omni'
vectara_prompt_text	str	The prompt text for the Vectara summarizer.	None
summary_num_results	int	The number of summary results.	5
summary_response_lang	str	The response language for the summary.	'eng'
n_sentences_before	int	Number of sentences before the summary.	2
n_sentences_after	int	Number of sentences after the summary.	2
offset	int	Number of results to skip.	0
lambda_val	Union[List[float] | float]	Lambda value (or list of values for each corpora) for the Vectara query, when using hybrid search.	0.005
semantics	Union[List[str], str]	Indicates whether the query is intended as a query or response. Include list if using multiple corpora specifying the query type for each corpus.	'default'
custom_dimensions	Union[List[Dict] | Dict]	Custom dimensions for the query (for each corpora).	{}
reranker	str	The reranker mode.	'slingshot'
rerank_k	int	Number of top-k documents for reranking.	50
rerank_limit	int	Maximum number of results to return after reranking.	None
rerank_cutoff	float	Minimum score threshold for results to include after reranking.	None
mmr_diversity_bias	float	MMR diversity bias.	0.2
udf_expression	str	The user defined expression for reranking results.	None
rerank_chain	List[Dict]	A list of rerankers to be applied sequentially. Each dictionary should specify the "type" of reranker (mmr, slingshot, udf) and any other parameters (e.g. "limit" or "cutoff" for any type, "diversity_bias" for mmr, and "user_function" for udf). If using slingshot/multilingual_reranker_v1, it must be first in the list.	None
max_response_chars	int	The desired maximum number of characters for the generated summary.	None
max_tokens	int	The maximum number of tokens to be returned by the LLM.	None
temperature	float	The sampling temperature; higher values lead to more randomness.	None
frequency_penalty	float	How much to penalize repeating tokens in the response, higher values reducing likelihood of repeating the same line.	None
presence_penalty	float	How much to penalize repeating tokens in the response, higher values increasing the diversity of topics.	None
include_citations	bool	Whether to include citations in the response. If True, uses markdown vectara citations that requires the Vectara scale plan.	True
save_history	bool	Whether to save the query in history.	False
fcs_threshold	float	A threshold for factual consistency. If set above 0, the tool notifies the calling agent that it "cannot respond" if FCS is too low.	0.0
verbose	bool	Whether to print verbose output.	False
vectara_base_url	str	The base URL for the Vectara API.	'https://api.vectara.io'
vectara_verify_ssl	bool	Whether to verify SSL certificates for the Vectara API.	True

Returns:

Name	Type	Description
VectaraTool	VectaraTool	A VectaraTool object.

Source code in vectara_agentic/tools.py

def create_rag_tool(
    self,
    tool_name: str,
    tool_description: str,
    tool_args_schema: type[BaseModel],
    tool_args_type: Dict[str, dict] = {},
    fixed_filter: str = "",
    vectara_summarizer: str = "vectara-summary-ext-24-05-med-omni",
    vectara_prompt_text: str = None,
    summary_num_results: int = 5,
    summary_response_lang: str = "eng",
    n_sentences_before: int = 2,
    n_sentences_after: int = 2,
    offset: int = 0,
    lambda_val: Union[List[float], float] = 0.005,
    semantics: Union[List[str] | str] = "default",
    custom_dimensions: Union[List[Dict], Dict] = {},
    reranker: str = "slingshot",
    rerank_k: int = 50,
    rerank_limit: Optional[int] = None,
    rerank_cutoff: Optional[float] = None,
    mmr_diversity_bias: float = 0.2,
    udf_expression: str = None,
    rerank_chain: List[Dict] = None,
    max_response_chars: Optional[int] = None,
    max_tokens: Optional[int] = None,
    temperature: Optional[float] = None,
    frequency_penalty: Optional[float] = None,
    presence_penalty: Optional[float] = None,
    include_citations: bool = True,
    save_history: bool = False,
    fcs_threshold: float = 0.0,
    verbose: bool = False,
    vectara_base_url: str = "https://api.vectara.io",
    vectara_verify_ssl: bool = True,
) -> VectaraTool:
    """
    Creates a RAG (Retrieve and Generate) tool.

    Args:
        tool_name (str): The name of the tool.
        tool_description (str): The description of the tool.
        tool_args_schema (BaseModel): The schema for the tool arguments.
        tool_args_type (Dict[str, dict], optional): attributes for each argument where they key is the field name
            and the value is a dictionary with the following keys:
            - 'type': the type of each filter attribute in Vectara (doc or part).
            - 'is_list': whether the filterable attribute is a list.
        fixed_filter (str, optional): A fixed Vectara filter condition to apply to all queries.
        vectara_summarizer (str, optional): The Vectara summarizer to use.
        vectara_prompt_text (str, optional): The prompt text for the Vectara summarizer.
        summary_num_results (int, optional): The number of summary results.
        summary_response_lang (str, optional): The response language for the summary.
        n_sentences_before (int, optional): Number of sentences before the summary.
        n_sentences_after (int, optional): Number of sentences after the summary.
        offset (int, optional): Number of results to skip.
        lambda_val (Union[List[float] | float], optional): Lambda value (or list of values for each corpora)
            for the Vectara query, when using hybrid search.
        semantics (Union[List[str], str], optional): Indicates whether the query is intended as a query or response.
            Include list if using multiple corpora specifying the query type for each corpus.
        custom_dimensions (Union[List[Dict] | Dict], optional): Custom dimensions for the query (for each corpora).
        reranker (str, optional): The reranker mode.
        rerank_k (int, optional): Number of top-k documents for reranking.
        rerank_limit (int, optional): Maximum number of results to return after reranking.
        rerank_cutoff (float, optional): Minimum score threshold for results to include after reranking.
        mmr_diversity_bias (float, optional): MMR diversity bias.
        udf_expression (str, optional): The user defined expression for reranking results.
        rerank_chain (List[Dict], optional): A list of rerankers to be applied sequentially.
            Each dictionary should specify the "type" of reranker (mmr, slingshot, udf)
            and any other parameters (e.g. "limit" or "cutoff" for any type,
            "diversity_bias" for mmr, and "user_function" for udf).
            If using slingshot/multilingual_reranker_v1, it must be first in the list.
        max_response_chars (int, optional): The desired maximum number of characters for the generated summary.
        max_tokens (int, optional): The maximum number of tokens to be returned by the LLM.
        temperature (float, optional): The sampling temperature; higher values lead to more randomness.
        frequency_penalty (float, optional): How much to penalize repeating tokens in the response,
            higher values reducing likelihood of repeating the same line.
        presence_penalty (float, optional): How much to penalize repeating tokens in the response,
            higher values increasing the diversity of topics.
        include_citations (bool, optional): Whether to include citations in the response.
            If True, uses markdown vectara citations that requires the Vectara scale plan.
        save_history (bool, optional): Whether to save the query in history.
        fcs_threshold (float, optional): A threshold for factual consistency.
            If set above 0, the tool notifies the calling agent that it "cannot respond" if FCS is too low.
        verbose (bool, optional): Whether to print verbose output.
        vectara_base_url (str, optional): The base URL for the Vectara API.
        vectara_verify_ssl (bool, optional): Whether to verify SSL certificates for the Vectara API.

    Returns:
        VectaraTool: A VectaraTool object.
    """

    vectara = VectaraIndex(
        vectara_api_key=self.vectara_api_key,
        vectara_corpus_key=self.vectara_corpus_key,
        x_source_str="vectara-agentic",
        base_url=vectara_base_url,
        verify_ssl=vectara_verify_ssl,
    )

    # Dynamically generate the RAG function
    def rag_function(*args, **kwargs) -> ToolOutput:
        """
        Dynamically generated function for RAG query with Vectara.
        """
        # Convert args to kwargs using the function signature
        sig = inspect.signature(rag_function)
        bound_args = sig.bind_partial(*args, **kwargs)
        bound_args.apply_defaults()
        kwargs = bound_args.arguments

        query = kwargs.pop("query")
        try:
            filter_string = _build_filter_string(kwargs, tool_args_type, fixed_filter)
        except ValueError as e:
            return ToolOutput(
                tool_name=rag_function.__name__,
                content=str(e),
                raw_input={"args": args, "kwargs": kwargs},
                raw_output={"response": str(e)},
            )

        vectara_query_engine = vectara.as_query_engine(
            summary_enabled=True,
            similarity_top_k=summary_num_results,
            summary_num_results=summary_num_results,
            summary_response_lang=summary_response_lang,
            summary_prompt_name=vectara_summarizer,
            prompt_text=vectara_prompt_text,
            reranker=reranker,
            rerank_k=rerank_k if rerank_k * self.num_corpora <= 100 else int(100 / self.num_corpora),
            rerank_limit=rerank_limit,
            rerank_cutoff=rerank_cutoff,
            mmr_diversity_bias=mmr_diversity_bias,
            udf_expression=udf_expression,
            rerank_chain=rerank_chain,
            n_sentence_before=n_sentences_before,
            n_sentence_after=n_sentences_after,
            offset=offset,
            lambda_val=lambda_val,
            semantics=semantics,
            custom_dimensions=custom_dimensions,
            filter=filter_string,
            max_response_chars=max_response_chars,
            max_tokens=max_tokens,
            temperature=temperature,
            frequency_penalty=frequency_penalty,
            presence_penalty=presence_penalty,
            citations_style="markdown" if include_citations else None,
            citations_url_pattern="{doc.url}" if include_citations else None,
            save_history=save_history,
            x_source_str="vectara-agentic",
            verbose=verbose,
        )
        response = vectara_query_engine.query(query)

        if len(response.source_nodes) == 0:
            msg = "Tool failed to generate a response since no matches were found."
            return ToolOutput(
                tool_name=rag_function.__name__,
                content=msg,
                raw_input={"args": args, "kwargs": kwargs},
                raw_output={"response": msg},
            )
        if str(response) == "None":
            msg = "Tool failed to generate a response."
            return ToolOutput(
                tool_name=rag_function.__name__,
                content=msg,
                raw_input={"args": args, "kwargs": kwargs},
                raw_output={"response": msg},
            )

        # Extract citation metadata
        pattern = r"\[(\d+)\]"
        matches = re.findall(pattern, response.response)
        citation_numbers = sorted(set(int(match) for match in matches))
        citation_metadata = ""
        keys_to_ignore = ["lang", "offset", "len"]
        for citation_number in citation_numbers:
            metadata = response.source_nodes[citation_number - 1].metadata
            citation_metadata += (
                f"[{citation_number}]: "
                + "; ".join(
                    [
                        f"{k}='{v}'"
                        for k, v in metadata.items()
                        if k not in keys_to_ignore
                    ]
                )
                + ".\n"
            )
        fcs = response.metadata["fcs"] if "fcs" in response.metadata else 0.0
        if fcs and fcs < fcs_threshold:
            msg = f"Could not answer the query due to suspected hallucination (fcs={fcs})."
            return ToolOutput(
                tool_name=rag_function.__name__,
                content=msg,
                raw_input={"args": args, "kwargs": kwargs},
                raw_output={"response": msg},
            )
        res = {
            "response": response.response,
            "references_metadata": citation_metadata,
        }
        if len(citation_metadata) > 0:
            tool_output = f"""
                Response: '''{res['response']}'''
                References:
                {res['references_metadata']}
            """
        else:
            tool_output = f"Response: '''{res['response']}'''"
        out = ToolOutput(
            tool_name=rag_function.__name__,
            content=tool_output,
            raw_input={"args": args, "kwargs": kwargs},
            raw_output=res,
        )
        return out

    fields = tool_args_schema.model_fields
    params = [
        inspect.Parameter(
            name=field_name,
            kind=inspect.Parameter.POSITIONAL_OR_KEYWORD,
            default=field_info.default,
            annotation=field_info,
        )
        for field_name, field_info in fields.items()
    ]

    # Create a new signature using the extracted parameters
    sig = inspect.Signature(params)
    rag_function.__signature__ = sig
    rag_function.__annotations__["return"] = dict[str, Any]
    rag_function.__name__ = "_" + re.sub(r"[^A-Za-z0-9_]", "_", tool_name)

    # Create the tool function signature string
    fields = []
    for name, field in tool_args_schema.model_fields.items():
        annotation = field.annotation
        type_name = annotation.__name__ if hasattr(annotation, '__name__') else str(annotation)
        fields.append(f"{name}: {type_name}")
    args_str = ", ".join(fields)
    function_str = f"{tool_name}({args_str}) -> str"

    # Create the tool
    tool = VectaraTool.from_defaults(
        fn=rag_function,
        name=tool_name,
        description=function_str + ". " + tool_description,
        fn_schema=tool_args_schema,
        tool_type=ToolType.QUERY,
    )
    return tool

create_search_tool(tool_name, tool_description, tool_args_schema, tool_args_type={}, fixed_filter='', lambda_val=0.005, semantics='default', custom_dimensions={}, offset=0, n_sentences_before=2, n_sentences_after=2, reranker='slingshot', rerank_k=50, rerank_limit=None, rerank_cutoff=None, mmr_diversity_bias=0.2, udf_expression=None, rerank_chain=None, save_history=True, verbose=False, vectara_base_url='https://api.vectara.io', vectara_verify_ssl=True)

Creates a Vectara search/retrieval tool

Parameters:

Name	Type	Description	Default
tool_name	str	The name of the tool.	required
tool_description	str	The description of the tool.	required
tool_args_schema	BaseModel	The schema for the tool arguments.	required
tool_args_type	Dict[str, str]	The type of each argument (doc or part).	{}
fixed_filter	str	A fixed Vectara filter condition to apply to all queries.	''
lambda_val	Union[List[float] | float]	Lambda value (or list of values for each corpora) for the Vectara query, when using hybrid search.	0.005
semantics	Union[List[str], str]	Indicates whether the query is intended as a query or response. Include list if using multiple corpora specifying the query type for each corpus.	'default'
custom_dimensions	Union[List[Dict] | Dict]	Custom dimensions for the query (for each corpora).	{}
offset	int	Number of results to skip.	0
n_sentences_before	int	Number of sentences before the matching document part.	2
n_sentences_after	int	Number of sentences after the matching document part.	2
reranker	str	The reranker mode.	'slingshot'
rerank_k	int	Number of top-k documents for reranking.	50
rerank_limit	int	Maximum number of results to return after reranking.	None
rerank_cutoff	float	Minimum score threshold for results to include after reranking.	None
mmr_diversity_bias	float	MMR diversity bias.	0.2
udf_expression	str	the user defined expression for reranking results.	None
rerank_chain	List[Dict]	A list of rerankers to be applied sequentially. Each dictionary should specify the "type" of reranker (mmr, slingshot, udf) and any other parameters (e.g. "limit" or "cutoff" for any type, "diversity_bias" for mmr, and "user_function" for udf). If using slingshot/multilingual_reranker_v1, it must be first in the list.	None
save_history	bool	Whether to save the query in history.	True
verbose	bool	Whether to print verbose output.	False
vectara_base_url	str	The base URL for the Vectara API.	'https://api.vectara.io'
vectara_verify_ssl	bool	Whether to verify SSL certificates for the Vectara API.	True

Returns:

Name	Type	Description
VectaraTool	VectaraTool	A VectaraTool object.

Source code in vectara_agentic/tools.py

def create_search_tool(
    self,
    tool_name: str,
    tool_description: str,
    tool_args_schema: type[BaseModel],
    tool_args_type: Dict[str, str] = {},
    fixed_filter: str = "",
    lambda_val: Union[List[float], float] = 0.005,
    semantics: Union[List[str] | str] = "default",
    custom_dimensions: Union[List[Dict], Dict] = {},
    offset: int = 0,
    n_sentences_before: int = 2,
    n_sentences_after: int = 2,
    reranker: str = "slingshot",
    rerank_k: int = 50,
    rerank_limit: Optional[int] = None,
    rerank_cutoff: Optional[float] = None,
    mmr_diversity_bias: float = 0.2,
    udf_expression: str = None,
    rerank_chain: List[Dict] = None,
    save_history: bool = True,
    verbose: bool = False,
    vectara_base_url: str = "https://api.vectara.io",
    vectara_verify_ssl: bool = True,
) -> VectaraTool:
    """
    Creates a Vectara search/retrieval tool

    Args:
        tool_name (str): The name of the tool.
        tool_description (str): The description of the tool.
        tool_args_schema (BaseModel): The schema for the tool arguments.
        tool_args_type (Dict[str, str], optional): The type of each argument (doc or part).
        fixed_filter (str, optional): A fixed Vectara filter condition to apply to all queries.
        lambda_val (Union[List[float] | float], optional): Lambda value (or list of values for each corpora)
            for the Vectara query, when using hybrid search.
        semantics (Union[List[str], str], optional): Indicates whether the query is intended as a query or response.
            Include list if using multiple corpora specifying the query type for each corpus.
        custom_dimensions (Union[List[Dict] | Dict], optional): Custom dimensions for the query (for each corpora).
        offset (int, optional): Number of results to skip.
        n_sentences_before (int, optional): Number of sentences before the matching document part.
        n_sentences_after (int, optional): Number of sentences after the matching document part.
        reranker (str, optional): The reranker mode.
        rerank_k (int, optional): Number of top-k documents for reranking.
        rerank_limit (int, optional): Maximum number of results to return after reranking.
        rerank_cutoff (float, optional): Minimum score threshold for results to include after reranking.
        mmr_diversity_bias (float, optional): MMR diversity bias.
        udf_expression (str, optional): the user defined expression for reranking results.
        rerank_chain (List[Dict], optional): A list of rerankers to be applied sequentially.
            Each dictionary should specify the "type" of reranker (mmr, slingshot, udf)
            and any other parameters (e.g. "limit" or "cutoff" for any type,
            "diversity_bias" for mmr, and "user_function" for udf).
            If using slingshot/multilingual_reranker_v1, it must be first in the list.
        save_history (bool, optional): Whether to save the query in history.
        verbose (bool, optional): Whether to print verbose output.
        vectara_base_url (str, optional): The base URL for the Vectara API.
        vectara_verify_ssl (bool, optional): Whether to verify SSL certificates for the Vectara API.

    Returns:
        VectaraTool: A VectaraTool object.
    """

    vectara = VectaraIndex(
        vectara_api_key=self.vectara_api_key,
        vectara_corpus_key=self.vectara_corpus_key,
        x_source_str="vectara-agentic",
        base_url=vectara_base_url,
        verify_ssl=vectara_verify_ssl,
    )

    # Dynamically generate the search function
    def search_function(*args, **kwargs) -> ToolOutput:
        """
        Dynamically generated function for semantic search Vectara.
        """
        # Convert args to kwargs using the function signature
        sig = inspect.signature(search_function)
        bound_args = sig.bind_partial(*args, **kwargs)
        bound_args.apply_defaults()
        kwargs = bound_args.arguments

        query = kwargs.pop("query")
        top_k = kwargs.pop("top_k", 10)
        try:
            filter_string = _build_filter_string(kwargs, tool_args_type, fixed_filter)
        except ValueError as e:
            return ToolOutput(
                tool_name=search_function.__name__,
                content=str(e),
                raw_input={"args": args, "kwargs": kwargs},
                raw_output={"response": str(e)},
            )

        vectara_retriever = vectara.as_retriever(
            summary_enabled=False,
            similarity_top_k=top_k,
            reranker=reranker,
            rerank_k=rerank_k if rerank_k * self.num_corpora <= 100 else int(100 / self.num_corpora),
            rerank_limit=rerank_limit,
            rerank_cutoff=rerank_cutoff,
            mmr_diversity_bias=mmr_diversity_bias,
            udf_expression=udf_expression,
            rerank_chain=rerank_chain,
            lambda_val=lambda_val,
            semantics=semantics,
            custom_dimensions=custom_dimensions,
            offset=offset,
            filter=filter_string,
            n_sentences_before=n_sentences_before,
            n_sentences_after=n_sentences_after,
            save_history=save_history,
            x_source_str="vectara-agentic",
            verbose=verbose,
        )
        response = vectara_retriever.retrieve(query)

        if len(response) == 0:
            msg = "Vectara Tool failed to retreive any results for the query."
            return ToolOutput(
                tool_name=search_function.__name__,
                content=msg,
                raw_input={"args": args, "kwargs": kwargs},
                raw_output={"response": msg},
            )
        tool_output = "Matching documents:\n"
        unique_ids = set()
        for doc in response:
            if doc.id_ in unique_ids:
                continue
            unique_ids.add(doc.id_)
            tool_output += f"document '{doc.id_}' metadata: {doc.metadata}\n"
        out = ToolOutput(
            tool_name=search_function.__name__,
            content=tool_output,
            raw_input={"args": args, "kwargs": kwargs},
            raw_output=response,
        )
        return out

    fields = tool_args_schema.model_fields
    params = [
        inspect.Parameter(
            name=field_name,
            kind=inspect.Parameter.POSITIONAL_OR_KEYWORD,
            default=field_info.default,
            annotation=field_info,
        )
        for field_name, field_info in fields.items()
    ]

    # Create a new signature using the extracted parameters
    sig = inspect.Signature(params)
    search_function.__signature__ = sig
    search_function.__annotations__["return"] = dict[str, Any]
    search_function.__name__ = "_" + re.sub(r"[^A-Za-z0-9_]", "_", tool_name)

    # Create the tool function signature string
    fields = []
    for name, field in tool_args_schema.model_fields.items():
        annotation = field.annotation
        type_name = annotation.__name__ if hasattr(annotation, '__name__') else str(annotation)
        fields.append(f"{name}: {type_name}")
    args_str = ", ".join(fields)
    function_str = f"{tool_name}({args_str}) -> str"

    # Create the tool
    tool = VectaraTool.from_defaults(
        fn=search_function,
        name=tool_name,
        description=function_str + ". " + tool_description,
        fn_schema=tool_args_schema,
        tool_type=ToolType.QUERY,
    )
    return tool