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,
        use_structured_planning: bool = False,
        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,
        fallback_agent_config: Optional[AgentConfig] = None,
        chat_history: Optional[list[Tuple[str, str]]] = None,
        validate_tools: bool = False,
        workflow_cls: Workflow = None,
        workflow_timeout: int = 120,
    ) -> 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.
            use_structured_planning (bool, optional)
                Whether or not we want to wrap the agent with LlamaIndex StructuredPlannerAgent.
            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.
            fallback_agent_config (AgentConfig, optional): The fallback configuration of the agent.
                This config is used when the main agent config fails multiple times.
            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.
            workflow_cls (Workflow, optional): The workflow class to be used with run(). Defaults to None.
            workflow_timeout (int, optional): The timeout for the workflow in seconds. Defaults to 120.
        """
        self.agent_config = agent_config or AgentConfig()
        self.agent_config_type = AgentConfigType.DEFAULT
        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.agent_type = self.agent_config.agent_type
        self.use_structured_planning = use_structured_planning
        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

        self.workflow_cls = workflow_cls
        self.workflow_timeout = workflow_timeout

        # 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.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)

        # Set up main agent and fallback agent
        self.agent = self._create_agent(self.agent_config, callback_manager)
        self.fallback_agent_config = fallback_agent_config
        if self.fallback_agent_config:
            self.fallback_agent = self._create_agent(self.fallback_agent_config, callback_manager)
        else:
            self.fallback_agent_config = None

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

    def _create_agent(
        self,
        config: AgentConfig,
        llm_callback_manager: CallbackManager
    ) -> Union[BaseAgent, AgentRunner]:
        """
        Creates the agent based on the configuration object.

        Args:

            config: The configuration of the agent.
            llm_callback_manager: The callback manager for the agent's llm.

        Returns:
            Union[BaseAgent, AgentRunner]: The configured agent object.
        """
        agent_type = config.agent_type
        llm = get_llm(LLMRole.MAIN, config=config)
        llm.callback_manager = llm_callback_manager

        if agent_type == AgentType.FUNCTION_CALLING:
            if config.tool_llm_provider == ModelProvider.OPENAI:
                raise ValueError(
                    "Vectara-agentic: Function calling agent type is not supported with the OpenAI LLM."
                )
            prompt = _get_prompt(GENERAL_PROMPT_TEMPLATE, self._topic, self._custom_instructions)
            agent = FunctionCallingAgent.from_tools(
                tools=self.tools,
                llm=llm,
                memory=self.memory,
                verbose=self.verbose,
                max_function_calls=config.max_reasoning_steps,
                callback_manager=llm_callback_manager,
                system_prompt=prompt,
                allow_parallel_tool_calls=True,
            )
        elif agent_type == AgentType.REACT:
            prompt = _get_prompt(REACT_PROMPT_TEMPLATE, self._topic, self._custom_instructions)
            agent = ReActAgent.from_tools(
                tools=self.tools,
                llm=llm,
                memory=self.memory,
                verbose=self.verbose,
                react_chat_formatter=ReActChatFormatter(system_header=prompt),
                max_iterations=config.max_reasoning_steps,
                callable_manager=llm_callback_manager,
            )
        elif agent_type == AgentType.OPENAI:
            if config.tool_llm_provider != ModelProvider.OPENAI:
                raise ValueError(
                    "Vectara-agentic: OPENAI agent type requires the OpenAI LLM."
                )
            prompt = _get_prompt(GENERAL_PROMPT_TEMPLATE, self._topic, self._custom_instructions)
            agent = OpenAIAgent.from_tools(
                tools=self.tools,
                llm=llm,
                memory=self.memory,
                verbose=self.verbose,
                callback_manager=llm_callback_manager,
                max_function_calls=config.max_reasoning_steps,
                system_prompt=prompt,
            )
        elif agent_type == AgentType.LLMCOMPILER:
            agent_worker = LLMCompilerAgentWorker.from_tools(
                tools=self.tools,
                llm=llm,
                verbose=self.verbose,
                callback_manager=llm_callback_manager,
            )
            agent_worker.system_prompt = _get_prompt(
                _get_llm_compiler_prompt(agent_worker.system_prompt, self._topic, self._custom_instructions),
                self._topic, self._custom_instructions
            )
            agent_worker.system_prompt_replan = _get_prompt(
                _get_llm_compiler_prompt(agent_worker.system_prompt_replan, self._topic, self._custom_instructions),
                self._topic, self._custom_instructions
            )
            agent = agent_worker.as_agent()
        elif agent_type == AgentType.LATS:
            agent_worker = LATSAgentWorker.from_tools(
                tools=self.tools,
                llm=llm,
                num_expansions=3,
                max_rollouts=-1,
                verbose=self.verbose,
                callback_manager=llm_callback_manager,
            )
            prompt = _get_prompt(REACT_PROMPT_TEMPLATE, self._topic, self._custom_instructions)
            agent_worker.chat_formatter = ReActChatFormatter(system_header=prompt)
            agent = agent_worker.as_agent()
        else:
            raise ValueError(f"Unknown agent type: {agent_type}")

        # Set up structured planner if needed
        if (self.use_structured_planning
            or self.agent_type in [AgentType.LLMCOMPILER, AgentType.LATS]):
            agent = StructuredPlannerAgent(
                agent_worker=agent.agent_worker,
                tools=self.tools,
                memory=self.memory,
                verbose=self.verbose,
                initial_plan_prompt=STRUCTURED_PLANNER_INITIAL_PLAN_PROMPT,
                plan_refine_prompt=STRUCTURED_PLANNER_PLAN_REFINE_PROMPT,
            )

        return agent

    def clear_memory(self) -> None:
        """
        Clear the agent's memory.
        """
        if self.agent_config_type == AgentConfigType.DEFAULT:
            self.agent.memory.reset()
        elif self.agent_config_type == AgentConfigType.FALLBACK and self.fallback_agent_config:
            self.fallback_agent.memory.reset()
        else:
            raise ValueError(f"Invalid agent config type {self.agent_config_type}")

    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_config.agent_type != other.agent_config.agent_type:
            print(
                f"Comparison failed: agent_type differs. (self.agent_config.agent_type: {self.agent_config.agent_type},"
                f" other.agent_config.agent_type: {other.agent_config.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 memory
        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(),
        validate_tools: bool = False,
        fallback_agent_config: Optional[AgentConfig] = None,
        chat_history: Optional[list[Tuple[str, str]]] = None,
        workflow_cls: Workflow = None,
        workflow_timeout: int = 120,
    ) -> "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.
            fallback_agent_config (AgentConfig, optional): The fallback configuration of the agent.
            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.
            workflow_cls (Workflow, optional): The workflow class to be used with run(). Defaults to None.
            workflow_timeout (int, optional): The timeout for the workflow in seconds. Defaults to 120.

        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,
            validate_tools=validate_tools,
            fallback_agent_config=fallback_agent_config,
            workflow_cls = workflow_cls, workflow_timeout = workflow_timeout,
        )

    @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,
        agent_config: AgentConfig = AgentConfig(),
        fallback_agent_config: Optional[AgentConfig] = None,
        chat_history: Optional[list[Tuple[str, str]]] = 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
            agent_config (AgentConfig, optional): The configuration of the agent.
            fallback_agent_config (AgentConfig, optional): The fallback configuration of the agent.
            chat_history (Tuple[str, str], optional): A list of user/agent chat pairs to initialize the agent memory.
            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,
            agent_config=agent_config,
            fallback_agent_config=fallback_agent_config,
            chat_history=chat_history,
        )

    def _switch_agent_config(self) -> None:
        """"
        Switch the configuration type of the agent.
        This function is called automatically to switch the agent configuration if the current configuration fails.
        """
        if self.agent_config_type == AgentConfigType.DEFAULT:
            self.agent_config_type = AgentConfigType.FALLBACK
        else:
            self.agent_config_type = AgentConfigType.DEFAULT

    def report(self, detailed: bool = False) -> None:
        """
        Get a report from the agent.

        Args:
            detailed (bool, optional): Whether to include detailed information. Defaults to False.

        Returns:
            str: The report from the agent.
        """
        print("Vectara agentic Report:")
        print(f"Agent Type = {self.agent_config.agent_type}")
        print(f"Topic = {self._topic}")
        print("Tools:")
        for tool in self.tools:
            if hasattr(tool, 'metadata'):
                if detailed:
                    print(f"- {tool.metadata.description}")
                else:
                    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,
        }

    def _get_current_agent(self):
        return self.agent if self.agent_config_type == AgentConfigType.DEFAULT else self.fallback_agent

    def _get_current_agent_type(self):
        return (
            self.agent_config.agent_type if self.agent_config_type == AgentConfigType.DEFAULT
            else self.fallback_agent_config.agent_type
        )

    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_type = self._get_current_agent_type()
        if agent_type != AgentType.LATS:
            return

        agent = self._get_current_agent()
        agent_response.response = str(agent.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))

    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.
        """
        max_attempts = 4 if self.fallback_agent_config else 2
        attempt = 0
        orig_llm = self.llm.metadata.model_name
        last_error = None
        while attempt < max_attempts:
            try:
                current_agent = self._get_current_agent()
                agent_response = await current_agent.achat(prompt)
                await self._aformat_for_lats(prompt, agent_response)
                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:
                last_error = e
                if attempt >= 2:
                    if self.verbose:
                        print(f"LLM call failed on attempt {attempt}. Switching agent configuration.")
                    self._switch_agent_config()
                time.sleep(1)
                attempt += 1

        return AgentResponse(
            response=(
                f"For {orig_llm} LLM - failure can't be resolved after "
                f"{max_attempts} attempts ({last_error}."
            )
        )

    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))

    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.
        """
        max_attempts = 4 if self.fallback_agent_config else 2
        attempt = 0
        orig_llm = self.llm.metadata.model_name
        while attempt < max_attempts:
            try:
                current_agent = self._get_current_agent()
                agent_response = await current_agent.astream_chat(prompt)
                original_async_response_gen = agent_response.async_response_gen

                # Define a wrapper to preserve streaming behavior while executing post-stream logic.
                async def _stream_response_wrapper():
                    async for token in original_async_response_gen():
                        yield token  # Yield tokens as they are generated
                    # Post-streaming additional logic:
                    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 the generator
                return agent_response

            except Exception as e:
                last_error = e
                if attempt >= 2:
                    if self.verbose:
                        print(f"LLM call failed on attempt {attempt}. Switching agent configuration.")
                    self._switch_agent_config()
                time.sleep(1)
                attempt += 1

        return AgentResponse(
            response=(
                f"For {orig_llm} LLM - failure can't be resolved after "
                f"{max_attempts} attempts ({last_error})."
            )
        )

    #
    # run() method for running a workflow
    # workflow will always get these arguments in the StartEvent: agent, tools, llm, verbose
    # the inputs argument comes from the call to run()
    #
    async def run(
        self,
        inputs: Any,
        verbose: bool = False
    ) -> Any:
        """
        Run a workflow using the agent.
        workflow class must be provided in the agent constructor.
        Args:
            inputs (Any): The inputs to the workflow.
            verbose (bool, optional): Whether to print verbose output. Defaults to False.
        Returns:
            Any: The output of the workflow.
        """
        # Create workflow
        if self.workflow_cls:
            workflow = self.workflow_cls(timeout=self.workflow_timeout, verbose=verbose)
        else:
            raise ValueError("Workflow is not defined.")

        # Validate inputs is in the form of workflow.InputsModel
        if not isinstance(inputs, self.workflow_cls.InputsModel):
            raise ValueError(f"Inputs must be an instance of {workflow.InputsModel}.")

        # run workflow
        result = await workflow.run(
            agent=self,
            tools=self.tools,
            llm=self.llm,
            verbose=verbose,
            inputs=inputs,
        )

        # return output in the form of workflow.OutputsModel
        try:
            output = workflow.OutputsModel.model_validate(result)
        except ValidationError as e:
            raise ValueError(f"Failed to map workflow output to model: {e}") from e

        return output

    #
    # 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:
            if hasattr(tool.metadata, "fn_schema"):
                fn_schema_cls = tool.metadata.fn_schema
                fn_schema_serialized = {
                    "schema": (
                        fn_schema_cls.model_json_schema()
                        if hasattr(fn_schema_cls, "model_json_schema")
                        else None
                    ),
                    "metadata": {
                        "module": fn_schema_cls.__module__,
                        "class": fn_schema_cls.__name__,
                    }
                }
            else:
                fn_schema_serialized = None

            tool_dict = {
                "tool_type": tool.metadata.tool_type.value,
                "name": tool.metadata.name,
                "description": tool.metadata.description,
                "fn": pickle.dumps(getattr(tool, 'fn', None)).decode("latin-1") if getattr(tool, 'fn', None) else None,
                "async_fn": pickle.dumps(getattr(tool, 'async_fn', None)).decode("latin-1")
                if getattr(tool, 'async_fn', None) else None,
                "fn_schema": fn_schema_serialized,
            }
            tool_info.append(tool_dict)

        return {
            "agent_type": self.agent_config.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(),
            "fallback_agent": self.fallback_agent_config.to_dict() if self.fallback_agent_config else None,
            "workflow_cls": self.workflow_cls if self.workflow_cls else None,
        }

    @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"])
        fallback_agent_config = (
            AgentConfig.from_dict(data["fallback_agent_config"])
            if data.get("fallback_agent_config")
            else None
        )
        tools = []

        for tool_data in data["tools"]:
            # Recreate the dynamic model using the schema info
            if tool_data.get("fn_schema"):
                schema_info = tool_data["fn_schema"]
                try:
                    module_name = schema_info["metadata"]["module"]
                    class_name = schema_info["metadata"]["class"]
                    mod = importlib.import_module(module_name)
                    fn_schema_cls = getattr(mod, class_name)
                    query_args_model = fn_schema_cls
                except Exception:
                    # Fallback: rebuild using the JSON schema
                    field_definitions = {}
                    for field, values in schema_info.get("schema", {}).get("properties", {}).items():
                        field_type = get_field_type(values)
                        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(
                        schema_info.get("schema", {}).get("title", "QueryArgs"),
                        **field_definitions
                    )
            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"],
            fallback_agent_config=fallback_agent_config,
            workflow_cls=data["workflow_cls"],
        )
        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, use_structured_planning=False, update_func=None, agent_progress_callback=None, query_logging_callback=None, agent_config=None, fallback_agent_config=None, chat_history=None, validate_tools=False, workflow_cls=None, workflow_timeout=120)

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.
use_structured_planning (bool, optional)
    Whether or not we want to wrap the agent with LlamaIndex StructuredPlannerAgent.
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.
fallback_agent_config (AgentConfig, optional): The fallback configuration of the agent.
    This config is used when the main agent config fails multiple times.
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.
workflow_cls (Workflow, optional): The workflow class to be used with run(). Defaults to None.
workflow_timeout (int, optional): The timeout for the workflow in seconds. Defaults to 120.

Source code in vectara_agentic/agent.py

def __init__(
    self,
    tools: list[FunctionTool],
    topic: str = "general",
    custom_instructions: str = "",
    verbose: bool = True,
    use_structured_planning: bool = False,
    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,
    fallback_agent_config: Optional[AgentConfig] = None,
    chat_history: Optional[list[Tuple[str, str]]] = None,
    validate_tools: bool = False,
    workflow_cls: Workflow = None,
    workflow_timeout: int = 120,
) -> 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.
        use_structured_planning (bool, optional)
            Whether or not we want to wrap the agent with LlamaIndex StructuredPlannerAgent.
        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.
        fallback_agent_config (AgentConfig, optional): The fallback configuration of the agent.
            This config is used when the main agent config fails multiple times.
        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.
        workflow_cls (Workflow, optional): The workflow class to be used with run(). Defaults to None.
        workflow_timeout (int, optional): The timeout for the workflow in seconds. Defaults to 120.
    """
    self.agent_config = agent_config or AgentConfig()
    self.agent_config_type = AgentConfigType.DEFAULT
    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.agent_type = self.agent_config.agent_type
    self.use_structured_planning = use_structured_planning
    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

    self.workflow_cls = workflow_cls
    self.workflow_timeout = workflow_timeout

    # 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.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)

    # Set up main agent and fallback agent
    self.agent = self._create_agent(self.agent_config, callback_manager)
    self.fallback_agent_config = fallback_agent_config
    if self.fallback_agent_config:
        self.fallback_agent = self._create_agent(self.fallback_agent_config, callback_manager)
    else:
        self.fallback_agent_config = None

    # Setup observability
    try:
        self.observability_enabled = setup_observer(self.agent_config, self.verbose)
    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

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.
    """
    max_attempts = 4 if self.fallback_agent_config else 2
    attempt = 0
    orig_llm = self.llm.metadata.model_name
    last_error = None
    while attempt < max_attempts:
        try:
            current_agent = self._get_current_agent()
            agent_response = await current_agent.achat(prompt)
            await self._aformat_for_lats(prompt, agent_response)
            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:
            last_error = e
            if attempt >= 2:
                if self.verbose:
                    print(f"LLM call failed on attempt {attempt}. Switching agent configuration.")
                self._switch_agent_config()
            time.sleep(1)
            attempt += 1

    return AgentResponse(
        response=(
            f"For {orig_llm} LLM - failure can't be resolved after "
            f"{max_attempts} attempts ({last_error}."
        )
    )

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

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.
    """
    max_attempts = 4 if self.fallback_agent_config else 2
    attempt = 0
    orig_llm = self.llm.metadata.model_name
    while attempt < max_attempts:
        try:
            current_agent = self._get_current_agent()
            agent_response = await current_agent.astream_chat(prompt)
            original_async_response_gen = agent_response.async_response_gen

            # Define a wrapper to preserve streaming behavior while executing post-stream logic.
            async def _stream_response_wrapper():
                async for token in original_async_response_gen():
                    yield token  # Yield tokens as they are generated
                # Post-streaming additional logic:
                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 the generator
            return agent_response

        except Exception as e:
            last_error = e
            if attempt >= 2:
                if self.verbose:
                    print(f"LLM call failed on attempt {attempt}. Switching agent configuration.")
                self._switch_agent_config()
            time.sleep(1)
            attempt += 1

    return AgentResponse(
        response=(
            f"For {orig_llm} LLM - failure can't be resolved after "
            f"{max_attempts} attempts ({last_error})."
        )
    )

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.
    """
    if self.agent_config_type == AgentConfigType.DEFAULT:
        self.agent.memory.reset()
    elif self.agent_config_type == AgentConfigType.FALLBACK and self.fallback_agent_config:
        self.fallback_agent.memory.reset()
    else:
        raise ValueError(f"Invalid agent config type {self.agent_config_type}")

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, agent_config=AgentConfig(), fallback_agent_config=None, chat_history=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
agent_config	AgentConfig	The configuration of the agent.	AgentConfig()
fallback_agent_config	AgentConfig	The fallback configuration of the agent.	None
chat_history	Tuple[str, str]	A list of user/agent chat pairs to initialize the agent memory.	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,
    agent_config: AgentConfig = AgentConfig(),
    fallback_agent_config: Optional[AgentConfig] = None,
    chat_history: Optional[list[Tuple[str, str]]] = 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
        agent_config (AgentConfig, optional): The configuration of the agent.
        fallback_agent_config (AgentConfig, optional): The fallback configuration of the agent.
        chat_history (Tuple[str, str], optional): A list of user/agent chat pairs to initialize the agent memory.
        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,
        agent_config=agent_config,
        fallback_agent_config=fallback_agent_config,
        chat_history=chat_history,
    )

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"])
    fallback_agent_config = (
        AgentConfig.from_dict(data["fallback_agent_config"])
        if data.get("fallback_agent_config")
        else None
    )
    tools = []

    for tool_data in data["tools"]:
        # Recreate the dynamic model using the schema info
        if tool_data.get("fn_schema"):
            schema_info = tool_data["fn_schema"]
            try:
                module_name = schema_info["metadata"]["module"]
                class_name = schema_info["metadata"]["class"]
                mod = importlib.import_module(module_name)
                fn_schema_cls = getattr(mod, class_name)
                query_args_model = fn_schema_cls
            except Exception:
                # Fallback: rebuild using the JSON schema
                field_definitions = {}
                for field, values in schema_info.get("schema", {}).get("properties", {}).items():
                    field_type = get_field_type(values)
                    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(
                    schema_info.get("schema", {}).get("title", "QueryArgs"),
                    **field_definitions
                )
        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"],
        fallback_agent_config=fallback_agent_config,
        workflow_cls=data["workflow_cls"],
    )
    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(), validate_tools=False, fallback_agent_config=None, chat_history=None, workflow_cls=None, workflow_timeout=120) 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.
fallback_agent_config (AgentConfig, optional): The fallback configuration of the agent.
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.
workflow_cls (Workflow, optional): The workflow class to be used with run(). Defaults to None.
workflow_timeout (int, optional): The timeout for the workflow in seconds. Defaults to 120.

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(),
    validate_tools: bool = False,
    fallback_agent_config: Optional[AgentConfig] = None,
    chat_history: Optional[list[Tuple[str, str]]] = None,
    workflow_cls: Workflow = None,
    workflow_timeout: int = 120,
) -> "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.
        fallback_agent_config (AgentConfig, optional): The fallback configuration of the agent.
        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.
        workflow_cls (Workflow, optional): The workflow class to be used with run(). Defaults to None.
        workflow_timeout (int, optional): The timeout for the workflow in seconds. Defaults to 120.

    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,
        validate_tools=validate_tools,
        fallback_agent_config=fallback_agent_config,
        workflow_cls = workflow_cls, workflow_timeout = workflow_timeout,
    )

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(detailed=False)

Get a report from the agent.

Parameters:

Name	Type	Description	Default
detailed	bool	Whether to include detailed information. Defaults to False.	False

Returns:

Name	Type	Description
str	None	The report from the agent.

Source code in vectara_agentic/agent.py

def report(self, detailed: bool = False) -> None:
    """
    Get a report from the agent.

    Args:
        detailed (bool, optional): Whether to include detailed information. Defaults to False.

    Returns:
        str: The report from the agent.
    """
    print("Vectara agentic Report:")
    print(f"Agent Type = {self.agent_config.agent_type}")
    print(f"Topic = {self._topic}")
    print("Tools:")
    for tool in self.tools:
        if hasattr(tool, 'metadata'):
            if detailed:
                print(f"- {tool.metadata.description}")
            else:
                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}")

run(inputs, verbose=False) async

Run a workflow using the agent. workflow class must be provided in the agent constructor. Args: inputs (Any): The inputs to the workflow. verbose (bool, optional): Whether to print verbose output. Defaults to False. Returns: Any: The output of the workflow.

Source code in vectara_agentic/agent.py

async def run(
    self,
    inputs: Any,
    verbose: bool = False
) -> Any:
    """
    Run a workflow using the agent.
    workflow class must be provided in the agent constructor.
    Args:
        inputs (Any): The inputs to the workflow.
        verbose (bool, optional): Whether to print verbose output. Defaults to False.
    Returns:
        Any: The output of the workflow.
    """
    # Create workflow
    if self.workflow_cls:
        workflow = self.workflow_cls(timeout=self.workflow_timeout, verbose=verbose)
    else:
        raise ValueError("Workflow is not defined.")

    # Validate inputs is in the form of workflow.InputsModel
    if not isinstance(inputs, self.workflow_cls.InputsModel):
        raise ValueError(f"Inputs must be an instance of {workflow.InputsModel}.")

    # run workflow
    result = await workflow.run(
        agent=self,
        tools=self.tools,
        llm=self.llm,
        verbose=verbose,
        inputs=inputs,
    )

    # return output in the form of workflow.OutputsModel
    try:
        output = workflow.OutputsModel.model_validate(result)
    except ValidationError as e:
        raise ValueError(f"Failed to map workflow output to model: {e}") from e

    return output

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:
        if hasattr(tool.metadata, "fn_schema"):
            fn_schema_cls = tool.metadata.fn_schema
            fn_schema_serialized = {
                "schema": (
                    fn_schema_cls.model_json_schema()
                    if hasattr(fn_schema_cls, "model_json_schema")
                    else None
                ),
                "metadata": {
                    "module": fn_schema_cls.__module__,
                    "class": fn_schema_cls.__name__,
                }
            }
        else:
            fn_schema_serialized = None

        tool_dict = {
            "tool_type": tool.metadata.tool_type.value,
            "name": tool.metadata.name,
            "description": tool.metadata.description,
            "fn": pickle.dumps(getattr(tool, 'fn', None)).decode("latin-1") if getattr(tool, 'fn', None) else None,
            "async_fn": pickle.dumps(getattr(tool, 'async_fn', None)).decode("latin-1")
            if getattr(tool, 'async_fn', None) else None,
            "fn_schema": fn_schema_serialized,
        }
        tool_info.append(tool_dict)

    return {
        "agent_type": self.agent_config.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(),
        "fallback_agent": self.fallback_agent_config.to_dict() if self.fallback_agent_config else None,
        "workflow_cls": self.workflow_cls if self.workflow_cls else None,
    }

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,
    }

AgentConfig dataclass

Centralized configuration for the Vectara Agentic utilities.

Each field can default to either a hard-coded value or an environment variable. For example, if you have environment variables you want to fall back on, you can default to them here.

Source code in vectara_agentic/agent_config.py

@dataclass(eq=True, frozen=True)
class AgentConfig:
    """
    Centralized configuration for the Vectara Agentic utilities.

    Each field can default to either a hard-coded value or an environment
    variable. For example, if you have environment variables you want to
    fall back on, you can default to them here.
    """

    # Agent type
    agent_type: AgentType = field(
        default_factory=lambda: AgentType(
            os.getenv("VECTARA_AGENTIC_AGENT_TYPE", AgentType.OPENAI.value)
        )
    )

    # Main LLM provider & model name
    main_llm_provider: ModelProvider = field(
        default_factory=lambda: ModelProvider(
            os.getenv("VECTARA_AGENTIC_MAIN_LLM_PROVIDER", ModelProvider.OPENAI.value)
        )
    )

    main_llm_model_name: str = field(
        default_factory=lambda: os.getenv("VECTARA_AGENTIC_MAIN_MODEL_NAME", "")
    )

    # Tool LLM provider & model name
    tool_llm_provider: ModelProvider = field(
        default_factory=lambda: ModelProvider(
            os.getenv("VECTARA_AGENTIC_TOOL_LLM_PROVIDER", ModelProvider.OPENAI.value)
        )
    )
    tool_llm_model_name: str = field(
        default_factory=lambda: os.getenv("VECTARA_AGENTIC_TOOL_MODEL_NAME", "")
    )

    # Params for Private LLM endpoint if used
    private_llm_api_base: str = field(
        default_factory=lambda: os.getenv("VECTARA_AGENTIC_PRIVATE_LLM_API_BASE",
                                          "http://private-endpoint.company.com:5000/v1")
    )
    private_llm_api_key: str = field(
        default_factory=lambda: os.getenv("VECTARA_AGENTIC_PRIVATE_LLM_API_KEY", "<private-api-key>")
    )

    # Observer
    observer: ObserverType = field(
        default_factory=lambda: ObserverType(
            os.getenv("VECTARA_AGENTIC_OBSERVER_TYPE", "NO_OBSERVER")
        )
    )

    # Endpoint API key
    endpoint_api_key: str = field(
        default_factory=lambda: os.getenv("VECTARA_AGENTIC_API_KEY", "dev-api-key")
    )

    # max reasoning steps
    # used for both OpenAI and React Agent types
    max_reasoning_steps: int = field(
        default_factory=lambda: int(os.getenv("VECTARA_AGENTIC_MAX_REASONING_STEPS", "50"))
    )

    def __post_init__(self):
        # Use object.__setattr__ since the dataclass is frozen
        if isinstance(self.agent_type, str):
            object.__setattr__(self, "agent_type", AgentType(self.agent_type))
        if isinstance(self.main_llm_provider, str):
            object.__setattr__(self, "main_llm_provider", ModelProvider(self.main_llm_provider))
        if isinstance(self.tool_llm_provider, str):
            object.__setattr__(self, "tool_llm_provider", ModelProvider(self.tool_llm_provider))
        if isinstance(self.observer, str):
            object.__setattr__(self, "observer", ObserverType(self.observer))

    def to_dict(self) -> dict:
        """
        Convert the AgentConfig to a dictionary.
        """
        return {
            "agent_type": self.agent_type.value,
            "main_llm_provider": self.main_llm_provider.value,
            "main_llm_model_name": self.main_llm_model_name,
            "tool_llm_provider": self.tool_llm_provider.value,
            "tool_llm_model_name": self.tool_llm_model_name,
            "observer": self.observer.value,
            "endpoint_api_key": self.endpoint_api_key,
            "max_reasoning_steps": self.max_reasoning_steps
        }

    @classmethod
    def from_dict(cls, config_dict: dict) -> "AgentConfig":
        """
        Create an AgentConfig from a dictionary.
        """
        return cls(
            agent_type=AgentType(config_dict["agent_type"]),
            main_llm_provider=ModelProvider(config_dict["main_llm_provider"]),
            main_llm_model_name=config_dict["main_llm_model_name"],
            tool_llm_provider=ModelProvider(config_dict["tool_llm_provider"]),
            tool_llm_model_name=config_dict["tool_llm_model_name"],
            observer=ObserverType(config_dict["observer"]),
            endpoint_api_key=config_dict["endpoint_api_key"],
            max_reasoning_steps=config_dict["max_reasoning_steps"]
        )

from_dict(config_dict) classmethod

Create an AgentConfig from a dictionary.

Source code in vectara_agentic/agent_config.py

@classmethod
def from_dict(cls, config_dict: dict) -> "AgentConfig":
    """
    Create an AgentConfig from a dictionary.
    """
    return cls(
        agent_type=AgentType(config_dict["agent_type"]),
        main_llm_provider=ModelProvider(config_dict["main_llm_provider"]),
        main_llm_model_name=config_dict["main_llm_model_name"],
        tool_llm_provider=ModelProvider(config_dict["tool_llm_provider"]),
        tool_llm_model_name=config_dict["tool_llm_model_name"],
        observer=ObserverType(config_dict["observer"]),
        endpoint_api_key=config_dict["endpoint_api_key"],
        max_reasoning_steps=config_dict["max_reasoning_steps"]
    )

to_dict()

Convert the AgentConfig to a dictionary.

Source code in vectara_agentic/agent_config.py

def to_dict(self) -> dict:
    """
    Convert the AgentConfig to a dictionary.
    """
    return {
        "agent_type": self.agent_type.value,
        "main_llm_provider": self.main_llm_provider.value,
        "main_llm_model_name": self.main_llm_model_name,
        "tool_llm_provider": self.tool_llm_provider.value,
        "tool_llm_model_name": self.tool_llm_model_name,
        "observer": self.observer.value,
        "endpoint_api_key": self.endpoint_api_key,
        "max_reasoning_steps": self.max_reasoning_steps
    }

AgentStatusType

Bases: Enum

Enumeration for different types of agent statuses.

Source code in vectara_agentic/types.py

class AgentStatusType(Enum):
    """Enumeration for different types of agent statuses."""

    AGENT_UPDATE = "agent_update"
    TOOL_CALL = "tool_call"
    TOOL_OUTPUT = "tool_output"
    AGENT_STEP = "agent_step"

AgentType

Bases: Enum

Enumeration for different types of agents.

Source code in vectara_agentic/types.py

class AgentType(Enum):
    """Enumeration for different types of agents."""

    REACT = "REACT"
    OPENAI = "OPENAI"
    FUNCTION_CALLING = "FUNCTION_CALLING"
    LLMCOMPILER = "LLMCOMPILER"
    LATS = "LATS"

LLMRole

Bases: Enum

Enumeration for different types of LLM roles.

Source code in vectara_agentic/types.py

class LLMRole(Enum):
    """Enumeration for different types of LLM roles."""

    MAIN = "MAIN"
    TOOL = "TOOL"

ModelProvider

Bases: Enum

Enumeration for different types of model providers.

Source code in vectara_agentic/types.py

class ModelProvider(Enum):
    """Enumeration for different types of model providers."""

    OPENAI = "OPENAI"
    ANTHROPIC = "ANTHROPIC"
    TOGETHER = "TOGETHER"
    GROQ = "GROQ"
    FIREWORKS = "FIREWORKS"
    COHERE = "COHERE"
    GEMINI = "GEMINI"
    BEDROCK = "BEDROCK"
    PRIVATE = "PRIVATE"

ObserverType

Bases: Enum

Enumeration for different types of observability integrations.

Source code in vectara_agentic/types.py

class ObserverType(Enum):
    """Enumeration for different types of observability integrations."""

    NO_OBSERVER = "NO_OBSERVER"
    ARIZE_PHOENIX = "ARIZE_PHOENIX"

ToolType

Bases: Enum

Enumeration for different types of tools.

Source code in vectara_agentic/types.py

class ToolType(Enum):
    """Enumeration for different types of tools."""
    QUERY = "query"
    ACTION = "action"

ToolsCatalog

A curated set of tools for vectara-agentic

Source code in vectara_agentic/tools_catalog.py

class ToolsCatalog:
    """
    A curated set of tools for vectara-agentic
    """
    def __init__(self, agent_config: AgentConfig):
        self.agent_config = agent_config

    @remove_self_from_signature
    def summarize_text(
        self,
        text: str = Field(description="the original text."),
        expertise: str = Field(
            description="the expertise to apply to the summarization.",
        ),
    ) -> str:
        """
        This is a helper tool.
        Use this tool to summarize text using a given expertise
        with no more than summary_max_length characters.

        Args:
            text (str): The original text.
            expertise (str): The expertise to apply to the summarization.

        Returns:
            str: The summarized text.
        """
        if not isinstance(expertise, str):
            return "Please provide a valid string for expertise."
        if not isinstance(text, str):
            return "Please provide a valid string for text."
        expertise = "general" if len(expertise) < 3 else expertise.lower()
        prompt = (
            f"As an expert in {expertise}, summarize the provided text "
            "into a concise summary.\n"
            f"Original text: {text}\nSummary:"
        )
        llm = get_llm(LLMRole.TOOL, config=self.agent_config)
        response = llm.complete(prompt)
        return response.text

    @remove_self_from_signature
    def rephrase_text(
        self,
        text: str = Field(description="the original text."),
        instructions: str = Field(description="the specific instructions for how to rephrase the text."),
    ) -> str:
        """
        This is a helper tool.
        Use this tool to rephrase the text according to the provided instructions.
        For example, instructions could be "as a 5 year old would say it."

        Args:
            text (str): The original text.
            instructions (str): The specific instructions for how to rephrase the text.

        Returns:
            str: The rephrased text.
        """
        prompt = (
            f"Rephrase the provided text according to the following instructions: {instructions}.\n"
            "If the input is Markdown, keep the output in Markdown as well.\n"
            f"Original text: {text}\nRephrased text:"
        )
        llm = get_llm(LLMRole.TOOL, config=self.agent_config)
        response = llm.complete(prompt)
        return response.text

    @remove_self_from_signature
    def critique_text(
        self,
        text: str = Field(description="the original text."),
        role: str = Field(default=None, description="the role of the person providing critique."),
        point_of_view: str = Field(default=None, description="the point of view with which to provide critique."),
    ) -> str:
        """
        This is a helper tool.
        Critique the text from the specified point of view.

        Args:
            text (str): The original text.
            role (str): The role of the person providing critique.
            point_of_view (str): The point of view with which to provide critique.

        Returns:
            str: The critique of the text.
        """
        if role:
            prompt = f"As a {role}, critique the provided text from the point of view of {point_of_view}."
        else:
            prompt = f"Critique the provided text from the point of view of {point_of_view}."
        prompt += "\nStructure the critique as bullet points.\n"
        prompt += f"Original text: {text}\nCritique:"
        llm = get_llm(LLMRole.TOOL, config=self.agent_config)
        response = llm.complete(prompt)
        return response.text

critique_text(text=Field(description='the original text.'), role=Field(default=None, description='the role of the person providing critique.'), point_of_view=Field(default=None, description='the point of view with which to provide critique.'))

This is a helper tool. Critique the text from the specified point of view.

Parameters:

Name	Type	Description	Default
text	str	The original text.	Field(description='the original text.')
role	str	The role of the person providing critique.	Field(default=None, description='the role of the person providing critique.')
point_of_view	str	The point of view with which to provide critique.	Field(default=None, description='the point of view with which to provide critique.')

Returns:

Name	Type	Description
str	str	The critique of the text.

Source code in vectara_agentic/tools_catalog.py

@remove_self_from_signature
def critique_text(
    self,
    text: str = Field(description="the original text."),
    role: str = Field(default=None, description="the role of the person providing critique."),
    point_of_view: str = Field(default=None, description="the point of view with which to provide critique."),
) -> str:
    """
    This is a helper tool.
    Critique the text from the specified point of view.

    Args:
        text (str): The original text.
        role (str): The role of the person providing critique.
        point_of_view (str): The point of view with which to provide critique.

    Returns:
        str: The critique of the text.
    """
    if role:
        prompt = f"As a {role}, critique the provided text from the point of view of {point_of_view}."
    else:
        prompt = f"Critique the provided text from the point of view of {point_of_view}."
    prompt += "\nStructure the critique as bullet points.\n"
    prompt += f"Original text: {text}\nCritique:"
    llm = get_llm(LLMRole.TOOL, config=self.agent_config)
    response = llm.complete(prompt)
    return response.text

rephrase_text(text=Field(description='the original text.'), instructions=Field(description='the specific instructions for how to rephrase the text.'))

This is a helper tool. Use this tool to rephrase the text according to the provided instructions. For example, instructions could be "as a 5 year old would say it."

Parameters:

Name	Type	Description	Default
text	str	The original text.	Field(description='the original text.')
instructions	str	The specific instructions for how to rephrase the text.	Field(description='the specific instructions for how to rephrase the text.')

Returns:

Name	Type	Description
str	str	The rephrased text.

Source code in vectara_agentic/tools_catalog.py

@remove_self_from_signature
def rephrase_text(
    self,
    text: str = Field(description="the original text."),
    instructions: str = Field(description="the specific instructions for how to rephrase the text."),
) -> str:
    """
    This is a helper tool.
    Use this tool to rephrase the text according to the provided instructions.
    For example, instructions could be "as a 5 year old would say it."

    Args:
        text (str): The original text.
        instructions (str): The specific instructions for how to rephrase the text.

    Returns:
        str: The rephrased text.
    """
    prompt = (
        f"Rephrase the provided text according to the following instructions: {instructions}.\n"
        "If the input is Markdown, keep the output in Markdown as well.\n"
        f"Original text: {text}\nRephrased text:"
    )
    llm = get_llm(LLMRole.TOOL, config=self.agent_config)
    response = llm.complete(prompt)
    return response.text

summarize_text(text=Field(description='the original text.'), expertise=Field(description='the expertise to apply to the summarization.'))

This is a helper tool. Use this tool to summarize text using a given expertise with no more than summary_max_length characters.

Parameters:

Name	Type	Description	Default
text	str	The original text.	Field(description='the original text.')
expertise	str	The expertise to apply to the summarization.	Field(description='the expertise to apply to the summarization.')

Returns:

Name	Type	Description
str	str	The summarized text.

Source code in vectara_agentic/tools_catalog.py

@remove_self_from_signature
def summarize_text(
    self,
    text: str = Field(description="the original text."),
    expertise: str = Field(
        description="the expertise to apply to the summarization.",
    ),
) -> str:
    """
    This is a helper tool.
    Use this tool to summarize text using a given expertise
    with no more than summary_max_length characters.

    Args:
        text (str): The original text.
        expertise (str): The expertise to apply to the summarization.

    Returns:
        str: The summarized text.
    """
    if not isinstance(expertise, str):
        return "Please provide a valid string for expertise."
    if not isinstance(text, str):
        return "Please provide a valid string for text."
    expertise = "general" if len(expertise) < 3 else expertise.lower()
    prompt = (
        f"As an expert in {expertise}, summarize the provided text "
        "into a concise summary.\n"
        f"Original text: {text}\nSummary:"
    )
    llm = get_llm(LLMRole.TOOL, config=self.agent_config)
    response = llm.complete(prompt)
    return response.text

ToolsFactory

A factory class for creating agent tools.

Source code in vectara_agentic/tools.py

class ToolsFactory:
    """
    A factory class for creating agent tools.
    """

    def __init__(self, agent_config: AgentConfig = None) -> None:
        self.agent_config = agent_config

    def create_tool(
        self, function: Callable, tool_type: ToolType = ToolType.QUERY
    ) -> VectaraTool:
        """
        Create a tool from a function.

        Args:
            function (Callable): a function to convert into a tool.
            tool_type (ToolType): the type of tool.

        Returns:
            VectaraTool: A VectaraTool object.
        """
        return VectaraTool.from_defaults(tool_type=tool_type, fn=function)

    def get_llama_index_tools(
        self,
        tool_package_name: str,
        tool_spec_name: str,
        tool_name_prefix: str = "",
        **kwargs: dict,
    ) -> List[VectaraTool]:
        """
        Get a tool from the llama_index hub.

        Args:
            tool_package_name (str): The name of the tool package.
            tool_spec_name (str): The name of the tool spec.
            tool_name_prefix (str, optional): The prefix to add to the tool names (added to every tool in the spec).
            kwargs (dict): The keyword arguments to pass to the tool constructor (see Hub for tool specific details).

        Returns:
            List[VectaraTool]: A list of VectaraTool objects.
        """
        # Dynamically install and import the module
        if tool_package_name not in LI_packages:
            raise ValueError(
                f"Tool package {tool_package_name} from LlamaIndex not supported by Vectara-agentic."
            )

        module_name = f"llama_index.tools.{tool_package_name}"
        module = importlib.import_module(module_name)

        # Get the tool spec class or function from the module
        tool_spec = getattr(module, tool_spec_name)
        func_type = LI_packages[tool_package_name]
        tools = tool_spec(**kwargs).to_tool_list()
        vtools = []
        for tool in tools:
            if len(tool_name_prefix) > 0:
                tool.metadata.name = tool_name_prefix + "_" + tool.metadata.name
            if isinstance(func_type, dict):
                if tool_spec_name not in func_type.keys():
                    raise ValueError(
                        f"Tool spec {tool_spec_name} not found in package {tool_package_name}."
                    )
                tool_type = func_type[tool_spec_name]
            else:
                tool_type = func_type
            vtool = VectaraTool(
                tool_type=tool_type,
                fn=tool.fn,
                metadata=tool.metadata,
                async_fn=tool.async_fn,
            )
            vtools.append(vtool)
        return vtools

    def standard_tools(self) -> List[FunctionTool]:
        """
        Create a list of standard tools.
        """
        tc = ToolsCatalog(self.agent_config)
        return [
            self.create_tool(tool)
            for tool in [tc.summarize_text, tc.rephrase_text, tc.critique_text]
        ]

    def guardrail_tools(self) -> List[FunctionTool]:
        """
        Create a list of guardrail tools to avoid controversial topics.
        """
        return [self.create_tool(get_bad_topics)]

    def financial_tools(self):
        """
        Create a list of financial tools.
        """
        return self.get_llama_index_tools(
            tool_package_name="yahoo_finance", tool_spec_name="YahooFinanceToolSpec"
        )

    def legal_tools(self) -> List[FunctionTool]:
        """
        Create a list of legal tools.
        """

        def summarize_legal_text(
            text: str = Field(description="the original text."),
        ) -> str:
            """
            Use this tool to summarize legal text with no more than summary_max_length characters.
            """
            tc = ToolsCatalog(self.agent_config)
            return tc.summarize_text(text, expertise="law")

        def critique_as_judge(
            text: str = Field(description="the original text."),
        ) -> str:
            """
            Critique the legal document.
            """
            tc = ToolsCatalog(self.agent_config)
            return tc.critique_text(
                text,
                role="judge",
                point_of_view="""
                an experienced judge evaluating a legal document to provide areas of concern
                or that may require further legal scrutiny or legal argument.
                """,
            )

        return [
            self.create_tool(tool) for tool in [summarize_legal_text, critique_as_judge]
        ]

    def database_tools(
        self,
        tool_name_prefix: str = "",
        content_description: Optional[str] = None,
        sql_database: Optional[SQLDatabase] = None,
        scheme: Optional[str] = None,
        host: str = "localhost",
        port: str = "5432",
        user: str = "postgres",
        password: str = "Password",
        dbname: str = "postgres",
        max_rows: int = 1000,
    ) -> List[VectaraTool]:
        """
        Returns a list of database tools.

        Args:

            tool_name_prefix (str, optional): The prefix to add to the tool names. Defaults to "".
            content_description (str, optional): The content description for the database. Defaults to None.
            sql_database (SQLDatabase, optional): The SQLDatabase object. Defaults to None.
            scheme (str, optional): The database scheme. Defaults to None.
            host (str, optional): The database host. Defaults to "localhost".
            port (str, optional): The database port. Defaults to "5432".
            user (str, optional): The database user. Defaults to "postgres".
            password (str, optional): The database password. Defaults to "Password".
            dbname (str, optional): The database name. Defaults to "postgres".
               You must specify either the sql_database object or the scheme, host, port, user, password, and dbname.
            max_rows (int, optional): if specified, instructs the load_data tool to never return more than max_rows
               rows. Defaults to 1000.

        Returns:
            List[VectaraTool]: A list of VectaraTool objects.
        """
        if sql_database:
            dbt = DatabaseTools(
                tool_name_prefix=tool_name_prefix,
                sql_database=sql_database,
                max_rows=max_rows,
            )
        else:
            if scheme in ["postgresql", "mysql", "sqlite", "mssql", "oracle"]:
                dbt = DatabaseTools(
                    tool_name_prefix=tool_name_prefix,
                    scheme=scheme,
                    host=host,
                    port=port,
                    user=user,
                    password=password,
                    dbname=dbname,
                    max_rows=max_rows,
                )
            else:
                raise ValueError(
                    "Please provide a SqlDatabase option or a valid DB scheme type "
                    " (postgresql, mysql, sqlite, mssql, oracle)."
                )

        # Update tools with description
        tools = dbt.to_tool_list()
        vtools = []
        for tool in tools:
            if content_description:
                tool.metadata.description = (
                    tool.metadata.description
                    + f"The database tables include data about {content_description}."
                )
            vtool = VectaraTool(
                tool_type=ToolType.QUERY,
                fn=tool.fn,
                async_fn=tool.async_fn,
                metadata=tool.metadata,
            )
            vtools.append(vtool)
        return vtools

create_tool(function, tool_type=ToolType.QUERY)

Create a tool from a function.

Parameters:

Name	Type	Description	Default
function	Callable	a function to convert into a tool.	required
tool_type	ToolType	the type of tool.	QUERY

Returns:

Name	Type	Description
VectaraTool	VectaraTool	A VectaraTool object.

Source code in vectara_agentic/tools.py

def create_tool(
    self, function: Callable, tool_type: ToolType = ToolType.QUERY
) -> VectaraTool:
    """
    Create a tool from a function.

    Args:
        function (Callable): a function to convert into a tool.
        tool_type (ToolType): the type of tool.

    Returns:
        VectaraTool: A VectaraTool object.
    """
    return VectaraTool.from_defaults(tool_type=tool_type, fn=function)

database_tools(tool_name_prefix='', content_description=None, sql_database=None, scheme=None, host='localhost', port='5432', user='postgres', password='Password', dbname='postgres', max_rows=1000)

Returns a list of database tools.

Args:

tool_name_prefix (str, optional): The prefix to add to the tool names. Defaults to "".
content_description (str, optional): The content description for the database. Defaults to None.
sql_database (SQLDatabase, optional): The SQLDatabase object. Defaults to None.
scheme (str, optional): The database scheme. Defaults to None.
host (str, optional): The database host. Defaults to "localhost".
port (str, optional): The database port. Defaults to "5432".
user (str, optional): The database user. Defaults to "postgres".
password (str, optional): The database password. Defaults to "Password".
dbname (str, optional): The database name. Defaults to "postgres".
   You must specify either the sql_database object or the scheme, host, port, user, password, and dbname.
max_rows (int, optional): if specified, instructs the load_data tool to never return more than max_rows
   rows. Defaults to 1000.

Returns:

Type	Description
List[VectaraTool]	List[VectaraTool]: A list of VectaraTool objects.

Source code in vectara_agentic/tools.py

def database_tools(
    self,
    tool_name_prefix: str = "",
    content_description: Optional[str] = None,
    sql_database: Optional[SQLDatabase] = None,
    scheme: Optional[str] = None,
    host: str = "localhost",
    port: str = "5432",
    user: str = "postgres",
    password: str = "Password",
    dbname: str = "postgres",
    max_rows: int = 1000,
) -> List[VectaraTool]:
    """
    Returns a list of database tools.

    Args:

        tool_name_prefix (str, optional): The prefix to add to the tool names. Defaults to "".
        content_description (str, optional): The content description for the database. Defaults to None.
        sql_database (SQLDatabase, optional): The SQLDatabase object. Defaults to None.
        scheme (str, optional): The database scheme. Defaults to None.
        host (str, optional): The database host. Defaults to "localhost".
        port (str, optional): The database port. Defaults to "5432".
        user (str, optional): The database user. Defaults to "postgres".
        password (str, optional): The database password. Defaults to "Password".
        dbname (str, optional): The database name. Defaults to "postgres".
           You must specify either the sql_database object or the scheme, host, port, user, password, and dbname.
        max_rows (int, optional): if specified, instructs the load_data tool to never return more than max_rows
           rows. Defaults to 1000.

    Returns:
        List[VectaraTool]: A list of VectaraTool objects.
    """
    if sql_database:
        dbt = DatabaseTools(
            tool_name_prefix=tool_name_prefix,
            sql_database=sql_database,
            max_rows=max_rows,
        )
    else:
        if scheme in ["postgresql", "mysql", "sqlite", "mssql", "oracle"]:
            dbt = DatabaseTools(
                tool_name_prefix=tool_name_prefix,
                scheme=scheme,
                host=host,
                port=port,
                user=user,
                password=password,
                dbname=dbname,
                max_rows=max_rows,
            )
        else:
            raise ValueError(
                "Please provide a SqlDatabase option or a valid DB scheme type "
                " (postgresql, mysql, sqlite, mssql, oracle)."
            )

    # Update tools with description
    tools = dbt.to_tool_list()
    vtools = []
    for tool in tools:
        if content_description:
            tool.metadata.description = (
                tool.metadata.description
                + f"The database tables include data about {content_description}."
            )
        vtool = VectaraTool(
            tool_type=ToolType.QUERY,
            fn=tool.fn,
            async_fn=tool.async_fn,
            metadata=tool.metadata,
        )
        vtools.append(vtool)
    return vtools

financial_tools()

Create a list of financial tools.

Source code in vectara_agentic/tools.py

def financial_tools(self):
    """
    Create a list of financial tools.
    """
    return self.get_llama_index_tools(
        tool_package_name="yahoo_finance", tool_spec_name="YahooFinanceToolSpec"
    )

get_llama_index_tools(tool_package_name, tool_spec_name, tool_name_prefix='', **kwargs)

Get a tool from the llama_index hub.

Parameters:

Name	Type	Description	Default
tool_package_name	str	The name of the tool package.	required
tool_spec_name	str	The name of the tool spec.	required
tool_name_prefix	str	The prefix to add to the tool names (added to every tool in the spec).	''
kwargs	dict	The keyword arguments to pass to the tool constructor (see Hub for tool specific details).	{}

Returns:

Type	Description
List[VectaraTool]	List[VectaraTool]: A list of VectaraTool objects.

Source code in vectara_agentic/tools.py

def get_llama_index_tools(
    self,
    tool_package_name: str,
    tool_spec_name: str,
    tool_name_prefix: str = "",
    **kwargs: dict,
) -> List[VectaraTool]:
    """
    Get a tool from the llama_index hub.

    Args:
        tool_package_name (str): The name of the tool package.
        tool_spec_name (str): The name of the tool spec.
        tool_name_prefix (str, optional): The prefix to add to the tool names (added to every tool in the spec).
        kwargs (dict): The keyword arguments to pass to the tool constructor (see Hub for tool specific details).

    Returns:
        List[VectaraTool]: A list of VectaraTool objects.
    """
    # Dynamically install and import the module
    if tool_package_name not in LI_packages:
        raise ValueError(
            f"Tool package {tool_package_name} from LlamaIndex not supported by Vectara-agentic."
        )

    module_name = f"llama_index.tools.{tool_package_name}"
    module = importlib.import_module(module_name)

    # Get the tool spec class or function from the module
    tool_spec = getattr(module, tool_spec_name)
    func_type = LI_packages[tool_package_name]
    tools = tool_spec(**kwargs).to_tool_list()
    vtools = []
    for tool in tools:
        if len(tool_name_prefix) > 0:
            tool.metadata.name = tool_name_prefix + "_" + tool.metadata.name
        if isinstance(func_type, dict):
            if tool_spec_name not in func_type.keys():
                raise ValueError(
                    f"Tool spec {tool_spec_name} not found in package {tool_package_name}."
                )
            tool_type = func_type[tool_spec_name]
        else:
            tool_type = func_type
        vtool = VectaraTool(
            tool_type=tool_type,
            fn=tool.fn,
            metadata=tool.metadata,
            async_fn=tool.async_fn,
        )
        vtools.append(vtool)
    return vtools

guardrail_tools()

Create a list of guardrail tools to avoid controversial topics.

Source code in vectara_agentic/tools.py

def guardrail_tools(self) -> List[FunctionTool]:
    """
    Create a list of guardrail tools to avoid controversial topics.
    """
    return [self.create_tool(get_bad_topics)]

legal_tools()

Create a list of legal tools.

Source code in vectara_agentic/tools.py

def legal_tools(self) -> List[FunctionTool]:
    """
    Create a list of legal tools.
    """

    def summarize_legal_text(
        text: str = Field(description="the original text."),
    ) -> str:
        """
        Use this tool to summarize legal text with no more than summary_max_length characters.
        """
        tc = ToolsCatalog(self.agent_config)
        return tc.summarize_text(text, expertise="law")

    def critique_as_judge(
        text: str = Field(description="the original text."),
    ) -> str:
        """
        Critique the legal document.
        """
        tc = ToolsCatalog(self.agent_config)
        return tc.critique_text(
            text,
            role="judge",
            point_of_view="""
            an experienced judge evaluating a legal document to provide areas of concern
            or that may require further legal scrutiny or legal argument.
            """,
        )

    return [
        self.create_tool(tool) for tool in [summarize_legal_text, critique_as_judge]
    ]

standard_tools()

Create a list of standard tools.

Source code in vectara_agentic/tools.py

def standard_tools(self) -> List[FunctionTool]:
    """
    Create a list of standard tools.
    """
    tc = ToolsCatalog(self.agent_config)
    return [
        self.create_tool(tool)
        for tool in [tc.summarize_text, tc.rephrase_text, tc.critique_text]
    ]

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 __str__(self) -> str:
        return f"Tool(name={self.metadata.name}, " f"Tool metadata={self.metadata})"

    def __repr__(self) -> str:
        return str(self)

    def __eq__(self, other):
        if not isinstance(other, VectaraTool):
            return False

        if self.metadata.tool_type != other.metadata.tool_type:
            return False

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

        # If schema is a dict-like object, compare the dict representation
        try:
            # Try to get schema as dict if possible
            if hasattr(self.metadata.fn_schema, "schema"):
                self_schema = self.metadata.fn_schema.schema
                other_schema = other.metadata.fn_schema.schema

                # Compare only properties and required fields
                self_props = self_schema.get("properties", {})
                other_props = other_schema.get("properties", {})

                self_required = self_schema.get("required", [])
                other_required = other_schema.get("required", [])

                return self_props.keys() == other_props.keys() and set(
                    self_required
                ) == set(other_required)
        except Exception:
            # If any exception occurs during schema comparison, fall back to name comparison
            pass

        return True

    def call(
        self, *args: Any, ctx: Optional[Context] = None, **kwargs: Any
    ) -> ToolOutput:
        try:
            return super().call(*args, ctx=ctx, **kwargs)
        except TypeError as e:
            sig = inspect.signature(self.metadata.fn_schema)
            valid_parameters = list(sig.parameters.keys())
            params_str = ", ".join(valid_parameters)

            err_output = ToolOutput(
                tool_name=self.metadata.name,
                content=(
                    f"Wrong argument used when calling {self.metadata.name}: {str(e)}. "
                    f"Valid arguments: {params_str}. please call the tool again with the correct arguments."
                ),
                raw_input={"args": args, "kwargs": kwargs},
                raw_output={"response": str(e)},
            )
            return err_output
        except Exception as e:
            err_output = ToolOutput(
                tool_name=self.metadata.name,
                content=f"Tool {self.metadata.name} 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 await super().acall(*args, ctx=ctx, **kwargs)
        except TypeError as e:
            sig = inspect.signature(self.metadata.fn_schema)
            valid_parameters = list(sig.parameters.keys())
            params_str = ", ".join(valid_parameters)

            err_output = ToolOutput(
                tool_name=self.metadata.name,
                content=(
                    f"Wrong argument used when calling {self.metadata.name}: {str(e)}. "
                    f"Valid arguments: {params_str}. please call the tool again with the correct arguments."
                ),
                raw_input={"args": args, "kwargs": kwargs},
                raw_output={"response": str(e)},
            )
            return err_output
        except Exception as e:
            err_output = ToolOutput(
                tool_name=self.metadata.name,
                content=f"Tool {self.metadata.name} 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(","))
        self.cache_expiry = 60 * 60  # 1 hour
        self.max_cache_size = 128

    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",
            vectara_base_url=vectara_base_url,
            vectara_verify_ssl=vectara_verify_ssl,
        )

        # Dynamically generate the search function
        def search_function(*args: Any, **kwargs: Any) -> 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)
            summarize = kwargs.pop("summarize", True)
            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},
                )
            unique_ids = set()
            docs = []
            for doc in response:
                if doc.id_ in unique_ids:
                    continue
                unique_ids.add(doc.id_)
                docs.append((doc.id_, doc.metadata))
            tool_output = "Matching documents:\n"
            if summarize:
                summaries_dict = asyncio.run(
                    summarize_documents(
                        self.vectara_corpus_key, self.vectara_api_key, list(unique_ids)
                    )
                )
                for doc_id, metadata in docs:
                    summary = summaries_dict.get(doc_id, "")
                    tool_output += f"document_id: '{doc_id}'\nmetadata: '{metadata}'\nsummary: '{summary}'\n\n"
            else:
                for doc_id, metadata in docs:
                    tool_output += (
                        f"document_id: '{doc_id}'\nmetadata: '{metadata}'\n\n"
                    )

            out = ToolOutput(
                tool_name=search_function.__name__,
                content=tool_output,
                raw_input={"args": args, "kwargs": kwargs},
                raw_output=response,
            )
            return out

        class SearchToolBaseParams(BaseModel):
            """Model for the base parameters of the search tool."""
            query: str = Field(
                ...,
                description="The search query to perform, always in the form of a question.",
            )
            top_k: int = Field(
                10, description="The number of top documents to retrieve."
            )
            summarize: bool = Field(
                True,
                description="Flag that indicates whether to summarize the retrieved documents.",
            )

        search_tool_extra_desc = (
            tool_description
            + "\n"
            + "This tool is meant to perform a search for relevant documents, it is not meant for asking questions."
        )

        tool = _create_tool_from_dynamic_function(
            search_function,
            tool_name,
            search_tool_extra_desc,
            SearchToolBaseParams,
            tool_args_schema,
        )
        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",
            vectara_base_url=vectara_base_url,
            vectara_verify_ssl=vectara_verify_ssl,
        )

        # Dynamically generate the RAG function
        def rag_function(*args: Any, **kwargs: Any) -> 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_sentences_before=n_sentences_before,
                n_sentences_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

        class RagToolBaseParams(BaseModel):
            """Model for the base parameters of the RAG tool."""
            query: str = Field(
                ...,
                description="The search query to perform, always in the form of a question",
            )

        tool = _create_tool_from_dynamic_function(
            rag_function,
            tool_name,
            tool_description,
            RagToolBaseParams,
            tool_args_schema,
        )
        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(","))
    self.cache_expiry = 60 * 60  # 1 hour
    self.max_cache_size = 128

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",
        vectara_base_url=vectara_base_url,
        vectara_verify_ssl=vectara_verify_ssl,
    )

    # Dynamically generate the RAG function
    def rag_function(*args: Any, **kwargs: Any) -> 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_sentences_before=n_sentences_before,
            n_sentences_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

    class RagToolBaseParams(BaseModel):
        """Model for the base parameters of the RAG tool."""
        query: str = Field(
            ...,
            description="The search query to perform, always in the form of a question",
        )

    tool = _create_tool_from_dynamic_function(
        rag_function,
        tool_name,
        tool_description,
        RagToolBaseParams,
        tool_args_schema,
    )
    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",
        vectara_base_url=vectara_base_url,
        vectara_verify_ssl=vectara_verify_ssl,
    )

    # Dynamically generate the search function
    def search_function(*args: Any, **kwargs: Any) -> 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)
        summarize = kwargs.pop("summarize", True)
        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},
            )
        unique_ids = set()
        docs = []
        for doc in response:
            if doc.id_ in unique_ids:
                continue
            unique_ids.add(doc.id_)
            docs.append((doc.id_, doc.metadata))
        tool_output = "Matching documents:\n"
        if summarize:
            summaries_dict = asyncio.run(
                summarize_documents(
                    self.vectara_corpus_key, self.vectara_api_key, list(unique_ids)
                )
            )
            for doc_id, metadata in docs:
                summary = summaries_dict.get(doc_id, "")
                tool_output += f"document_id: '{doc_id}'\nmetadata: '{metadata}'\nsummary: '{summary}'\n\n"
        else:
            for doc_id, metadata in docs:
                tool_output += (
                    f"document_id: '{doc_id}'\nmetadata: '{metadata}'\n\n"
                )

        out = ToolOutput(
            tool_name=search_function.__name__,
            content=tool_output,
            raw_input={"args": args, "kwargs": kwargs},
            raw_output=response,
        )
        return out

    class SearchToolBaseParams(BaseModel):
        """Model for the base parameters of the search tool."""
        query: str = Field(
            ...,
            description="The search query to perform, always in the form of a question.",
        )
        top_k: int = Field(
            10, description="The number of top documents to retrieve."
        )
        summarize: bool = Field(
            True,
            description="Flag that indicates whether to summarize the retrieved documents.",
        )

    search_tool_extra_desc = (
        tool_description
        + "\n"
        + "This tool is meant to perform a search for relevant documents, it is not meant for asking questions."
    )

    tool = _create_tool_from_dynamic_function(
        search_function,
        tool_name,
        search_tool_extra_desc,
        SearchToolBaseParams,
        tool_args_schema,
    )
    return tool

create_app(agent, config)

Create a FastAPI application with a chat endpoint.

Source code in vectara_agentic/agent_endpoint.py

def create_app(agent: Agent, config: AgentConfig) -> FastAPI:
    """
    Create a FastAPI application with a chat endpoint.
    """
    app = FastAPI()
    logger = logging.getLogger("uvicorn.error")
    logging.basicConfig(level=logging.INFO)
    endpoint_api_key = config.endpoint_api_key

    @app.get("/chat", summary="Chat with the agent")
    async def chat(message: str, api_key: str = Depends(api_key_header)):
        logger.info(f"Received message: {message}")
        if api_key != endpoint_api_key:
            logger.warning("Unauthorized access attempt")
            raise HTTPException(status_code=403, detail="Unauthorized")

        if not message:
            logger.error("No message provided in the request")
            raise HTTPException(status_code=400, detail="No message provided")

        try:
            response = agent.chat(message)
            logger.info(f"Generated response: {response}")
            return {"response": response}
        except Exception as e:
            logger.error(f"Error during agent processing: {e}")
            raise HTTPException(status_code=500, detail="Internal server error") from e

    return app

start_app(agent, host='0.0.0.0', port=8000)

Start the FastAPI server.

Parameters:

Name	Type	Description	Default
host	str	The host address for the API. Defaults to '127.0.0.1'.	'0.0.0.0'
port	int	The port for the API. Defaults to 8000.	8000

Source code in vectara_agentic/agent_endpoint.py

def start_app(agent: Agent, host='0.0.0.0', port=8000):
    """
    Start the FastAPI server.

    Args:
        host (str, optional): The host address for the API. Defaults to '127.0.0.1'.
        port (int, optional): The port for the API. Defaults to 8000.
    """
    app = create_app(agent, config=AgentConfig())
    uvicorn.run(app, host=host, port=port)