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 = "",
        general_instructions: str = GENERAL_INSTRUCTIONS,
        verbose: bool = False,
        agent_progress_callback: Optional[
            Callable[[AgentStatusType, dict, 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: Optional["Workflow"] = None,
        workflow_timeout: int = 120,
        vectara_api_key: Optional[str] = None,
        session_id: Optional[str] = None,
    ) -> 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 ''.
            general_instructions (str, optional): General instructions for the agent.
                The Agent has a default set of instructions that are crafted to help it operate effectively.
                This allows you to customize the agent's behavior and personality, but use with caution.
            verbose (bool, optional): Whether the agent should print its steps. Defaults to False.
            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.
                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.
            vectara_api_key (str, optional): The Vectara API key for VHC computation. Defaults to None.
            session_id (str, optional): The session ID for memory persistence.
                                        If None, auto-generates from topic and date. Defaults to None.
        """
        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, vhc_eligible=False)
            ]
        self.agent_type = self.agent_config.agent_type
        self._llm = None  # Lazy loading
        self._custom_instructions = custom_instructions
        self._general_instructions = general_instructions
        self._topic = topic
        self.agent_progress_callback = agent_progress_callback

        self.query_logging_callback = query_logging_callback
        self.workflow_cls = workflow_cls
        self.workflow_timeout = workflow_timeout
        self.vectara_api_key = vectara_api_key or os.environ.get("VECTARA_API_KEY", "")

        # Sanitize tools for Gemini if needed
        if self.agent_config.main_llm_provider == ModelProvider.GEMINI:
            self.tools = sanitize_tools_for_gemini(self.tools)

        # Validate tools
        if validate_tools:
            validate_tool_consistency(
                self.tools, self._custom_instructions, self.agent_config
            )

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

        self.session_id = (
            session_id
            or getattr(self, "session_id", None)
            or f"{topic}:{date.today().isoformat()}"
        )

        chat_store = SimpleChatStore()
        self.memory = ChatMemoryBuffer.from_defaults(
            chat_store=chat_store,
            chat_store_key=self.session_id,
            token_limit=65536
        )
        if chat_history:
            msgs = []
            for u, a in chat_history:
                msgs.append(ChatMessage.from_str(u, role=MessageRole.USER))
                msgs.append(ChatMessage.from_str(a, role=MessageRole.ASSISTANT))
            self.memory.put_messages(msgs)

        # Set up main agent and fallback agent
        self._agent = None  # Lazy loading
        self.fallback_agent_config = fallback_agent_config
        self._fallback_agent = None  # Lazy loading

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

        # VHC state tracking
        self._vhc_cache = {}  # Cache VHC results by query hash
        self._last_query = None
        self._last_response = None
        self._current_tool_outputs = []  # Store tool outputs from current query for VHC

    @property
    def llm(self):
        """Lazy-loads the LLM."""
        if self._llm is None:
            self._llm = get_llm(LLMRole.MAIN, config=self.agent_config)
        return self._llm

    @property
    def agent(self):
        """Lazy-loads the agent."""
        if self._agent is None:
            self._agent = self._create_agent(self.agent_config, self.callback_manager)
        return self._agent

    @property
    def fallback_agent(self):
        """Lazy-loads the fallback agent."""
        if self._fallback_agent is None and self.fallback_agent_config:
            self._fallback_agent = self._create_agent(
                self.fallback_agent_config, self.callback_manager
            )
        return self._fallback_agent

    def _create_agent(
        self, config: AgentConfig, llm_callback_manager: "CallbackManager"
    ) -> "BaseWorkflowAgent":
        """
        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:
            BaseWorkflowAgent: The configured agent object.
        """
        # Use the same LLM instance for consistency
        llm = (
            self.llm
            if config == self.agent_config
            else get_llm(LLMRole.MAIN, config=config)
        )
        llm.callback_manager = llm_callback_manager

        return create_agent_from_config(
            tools=self.tools,
            llm=llm,
            memory=self.memory,
            config=config,
            callback_manager=llm_callback_manager,
            general_instructions=self._general_instructions,
            topic=self._topic,
            custom_instructions=self._custom_instructions,
            verbose=self.verbose,
        )

    def clear_memory(self) -> None:
        """Clear the agent's memory and reset agent instances to ensure consistency."""
        self.memory.reset()
        # Clear agent instances so they get recreated with the cleared memory
        self._agent = None
        self._fallback_agent = None

    def __eq__(self, other):
        if not isinstance(other, Agent):
            logger.debug(
                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:
            logger.debug(
                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:
            logger.debug(
                "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:
            logger.debug(
                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:
            logger.debug(
                "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:
            logger.debug(
                f"Comparison failed: verbose differs. (self.verbose: {self.verbose}, other.verbose: {other.verbose})"
            )
            return False

        # Compare agent memory
        if self.memory.get() != other.memory.get():
            logger.debug("Comparison failed: agent memory differs.")
            return False

        # If all comparisons pass
        logger.debug("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,
        agent_progress_callback: Optional[
            Callable[[AgentStatusType, dict, 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: Optional["Workflow"] = None,
        workflow_timeout: int = 120,
        session_id: Optional[str] = None,
    ) -> "Agent":
        """
        Create an agent from tools, agent type, and language model.

        Args:

            tools (list[FunctionTool]): A list of tools to be used by the agent.
            topic (str, optional): The topic for the agent. Defaults to 'general'.
            custom_instructions (str, optional): custom instructions for the agent. Defaults to ''.
            verbose (bool, optional): Whether the agent should print its steps. Defaults to True.
            agent_progress_callback (Callable): A callback function the code calls on any agent updates.
            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.
            session_id (str, optional): The session ID for memory persistence.
                                        If None, auto-generates from topic and date. Defaults to None.

        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,
            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,
            session_id=session_id,
        )

    @classmethod
    def from_corpus(
        cls,
        tool_name: str,
        data_description: str,
        assistant_specialty: str,
        general_instructions: str = GENERAL_INSTRUCTIONS,
        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, dict, 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: Optional[str] = None,
        vectara_rerank_chain: Optional[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,
        return_direct: bool = False,
        session_id: Optional[str] = None,
    ) -> "Agent":
        """Create an agent from a single Vectara corpus using the factory function.

        Args:
            tool_name (str): Name of the tool to be created.
            data_description (str): Description of the data/corpus.
            assistant_specialty (str): The specialty/topic of the assistant.
            session_id (str, optional): The session ID for memory persistence.
                                        If None, auto-generates from topic and date. Defaults to None.
            ... (other parameters as documented in factory function)
        """
        # Use the factory function to avoid code duplication
        config = create_agent_from_corpus(
            tool_name=tool_name,
            data_description=data_description,
            assistant_specialty=assistant_specialty,
            general_instructions=general_instructions,
            vectara_corpus_key=vectara_corpus_key,
            vectara_api_key=vectara_api_key,
            agent_config=agent_config,
            fallback_agent_config=fallback_agent_config,
            verbose=verbose,
            vectara_filter_fields=vectara_filter_fields,
            vectara_offset=vectara_offset,
            vectara_lambda_val=vectara_lambda_val,
            vectara_semantics=vectara_semantics,
            vectara_custom_dimensions=vectara_custom_dimensions,
            vectara_reranker=vectara_reranker,
            vectara_rerank_k=vectara_rerank_k,
            vectara_rerank_limit=vectara_rerank_limit,
            vectara_rerank_cutoff=vectara_rerank_cutoff,
            vectara_diversity_bias=vectara_diversity_bias,
            vectara_udf_expression=vectara_udf_expression,
            vectara_rerank_chain=vectara_rerank_chain,
            vectara_n_sentences_before=vectara_n_sentences_before,
            vectara_n_sentences_after=vectara_n_sentences_after,
            vectara_summary_num_results=vectara_summary_num_results,
            vectara_summarizer=vectara_summarizer,
            vectara_summary_response_language=vectara_summary_response_language,
            vectara_summary_prompt_text=vectara_summary_prompt_text,
            vectara_max_response_chars=vectara_max_response_chars,
            vectara_max_tokens=vectara_max_tokens,
            vectara_temperature=vectara_temperature,
            vectara_frequency_penalty=vectara_frequency_penalty,
            vectara_presence_penalty=vectara_presence_penalty,
            vectara_save_history=vectara_save_history,
            return_direct=return_direct,
        )

        return cls(
            chat_history=chat_history,
            agent_progress_callback=agent_progress_callback,
            query_logging_callback=query_logging_callback,
            session_id=session_id,
            **config,
        )

    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.
        Ensures memory consistency by clearing agent instances so they are recreated with current memory.
        """
        if self.agent_config_type == AgentConfigType.DEFAULT:
            self.agent_config_type = AgentConfigType.FALLBACK
            # Clear the fallback agent so it gets recreated with current memory
            self._fallback_agent = None
        else:
            self.agent_config_type = AgentConfigType.DEFAULT
            # Clear the main agent so it gets recreated with current memory
            self._agent = None

    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.
        """
        logger.info("Vectara agentic Report:")
        logger.info(f"Agent Type = {self.agent_config.agent_type}")
        logger.info(f"Topic = {self._topic}")
        logger.info("Tools:")
        for tool in self.tools:
            if hasattr(tool, "metadata"):
                if detailed:
                    logger.info(f"- {tool.metadata.description}")
                else:
                    logger.info(f"- {tool.metadata.name}")
            else:
                logger.info("- tool without metadata")
        logger.info(
            f"Agent LLM = {get_llm(LLMRole.MAIN, config=self.agent_config).metadata.model_name}"
        )
        logger.info(
            f"Tool LLM = {get_llm(LLMRole.TOOL, config=self.agent_config).metadata.model_name}"
        )

    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
            or not self.fallback_agent_config
            else self.fallback_agent_config.agent_type
        )

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

        Args:
            prompt (str): The chat prompt.

        Returns:
            AgentResponse: The response from the agent.
        """
        try:
            _ = asyncio.get_running_loop()
        except RuntimeError:
            return asyncio.run(self.achat(prompt))

        # We are inside a running loop (Jupyter, uvicorn, etc.)
        raise RuntimeError(
            "Use `await agent.achat(...)` inside an event loop (e.g. Jupyter)."
        )

    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.
        """
        if not prompt:
            return AgentResponse(response="")

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

                # Deal with workflow-based agent types (Function Calling and ReAct)
                if self._get_current_agent_type() in [
                    AgentType.FUNCTION_CALLING,
                    AgentType.REACT,
                ]:
                    from llama_index.core.workflow import Context

                    # Create context and pass memory to the workflow agent
                    # According to LlamaIndex docs, we should let the workflow manage memory internally
                    ctx = Context(current_agent)

                    handler = current_agent.run(
                        user_msg=prompt, memory=self.memory, ctx=ctx
                    )

                    # Listen to workflow events if progress callback is set
                    if self.agent_progress_callback:
                        # Create event tracker for consistent event ID generation
                        from .agent_core.streaming import ToolEventTracker

                        event_tracker = ToolEventTracker()

                        async for event in handler.stream_events():
                            # Use consistent event ID tracking to ensure tool calls and outputs are paired
                            event_id = event_tracker.get_event_id(event)

                            # Handle different types of workflow events using same logic as FunctionCallingStreamHandler
                            from llama_index.core.agent.workflow import (
                                ToolCall,
                                ToolCallResult,
                                AgentInput,
                                AgentOutput,
                            )

                            if isinstance(event, ToolCall):
                                self.agent_progress_callback(
                                    status_type=AgentStatusType.TOOL_CALL,
                                    msg={
                                        "tool_name": event.tool_name,
                                        "arguments": json.dumps(event.tool_kwargs),
                                    },
                                    event_id=event_id,
                                )
                            elif isinstance(event, ToolCallResult):
                                self.agent_progress_callback(
                                    status_type=AgentStatusType.TOOL_OUTPUT,
                                    msg={
                                        "tool_name": event.tool_name,
                                        "content": str(event.tool_output),
                                    },
                                    event_id=event_id,
                                )
                            elif isinstance(event, AgentInput):
                                self.agent_progress_callback(
                                    status_type=AgentStatusType.AGENT_UPDATE,
                                    msg={"content": f"Agent input: {event.input}"},
                                    event_id=event_id,
                                )
                            elif isinstance(event, AgentOutput):
                                self.agent_progress_callback(
                                    status_type=AgentStatusType.AGENT_UPDATE,
                                    msg={"content": f"Agent output: {event.response}"},
                                    event_id=event_id,
                                )

                    result = await handler

                    # Ensure we have an AgentResponse object with a string response
                    if hasattr(result, "response"):
                        response_text = result.response
                    else:
                        response_text = str(result)

                    # Handle case where response is a ChatMessage object
                    if hasattr(response_text, "content"):
                        response_text = response_text.content
                    elif not isinstance(response_text, str):
                        response_text = str(response_text)

                    if response_text is None or response_text == "None":
                        # Try to find tool outputs in the result object
                        response_text = None

                        # Check various possible locations for tool outputs
                        if hasattr(result, "tool_outputs") and result.tool_outputs:
                            # Get the latest tool output
                            latest_output = (
                                result.tool_outputs[-1]
                                if isinstance(result.tool_outputs, list)
                                else result.tool_outputs
                            )
                            response_text = str(latest_output)

                        # Check if there are tool_calls with results
                        elif hasattr(result, "tool_calls") and result.tool_calls:
                            # Tool calls might contain the outputs - let's try to extract them
                            for tool_call in result.tool_calls:
                                if (
                                    hasattr(tool_call, "tool_output")
                                    and tool_call.tool_output is not None
                                ):
                                    response_text = str(tool_call.tool_output)
                                    break

                        elif hasattr(result, "sources") or hasattr(
                            result, "source_nodes"
                        ):
                            sources = getattr(
                                result, "sources", getattr(result, "source_nodes", [])
                            )
                            if (
                                sources
                                and len(sources) > 0
                                and hasattr(sources[0], "text")
                            ):
                                response_text = sources[0].text

                        # Check for workflow context or chat history that might contain tool results
                        elif hasattr(result, "chat_history"):
                            # Look for the most recent assistant message that might contain tool results
                            chat_history = result.chat_history
                            if chat_history and len(chat_history) > 0:
                                for msg in reversed(chat_history):
                                    if (
                                        msg.role == MessageRole.TOOL
                                        and msg.content
                                        and str(msg.content).strip()
                                    ):
                                        response_text = msg.content
                                        break
                                    if (
                                        hasattr(msg, "content")
                                        and msg.content
                                        and str(msg.content).strip()
                                    ):
                                        response_text = msg.content
                                        break

                        # If we still don't have a response, provide a fallback
                        if response_text is None or response_text == "None":
                            response_text = "Response completed."

                    agent_response = AgentResponse(
                        response=response_text, metadata=getattr(result, "metadata", {})
                    )

                    # Retrieve updated memory from workflow context
                    # According to LlamaIndex docs, workflow agents manage memory internally
                    # and we can access it via ctx.store.get("memory")
                    try:
                        workflow_memory = await ctx.store.get("memory")
                        if workflow_memory:
                            # Update our external memory with the workflow's memory
                            self.memory = workflow_memory
                    except Exception as e:
                        # If we can't retrieve workflow memory, fall back to manual management
                        warning_msg = (
                            f"Could not retrieve workflow memory, falling back to "
                            f"manual management: {e}"
                        )
                        logger.warning(warning_msg)
                        user_msg = ChatMessage.from_str(prompt, role=MessageRole.USER)
                        assistant_msg = ChatMessage.from_str(
                            response_text, role=MessageRole.ASSISTANT
                        )
                        self.memory.put_messages([user_msg, assistant_msg])

                # Standard chat interaction for other agent types
                else:
                    agent_response = await current_agent.achat(prompt)

                # Post processing after response is generated
                agent_response.metadata = agent_response.metadata or {}
                user_metadata = agent_response.metadata
                agent_response = await execute_post_stream_processing(
                    agent_response, prompt, self, user_metadata
                )
                return agent_response

            except Exception as e:
                last_error = e
                if self.verbose:
                    logger.warning(
                        f"LLM call failed on attempt {attempt}. " f"Error: {e}."
                    )
                if attempt >= 2 and self.fallback_agent_config:
                    self._switch_agent_config()
                await asyncio.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:
        """
        Interact with the agent using a chat prompt with streaming.
        Args:
            prompt (str): The chat prompt.
        Returns:
            AgentStreamingResponse: The streaming response from the agent.
        """
        try:
            _ = asyncio.get_running_loop()
        except RuntimeError:
            return asyncio.run(self.astream_chat(prompt))
        raise RuntimeError(
            "Use `await agent.astream_chat(...)` inside an event loop (e.g. Jupyter)."
        )

    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.
        """
        # Store query for VHC processing and clear previous tool outputs
        self._last_query = prompt
        self._clear_tool_outputs()
        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()
                user_meta: Dict[str, Any] = {}

                # Deal with Function Calling agent type
                if self._get_current_agent_type() == AgentType.FUNCTION_CALLING:
                    from llama_index.core.workflow import Context

                    # Create context and pass memory to the workflow agent
                    # According to LlamaIndex docs, we should let the workflow manage memory internally
                    ctx = Context(current_agent)

                    handler = current_agent.run(
                        user_msg=prompt, memory=self.memory, ctx=ctx
                    )

                    # Use the dedicated FunctionCallingStreamHandler
                    stream_handler = FunctionCallingStreamHandler(self, handler, prompt)
                    streaming_adapter = stream_handler.create_streaming_response(
                        user_meta
                    )

                    return AgentStreamingResponse(
                        base=streaming_adapter, metadata=user_meta
                    )

                #
                # For other agent types, use the standard async chat method
                #
                li_stream = await current_agent.astream_chat(prompt)
                orig_async = li_stream.async_response_gen

                # Define a wrapper to preserve streaming behavior while executing post-stream logic.
                async def _stream_response_wrapper():
                    async for tok in orig_async():
                        yield tok

                    # Use shared post-processing function
                    await execute_post_stream_processing(
                        li_stream, prompt, self, user_meta
                    )

                li_stream.async_response_gen = _stream_response_wrapper
                return AgentStreamingResponse(base=li_stream, metadata=user_meta)

            except Exception as e:
                last_error = e
                if attempt >= 2 and self.fallback_agent_config:
                    self._switch_agent_config()
                await asyncio.sleep(1)
                attempt += 1

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

    def _clear_tool_outputs(self):
        """Clear stored tool outputs at the start of a new query."""
        self._current_tool_outputs.clear()
        logging.info("🔧 [TOOL_STORAGE] Cleared stored tool outputs for new query")

    def _add_tool_output(self, tool_name: str, content: str):
        """Add a tool output to the current collection for VHC."""
        tool_output = {
            'status_type': 'TOOL_OUTPUT',
            'content': content,
            'tool_name': tool_name
        }
        self._current_tool_outputs.append(tool_output)
        logging.info(f"🔧 [TOOL_STORAGE] Added tool output from '{tool_name}': {len(content)} chars")

    def _get_stored_tool_outputs(self) -> List[dict]:
        """Get the stored tool outputs from the current query."""
        logging.info(f"🔧 [TOOL_STORAGE] Retrieved {len(self._current_tool_outputs)} stored tool outputs")
        return self._current_tool_outputs.copy()

    async def acompute_vhc(self) -> Dict[str, Any]:
        """
        Compute VHC for the last query/response pair (async version).
        Results are cached for subsequent calls. Tool outputs are automatically
        collected during streaming and used internally.

        Returns:
            Dict[str, Any]: Dictionary containing 'corrected_text' and 'corrections'
        """
        logging.info(
            f"🔍🔍🔍 [VHC_AGENT_ENTRY] UNIQUE_DEBUG_MESSAGE acompute_vhc method called - "
            f"stored_tool_outputs_count={len(self._current_tool_outputs)}"
        )
        logging.info(
            f"🔍🔍🔍 [VHC_AGENT_ENTRY] _last_query: {'set' if self._last_query else 'None'}"
        )

        if not self._last_query:
            logging.info("🔍 [VHC_AGENT] Returning early - no _last_query")
            return {"corrected_text": None, "corrections": []}

        # For VHC to work, we need the response text from memory
        # Get the latest assistant response from memory
        messages = self.memory.get()
        logging.info(
            f"🔍 [VHC_AGENT] memory.get() returned {len(messages) if messages else 0} messages"
        )

        if not messages:
            logging.info("🔍 [VHC_AGENT] Returning early - no messages in memory")
            return {"corrected_text": None, "corrections": []}

        # Find the last assistant message
        last_response = None
        for msg in reversed(messages):
            if msg.role == MessageRole.ASSISTANT:
                last_response = msg.content
                break

        logging.info(
            f"🔍 [VHC_AGENT] Found last_response: {'set' if last_response else 'None'}"
        )

        if not last_response:
            logging.info("🔍 [VHC_AGENT] Returning early - no last assistant response found")
            return {"corrected_text": None, "corrections": []}

        # Update stored response for caching
        self._last_response = last_response

        # Create cache key from query + response
        cache_key = hash(f"{self._last_query}:{self._last_response}")

        # Return cached results if available
        if cache_key in self._vhc_cache:
            return self._vhc_cache[cache_key]

        # Check if we have VHC API key
        logging.info(
            f"🔍 [VHC_AGENT] acompute_vhc called with vectara_api_key={'set' if self.vectara_api_key else 'None'}"
        )
        if not self.vectara_api_key:
            logging.info("🔍 [VHC_AGENT] No vectara_api_key - returning early with None")
            return {"corrected_text": None, "corrections": []}

        # Compute VHC using existing library function
        from .agent_core.utils.hallucination import analyze_hallucinations

        try:
            # Use stored tool outputs from current query
            stored_tool_outputs = self._get_stored_tool_outputs()
            logging.info(f"🔧 [VHC_AGENT] Using {len(stored_tool_outputs)} stored tool outputs for VHC")

            corrected_text, corrections = analyze_hallucinations(
                query=self._last_query,
                chat_history=self.memory.get(),
                agent_response=self._last_response,
                tools=self.tools,
                vectara_api_key=self.vectara_api_key,
                tool_outputs=stored_tool_outputs,
            )

            # Cache results
            results = {"corrected_text": corrected_text, "corrections": corrections}
            self._vhc_cache[cache_key] = results

            return results

        except Exception as e:
            logger.error(f"VHC computation failed: {e}")
            return {"corrected_text": None, "corrections": []}

    def compute_vhc(self) -> Dict[str, Any]:
        """
        Compute VHC for the last query/response pair (sync version).
        Results are cached for subsequent calls. Tool outputs are automatically
        collected during streaming and used internally.

        Returns:
            Dict[str, Any]: Dictionary containing 'corrected_text' and 'corrections'
        """
        try:
            loop = asyncio.get_event_loop()
            return loop.run_until_complete(self.acompute_vhc())
        except RuntimeError:
            # No event loop running, create a new one
            return asyncio.run(self.acompute_vhc())

    #
    # 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 or context 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}.")

        outputs_model_on_fail_cls = getattr(
            workflow.__class__, "OutputModelOnFail", None
        )
        if outputs_model_on_fail_cls:
            fields_without_default = []
            for name, field_info in outputs_model_on_fail_cls.model_fields.items():
                if field_info.default_factory is PydanticUndefined:
                    fields_without_default.append(name)
            if fields_without_default:
                raise ValueError(
                    f"Fields without default values: {fields_without_default}"
                )

        from llama_index.core.workflow import Context

        workflow_context = Context(workflow=workflow)
        try:
            # run workflow
            result = await workflow.run(
                ctx=workflow_context,
                agent=self,
                tools=self.tools,
                llm=self.llm,
                verbose=verbose,
                inputs=inputs,
            )

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

        except Exception as e:
            _missing = object()
            if outputs_model_on_fail_cls:
                model_fields = outputs_model_on_fail_cls.model_fields
                input_dict = {}
                for key in model_fields:
                    value = await workflow_context.get(key, default=_missing)
                    if value is not _missing:
                        input_dict[key] = value
                output = outputs_model_on_fail_cls.model_validate(input_dict)
            else:
                logger.warning(
                    f"Vectara Agentic: Workflow failed with unexpected error: {e}"
                )
                raise type(e)(str(e)).with_traceback(e.__traceback__)

        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_progress_callback: Optional[
            Callable[[AgentStatusType, dict, str], None]
        ] = None,
        query_logging_callback: Optional[Callable[[str, str], None]] = None,
    ) -> "Agent":
        """Create an Agent instance from a JSON string."""
        return cls.from_dict(
            json.loads(data), agent_progress_callback, query_logging_callback
        )

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

    @classmethod
    def from_dict(
        cls,
        data: Dict[str, Any],
        agent_progress_callback: Optional[Callable] = None,
        query_logging_callback: Optional[Callable] = None,
    ) -> "Agent":
        """Create an Agent instance from a dictionary."""
        return deserialize_agent_from_dict(
            cls, data, agent_progress_callback, query_logging_callback
        )

    def cleanup(self) -> None:
        """Clean up resources used by the agent."""
        from ._observability import shutdown_observer

        if hasattr(self, 'agent') and hasattr(self.agent, '_llm'):
            llm = self.agent._llm
            if hasattr(llm, 'client') and hasattr(llm.client, 'close'):
                try:
                    if asyncio.iscoroutinefunction(llm.client.close):
                        asyncio.run(llm.client.close())
                    else:
                        llm.client.close()
                except Exception:
                    pass

        # Shutdown observability connections
        shutdown_observer()

    def __enter__(self):
        """Context manager entry."""
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Context manager exit with cleanup."""
        self.cleanup()

agent property

Lazy-loads the agent.

fallback_agent property

Lazy-loads the fallback agent.

llm property

Lazy-loads the LLM.

__enter__()

Context manager entry.

Source code in vectara_agentic/agent.py

def __enter__(self):
    """Context manager entry."""
    return self

__exit__(exc_type, exc_val, exc_tb)

Context manager exit with cleanup.

Source code in vectara_agentic/agent.py

def __exit__(self, exc_type, exc_val, exc_tb):
    """Context manager exit with cleanup."""
    self.cleanup()

__init__(tools, topic='general', custom_instructions='', general_instructions=GENERAL_INSTRUCTIONS, verbose=False, 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, vectara_api_key=None, session_id=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 ''.
general_instructions (str, optional): General instructions for the agent.
    The Agent has a default set of instructions that are crafted to help it operate effectively.
    This allows you to customize the agent's behavior and personality, but use with caution.
verbose (bool, optional): Whether the agent should print its steps. Defaults to False.
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.
    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.
vectara_api_key (str, optional): The Vectara API key for VHC computation. Defaults to None.
session_id (str, optional): The session ID for memory persistence.
                            If None, auto-generates from topic and date. Defaults to None.

Source code in vectara_agentic/agent.py

def __init__(
    self,
    tools: List["FunctionTool"],
    topic: str = "general",
    custom_instructions: str = "",
    general_instructions: str = GENERAL_INSTRUCTIONS,
    verbose: bool = False,
    agent_progress_callback: Optional[
        Callable[[AgentStatusType, dict, 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: Optional["Workflow"] = None,
    workflow_timeout: int = 120,
    vectara_api_key: Optional[str] = None,
    session_id: Optional[str] = None,
) -> 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 ''.
        general_instructions (str, optional): General instructions for the agent.
            The Agent has a default set of instructions that are crafted to help it operate effectively.
            This allows you to customize the agent's behavior and personality, but use with caution.
        verbose (bool, optional): Whether the agent should print its steps. Defaults to False.
        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.
            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.
        vectara_api_key (str, optional): The Vectara API key for VHC computation. Defaults to None.
        session_id (str, optional): The session ID for memory persistence.
                                    If None, auto-generates from topic and date. Defaults to None.
    """
    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, vhc_eligible=False)
        ]
    self.agent_type = self.agent_config.agent_type
    self._llm = None  # Lazy loading
    self._custom_instructions = custom_instructions
    self._general_instructions = general_instructions
    self._topic = topic
    self.agent_progress_callback = agent_progress_callback

    self.query_logging_callback = query_logging_callback
    self.workflow_cls = workflow_cls
    self.workflow_timeout = workflow_timeout
    self.vectara_api_key = vectara_api_key or os.environ.get("VECTARA_API_KEY", "")

    # Sanitize tools for Gemini if needed
    if self.agent_config.main_llm_provider == ModelProvider.GEMINI:
        self.tools = sanitize_tools_for_gemini(self.tools)

    # Validate tools
    if validate_tools:
        validate_tool_consistency(
            self.tools, self._custom_instructions, self.agent_config
        )

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

    self.session_id = (
        session_id
        or getattr(self, "session_id", None)
        or f"{topic}:{date.today().isoformat()}"
    )

    chat_store = SimpleChatStore()
    self.memory = ChatMemoryBuffer.from_defaults(
        chat_store=chat_store,
        chat_store_key=self.session_id,
        token_limit=65536
    )
    if chat_history:
        msgs = []
        for u, a in chat_history:
            msgs.append(ChatMessage.from_str(u, role=MessageRole.USER))
            msgs.append(ChatMessage.from_str(a, role=MessageRole.ASSISTANT))
        self.memory.put_messages(msgs)

    # Set up main agent and fallback agent
    self._agent = None  # Lazy loading
    self.fallback_agent_config = fallback_agent_config
    self._fallback_agent = None  # Lazy loading

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

    # VHC state tracking
    self._vhc_cache = {}  # Cache VHC results by query hash
    self._last_query = None
    self._last_response = None
    self._current_tool_outputs = []  # Store tool outputs from current query for VHC

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.
    """
    if not prompt:
        return AgentResponse(response="")

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

            # Deal with workflow-based agent types (Function Calling and ReAct)
            if self._get_current_agent_type() in [
                AgentType.FUNCTION_CALLING,
                AgentType.REACT,
            ]:
                from llama_index.core.workflow import Context

                # Create context and pass memory to the workflow agent
                # According to LlamaIndex docs, we should let the workflow manage memory internally
                ctx = Context(current_agent)

                handler = current_agent.run(
                    user_msg=prompt, memory=self.memory, ctx=ctx
                )

                # Listen to workflow events if progress callback is set
                if self.agent_progress_callback:
                    # Create event tracker for consistent event ID generation
                    from .agent_core.streaming import ToolEventTracker

                    event_tracker = ToolEventTracker()

                    async for event in handler.stream_events():
                        # Use consistent event ID tracking to ensure tool calls and outputs are paired
                        event_id = event_tracker.get_event_id(event)

                        # Handle different types of workflow events using same logic as FunctionCallingStreamHandler
                        from llama_index.core.agent.workflow import (
                            ToolCall,
                            ToolCallResult,
                            AgentInput,
                            AgentOutput,
                        )

                        if isinstance(event, ToolCall):
                            self.agent_progress_callback(
                                status_type=AgentStatusType.TOOL_CALL,
                                msg={
                                    "tool_name": event.tool_name,
                                    "arguments": json.dumps(event.tool_kwargs),
                                },
                                event_id=event_id,
                            )
                        elif isinstance(event, ToolCallResult):
                            self.agent_progress_callback(
                                status_type=AgentStatusType.TOOL_OUTPUT,
                                msg={
                                    "tool_name": event.tool_name,
                                    "content": str(event.tool_output),
                                },
                                event_id=event_id,
                            )
                        elif isinstance(event, AgentInput):
                            self.agent_progress_callback(
                                status_type=AgentStatusType.AGENT_UPDATE,
                                msg={"content": f"Agent input: {event.input}"},
                                event_id=event_id,
                            )
                        elif isinstance(event, AgentOutput):
                            self.agent_progress_callback(
                                status_type=AgentStatusType.AGENT_UPDATE,
                                msg={"content": f"Agent output: {event.response}"},
                                event_id=event_id,
                            )

                result = await handler

                # Ensure we have an AgentResponse object with a string response
                if hasattr(result, "response"):
                    response_text = result.response
                else:
                    response_text = str(result)

                # Handle case where response is a ChatMessage object
                if hasattr(response_text, "content"):
                    response_text = response_text.content
                elif not isinstance(response_text, str):
                    response_text = str(response_text)

                if response_text is None or response_text == "None":
                    # Try to find tool outputs in the result object
                    response_text = None

                    # Check various possible locations for tool outputs
                    if hasattr(result, "tool_outputs") and result.tool_outputs:
                        # Get the latest tool output
                        latest_output = (
                            result.tool_outputs[-1]
                            if isinstance(result.tool_outputs, list)
                            else result.tool_outputs
                        )
                        response_text = str(latest_output)

                    # Check if there are tool_calls with results
                    elif hasattr(result, "tool_calls") and result.tool_calls:
                        # Tool calls might contain the outputs - let's try to extract them
                        for tool_call in result.tool_calls:
                            if (
                                hasattr(tool_call, "tool_output")
                                and tool_call.tool_output is not None
                            ):
                                response_text = str(tool_call.tool_output)
                                break

                    elif hasattr(result, "sources") or hasattr(
                        result, "source_nodes"
                    ):
                        sources = getattr(
                            result, "sources", getattr(result, "source_nodes", [])
                        )
                        if (
                            sources
                            and len(sources) > 0
                            and hasattr(sources[0], "text")
                        ):
                            response_text = sources[0].text

                    # Check for workflow context or chat history that might contain tool results
                    elif hasattr(result, "chat_history"):
                        # Look for the most recent assistant message that might contain tool results
                        chat_history = result.chat_history
                        if chat_history and len(chat_history) > 0:
                            for msg in reversed(chat_history):
                                if (
                                    msg.role == MessageRole.TOOL
                                    and msg.content
                                    and str(msg.content).strip()
                                ):
                                    response_text = msg.content
                                    break
                                if (
                                    hasattr(msg, "content")
                                    and msg.content
                                    and str(msg.content).strip()
                                ):
                                    response_text = msg.content
                                    break

                    # If we still don't have a response, provide a fallback
                    if response_text is None or response_text == "None":
                        response_text = "Response completed."

                agent_response = AgentResponse(
                    response=response_text, metadata=getattr(result, "metadata", {})
                )

                # Retrieve updated memory from workflow context
                # According to LlamaIndex docs, workflow agents manage memory internally
                # and we can access it via ctx.store.get("memory")
                try:
                    workflow_memory = await ctx.store.get("memory")
                    if workflow_memory:
                        # Update our external memory with the workflow's memory
                        self.memory = workflow_memory
                except Exception as e:
                    # If we can't retrieve workflow memory, fall back to manual management
                    warning_msg = (
                        f"Could not retrieve workflow memory, falling back to "
                        f"manual management: {e}"
                    )
                    logger.warning(warning_msg)
                    user_msg = ChatMessage.from_str(prompt, role=MessageRole.USER)
                    assistant_msg = ChatMessage.from_str(
                        response_text, role=MessageRole.ASSISTANT
                    )
                    self.memory.put_messages([user_msg, assistant_msg])

            # Standard chat interaction for other agent types
            else:
                agent_response = await current_agent.achat(prompt)

            # Post processing after response is generated
            agent_response.metadata = agent_response.metadata or {}
            user_metadata = agent_response.metadata
            agent_response = await execute_post_stream_processing(
                agent_response, prompt, self, user_metadata
            )
            return agent_response

        except Exception as e:
            last_error = e
            if self.verbose:
                logger.warning(
                    f"LLM call failed on attempt {attempt}. " f"Error: {e}."
                )
            if attempt >= 2 and self.fallback_agent_config:
                self._switch_agent_config()
            await asyncio.sleep(1)
            attempt += 1

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

acompute_vhc() async

Compute VHC for the last query/response pair (async version). Results are cached for subsequent calls. Tool outputs are automatically collected during streaming and used internally.

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Dictionary containing 'corrected_text' and 'corrections'

Source code in vectara_agentic/agent.py

async def acompute_vhc(self) -> Dict[str, Any]:
    """
    Compute VHC for the last query/response pair (async version).
    Results are cached for subsequent calls. Tool outputs are automatically
    collected during streaming and used internally.

    Returns:
        Dict[str, Any]: Dictionary containing 'corrected_text' and 'corrections'
    """
    logging.info(
        f"🔍🔍🔍 [VHC_AGENT_ENTRY] UNIQUE_DEBUG_MESSAGE acompute_vhc method called - "
        f"stored_tool_outputs_count={len(self._current_tool_outputs)}"
    )
    logging.info(
        f"🔍🔍🔍 [VHC_AGENT_ENTRY] _last_query: {'set' if self._last_query else 'None'}"
    )

    if not self._last_query:
        logging.info("🔍 [VHC_AGENT] Returning early - no _last_query")
        return {"corrected_text": None, "corrections": []}

    # For VHC to work, we need the response text from memory
    # Get the latest assistant response from memory
    messages = self.memory.get()
    logging.info(
        f"🔍 [VHC_AGENT] memory.get() returned {len(messages) if messages else 0} messages"
    )

    if not messages:
        logging.info("🔍 [VHC_AGENT] Returning early - no messages in memory")
        return {"corrected_text": None, "corrections": []}

    # Find the last assistant message
    last_response = None
    for msg in reversed(messages):
        if msg.role == MessageRole.ASSISTANT:
            last_response = msg.content
            break

    logging.info(
        f"🔍 [VHC_AGENT] Found last_response: {'set' if last_response else 'None'}"
    )

    if not last_response:
        logging.info("🔍 [VHC_AGENT] Returning early - no last assistant response found")
        return {"corrected_text": None, "corrections": []}

    # Update stored response for caching
    self._last_response = last_response

    # Create cache key from query + response
    cache_key = hash(f"{self._last_query}:{self._last_response}")

    # Return cached results if available
    if cache_key in self._vhc_cache:
        return self._vhc_cache[cache_key]

    # Check if we have VHC API key
    logging.info(
        f"🔍 [VHC_AGENT] acompute_vhc called with vectara_api_key={'set' if self.vectara_api_key else 'None'}"
    )
    if not self.vectara_api_key:
        logging.info("🔍 [VHC_AGENT] No vectara_api_key - returning early with None")
        return {"corrected_text": None, "corrections": []}

    # Compute VHC using existing library function
    from .agent_core.utils.hallucination import analyze_hallucinations

    try:
        # Use stored tool outputs from current query
        stored_tool_outputs = self._get_stored_tool_outputs()
        logging.info(f"🔧 [VHC_AGENT] Using {len(stored_tool_outputs)} stored tool outputs for VHC")

        corrected_text, corrections = analyze_hallucinations(
            query=self._last_query,
            chat_history=self.memory.get(),
            agent_response=self._last_response,
            tools=self.tools,
            vectara_api_key=self.vectara_api_key,
            tool_outputs=stored_tool_outputs,
        )

        # Cache results
        results = {"corrected_text": corrected_text, "corrections": corrections}
        self._vhc_cache[cache_key] = results

        return results

    except Exception as e:
        logger.error(f"VHC computation failed: {e}")
        return {"corrected_text": None, "corrections": []}

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.
    """
    # Store query for VHC processing and clear previous tool outputs
    self._last_query = prompt
    self._clear_tool_outputs()
    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()
            user_meta: Dict[str, Any] = {}

            # Deal with Function Calling agent type
            if self._get_current_agent_type() == AgentType.FUNCTION_CALLING:
                from llama_index.core.workflow import Context

                # Create context and pass memory to the workflow agent
                # According to LlamaIndex docs, we should let the workflow manage memory internally
                ctx = Context(current_agent)

                handler = current_agent.run(
                    user_msg=prompt, memory=self.memory, ctx=ctx
                )

                # Use the dedicated FunctionCallingStreamHandler
                stream_handler = FunctionCallingStreamHandler(self, handler, prompt)
                streaming_adapter = stream_handler.create_streaming_response(
                    user_meta
                )

                return AgentStreamingResponse(
                    base=streaming_adapter, metadata=user_meta
                )

            #
            # For other agent types, use the standard async chat method
            #
            li_stream = await current_agent.astream_chat(prompt)
            orig_async = li_stream.async_response_gen

            # Define a wrapper to preserve streaming behavior while executing post-stream logic.
            async def _stream_response_wrapper():
                async for tok in orig_async():
                    yield tok

                # Use shared post-processing function
                await execute_post_stream_processing(
                    li_stream, prompt, self, user_meta
                )

            li_stream.async_response_gen = _stream_response_wrapper
            return AgentStreamingResponse(base=li_stream, metadata=user_meta)

        except Exception as e:
            last_error = e
            if attempt >= 2 and self.fallback_agent_config:
                self._switch_agent_config()
            await asyncio.sleep(1)
            attempt += 1

    return AgentStreamingResponse.from_error(
        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:
    """
    Interact with the agent using a chat prompt.

    Args:
        prompt (str): The chat prompt.

    Returns:
        AgentResponse: The response from the agent.
    """
    try:
        _ = asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(self.achat(prompt))

    # We are inside a running loop (Jupyter, uvicorn, etc.)
    raise RuntimeError(
        "Use `await agent.achat(...)` inside an event loop (e.g. Jupyter)."
    )

cleanup()

Clean up resources used by the agent.

Source code in vectara_agentic/agent.py

def cleanup(self) -> None:
    """Clean up resources used by the agent."""
    from ._observability import shutdown_observer

    if hasattr(self, 'agent') and hasattr(self.agent, '_llm'):
        llm = self.agent._llm
        if hasattr(llm, 'client') and hasattr(llm.client, 'close'):
            try:
                if asyncio.iscoroutinefunction(llm.client.close):
                    asyncio.run(llm.client.close())
                else:
                    llm.client.close()
            except Exception:
                pass

    # Shutdown observability connections
    shutdown_observer()

clear_memory()

Clear the agent's memory and reset agent instances to ensure consistency.

Source code in vectara_agentic/agent.py

def clear_memory(self) -> None:
    """Clear the agent's memory and reset agent instances to ensure consistency."""
    self.memory.reset()
    # Clear agent instances so they get recreated with the cleared memory
    self._agent = None
    self._fallback_agent = None

compute_vhc()

Compute VHC for the last query/response pair (sync version). Results are cached for subsequent calls. Tool outputs are automatically collected during streaming and used internally.

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Dictionary containing 'corrected_text' and 'corrections'

Source code in vectara_agentic/agent.py

def compute_vhc(self) -> Dict[str, Any]:
    """
    Compute VHC for the last query/response pair (sync version).
    Results are cached for subsequent calls. Tool outputs are automatically
    collected during streaming and used internally.

    Returns:
        Dict[str, Any]: Dictionary containing 'corrected_text' and 'corrections'
    """
    try:
        loop = asyncio.get_event_loop()
        return loop.run_until_complete(self.acompute_vhc())
    except RuntimeError:
        # No event loop running, create a new one
        return asyncio.run(self.acompute_vhc())

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, general_instructions=GENERAL_INSTRUCTIONS, 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, return_direct=False, session_id=None) classmethod

Create an agent from a single Vectara corpus using the factory function.

Parameters:

Name	Type	Description	Default
tool_name	str	Name of the tool to be created.	required
data_description	str	Description of the data/corpus.	required
assistant_specialty	str	The specialty/topic of the assistant.	required
session_id	str	The session ID for memory persistence. If None, auto-generates from topic and date. Defaults to None.	None

Source code in vectara_agentic/agent.py

@classmethod
def from_corpus(
    cls,
    tool_name: str,
    data_description: str,
    assistant_specialty: str,
    general_instructions: str = GENERAL_INSTRUCTIONS,
    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, dict, 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: Optional[str] = None,
    vectara_rerank_chain: Optional[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,
    return_direct: bool = False,
    session_id: Optional[str] = None,
) -> "Agent":
    """Create an agent from a single Vectara corpus using the factory function.

    Args:
        tool_name (str): Name of the tool to be created.
        data_description (str): Description of the data/corpus.
        assistant_specialty (str): The specialty/topic of the assistant.
        session_id (str, optional): The session ID for memory persistence.
                                    If None, auto-generates from topic and date. Defaults to None.
        ... (other parameters as documented in factory function)
    """
    # Use the factory function to avoid code duplication
    config = create_agent_from_corpus(
        tool_name=tool_name,
        data_description=data_description,
        assistant_specialty=assistant_specialty,
        general_instructions=general_instructions,
        vectara_corpus_key=vectara_corpus_key,
        vectara_api_key=vectara_api_key,
        agent_config=agent_config,
        fallback_agent_config=fallback_agent_config,
        verbose=verbose,
        vectara_filter_fields=vectara_filter_fields,
        vectara_offset=vectara_offset,
        vectara_lambda_val=vectara_lambda_val,
        vectara_semantics=vectara_semantics,
        vectara_custom_dimensions=vectara_custom_dimensions,
        vectara_reranker=vectara_reranker,
        vectara_rerank_k=vectara_rerank_k,
        vectara_rerank_limit=vectara_rerank_limit,
        vectara_rerank_cutoff=vectara_rerank_cutoff,
        vectara_diversity_bias=vectara_diversity_bias,
        vectara_udf_expression=vectara_udf_expression,
        vectara_rerank_chain=vectara_rerank_chain,
        vectara_n_sentences_before=vectara_n_sentences_before,
        vectara_n_sentences_after=vectara_n_sentences_after,
        vectara_summary_num_results=vectara_summary_num_results,
        vectara_summarizer=vectara_summarizer,
        vectara_summary_response_language=vectara_summary_response_language,
        vectara_summary_prompt_text=vectara_summary_prompt_text,
        vectara_max_response_chars=vectara_max_response_chars,
        vectara_max_tokens=vectara_max_tokens,
        vectara_temperature=vectara_temperature,
        vectara_frequency_penalty=vectara_frequency_penalty,
        vectara_presence_penalty=vectara_presence_penalty,
        vectara_save_history=vectara_save_history,
        return_direct=return_direct,
    )

    return cls(
        chat_history=chat_history,
        agent_progress_callback=agent_progress_callback,
        query_logging_callback=query_logging_callback,
        session_id=session_id,
        **config,
    )

from_dict(data, agent_progress_callback=None, query_logging_callback=None) 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_progress_callback: Optional[Callable] = None,
    query_logging_callback: Optional[Callable] = None,
) -> "Agent":
    """Create an Agent instance from a dictionary."""
    return deserialize_agent_from_dict(
        cls, data, agent_progress_callback, query_logging_callback
    )

from_tools(tools, topic='general', custom_instructions='', verbose=True, 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, session_id=None) classmethod

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

Args:

tools (list[FunctionTool]): A list of tools to be used by the agent.
topic (str, optional): The topic for the agent. Defaults to 'general'.
custom_instructions (str, optional): custom instructions for the agent. Defaults to ''.
verbose (bool, optional): Whether the agent should print its steps. Defaults to True.
agent_progress_callback (Callable): A callback function the code calls on any agent updates.
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.
session_id (str, optional): The session ID for memory persistence.
                            If None, auto-generates from topic and date. Defaults to None.

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,
    agent_progress_callback: Optional[
        Callable[[AgentStatusType, dict, 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: Optional["Workflow"] = None,
    workflow_timeout: int = 120,
    session_id: Optional[str] = None,
) -> "Agent":
    """
    Create an agent from tools, agent type, and language model.

    Args:

        tools (list[FunctionTool]): A list of tools to be used by the agent.
        topic (str, optional): The topic for the agent. Defaults to 'general'.
        custom_instructions (str, optional): custom instructions for the agent. Defaults to ''.
        verbose (bool, optional): Whether the agent should print its steps. Defaults to True.
        agent_progress_callback (Callable): A callback function the code calls on any agent updates.
        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.
        session_id (str, optional): The session ID for memory persistence.
                                    If None, auto-generates from topic and date. Defaults to None.

    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,
        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,
        session_id=session_id,
    )

loads(data, agent_progress_callback=None, query_logging_callback=None) classmethod

Create an Agent instance from a JSON string.

Source code in vectara_agentic/agent.py

@classmethod
def loads(
    cls,
    data: str,
    agent_progress_callback: Optional[
        Callable[[AgentStatusType, dict, str], None]
    ] = None,
    query_logging_callback: Optional[Callable[[str, str], None]] = None,
) -> "Agent":
    """Create an Agent instance from a JSON string."""
    return cls.from_dict(
        json.loads(data), agent_progress_callback, query_logging_callback
    )

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.
    """
    logger.info("Vectara agentic Report:")
    logger.info(f"Agent Type = {self.agent_config.agent_type}")
    logger.info(f"Topic = {self._topic}")
    logger.info("Tools:")
    for tool in self.tools:
        if hasattr(tool, "metadata"):
            if detailed:
                logger.info(f"- {tool.metadata.description}")
            else:
                logger.info(f"- {tool.metadata.name}")
        else:
            logger.info("- tool without metadata")
    logger.info(
        f"Agent LLM = {get_llm(LLMRole.MAIN, config=self.agent_config).metadata.model_name}"
    )
    logger.info(
        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 or context 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 or context 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}.")

    outputs_model_on_fail_cls = getattr(
        workflow.__class__, "OutputModelOnFail", None
    )
    if outputs_model_on_fail_cls:
        fields_without_default = []
        for name, field_info in outputs_model_on_fail_cls.model_fields.items():
            if field_info.default_factory is PydanticUndefined:
                fields_without_default.append(name)
        if fields_without_default:
            raise ValueError(
                f"Fields without default values: {fields_without_default}"
            )

    from llama_index.core.workflow import Context

    workflow_context = Context(workflow=workflow)
    try:
        # run workflow
        result = await workflow.run(
            ctx=workflow_context,
            agent=self,
            tools=self.tools,
            llm=self.llm,
            verbose=verbose,
            inputs=inputs,
        )

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

    except Exception as e:
        _missing = object()
        if outputs_model_on_fail_cls:
            model_fields = outputs_model_on_fail_cls.model_fields
            input_dict = {}
            for key in model_fields:
                value = await workflow_context.get(key, default=_missing)
                if value is not _missing:
                    input_dict[key] = value
            output = outputs_model_on_fail_cls.model_validate(input_dict)
        else:
            logger.warning(
                f"Vectara Agentic: Workflow failed with unexpected error: {e}"
            )
            raise type(e)(str(e)).with_traceback(e.__traceback__)

    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:
    """
    Interact with the agent using a chat prompt with streaming.
    Args:
        prompt (str): The chat prompt.
    Returns:
        AgentStreamingResponse: The streaming response from the agent.
    """
    try:
        _ = asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(self.astream_chat(prompt))
    raise RuntimeError(
        "Use `await agent.astream_chat(...)` inside an event loop (e.g. Jupyter)."
    )

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."""
    return serialize_agent_to_dict(self)

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.FUNCTION_CALLING.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")
    )

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

    @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"],
        )

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"],
    )

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

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"
    FUNCTION_CALLING = "FUNCTION_CALLING"

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"
    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,
        vhc_eligible: bool = True,
    ) -> 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, vhc_eligible=vhc_eligible
        )

    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, vhc_eligible=True)
            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, vhc_eligible=False)]

    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, vhc_eligible=False)
            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,
                vhc_eligible=True,
            )
            vtools.append(vtool)
        return vtools

create_tool(function, tool_type=ToolType.QUERY, vhc_eligible=True)

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,
    vhc_eligible: bool = True,
) -> 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, vhc_eligible=vhc_eligible
    )

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,
            vhc_eligible=True,
        )
        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, vhc_eligible=False)]

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, vhc_eligible=False)
        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, vhc_eligible=True)
        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/tool_utils.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,
        vhc_eligible: bool = True,
    ) -> None:
        # Use Pydantic v2 compatible method for extracting metadata
        metadata_dict = (
            metadata.model_dump()
            if hasattr(metadata, "model_dump")
            else metadata.dict() if hasattr(metadata, "dict") else metadata.__dict__
        )
        vm = VectaraToolMetadata(
            tool_type=tool_type, vhc_eligible=vhc_eligible, **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,
        partial_params: Optional[Dict[str, Any]] = None,
        tool_type: ToolType = ToolType.QUERY,
        vhc_eligible: bool = True,
    ) -> "VectaraTool":
        tool = FunctionTool.from_defaults(
            fn,
            name,
            description,
            return_direct,
            fn_schema,
            async_fn,
            tool_metadata,
            callback,
            async_callback,
            partial_params,
        )
        vectara_tool = cls(
            tool_type=tool_type,
            fn=tool.fn,
            metadata=tool.metadata,
            async_fn=tool.async_fn,
            vhc_eligible=vhc_eligible,
        )
        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):
        try:
            # Try to get schema as dict if possible
            self_schema = self.metadata.fn_schema.model_json_schema()
            other_schema = other.metadata.fn_schema.model_json_schema()
        except Exception:
            return False

        is_equal = (
            isinstance(other, VectaraTool)
            and self.metadata.tool_type == other.metadata.tool_type
            and self.metadata.name == other.metadata.name
            and self_schema == other_schema
        )
        return is_equal

    def _create_tool_error_output(
        self, error: Exception, args: Any, kwargs: Any, include_traceback: bool = False
    ) -> ToolOutput:
        """Create standardized error output for tool execution failures."""
        if isinstance(error, TypeError):
            # Parameter validation error handling
            sig = inspect.signature(self.metadata.fn_schema)
            valid_parameters = list(sig.parameters.keys())
            params_str = ", ".join(valid_parameters)
            return ToolOutput(
                tool_name=self.metadata.name,
                content=(
                    f"Wrong argument used when calling {self.metadata.name}: {str(error)}. "
                    f"Valid arguments: {params_str}. Please call the tool again with the correct arguments."
                ),
                raw_input={"args": args, "kwargs": kwargs},
                raw_output={"response": str(error)},
            )
        else:
            # General execution error handling
            content = f"Tool {self.metadata.name} Malfunction: {str(error)}"
            if include_traceback:
                content += f", traceback: {traceback.format_exc()}"

            return ToolOutput(
                tool_name=self.metadata.name,
                content=content,
                raw_input={"args": args, "kwargs": kwargs},
                raw_output={"response": str(error)},
            )

    def call(
        self, *args: Any, ctx: Optional[Context] = None, **kwargs: Any
    ) -> ToolOutput:
        try:
            # Only pass ctx if it's not None to avoid passing unwanted kwargs to the function
            if ctx is not None:
                result = super().call(*args, ctx=ctx, **kwargs)
            else:
                result = super().call(*args, **kwargs)
            return self._format_tool_output(result)
        except Exception as e:
            return self._create_tool_error_output(e, args, kwargs)

    async def acall(
        self, *args: Any, ctx: Optional[Context] = None, **kwargs: Any
    ) -> ToolOutput:
        try:
            # Only pass ctx if it's not None to avoid passing unwanted kwargs to the function
            if ctx is not None:
                result = await super().acall(*args, ctx=ctx, **kwargs)
            else:
                result = await super().acall(*args, **kwargs)
            return self._format_tool_output(result)
        except Exception as e:
            return self._create_tool_error_output(
                e, args, kwargs, include_traceback=True
            )

    def _format_tool_output(self, result: ToolOutput) -> ToolOutput:
        """Format tool output by converting human-readable wrappers to formatted content immediately."""
        import logging

        # If the raw_output has human-readable formatting, use it for the content
        if hasattr(result, "raw_output") and _is_human_readable_output(
            result.raw_output
        ):
            try:
                formatted_content = result.raw_output.to_human_readable()
                # Replace the content with the formatted version
                result.content = formatted_content
            except Exception as e:
                logging.warning(
                    f"{self.metadata.name}: Failed to convert to human-readable: {e}"
                )

        return result

VectaraToolFactory

A factory class for creating Vectara RAG tools.

Source code in vectara_agentic/tools.py

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

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

    def create_search_tool(
        self,
        tool_name: str,
        tool_description: str,
        tool_args_schema: type[BaseModel] = None,
        tool_args_type: Dict[str, str] = {},
        summarize_docs: Optional[bool] = None,
        summarize_llm_name: Optional[str] = None,
        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,
        return_direct: bool = False,
        save_history: bool = True,
        verbose: bool = False,
        vectara_base_url: str = "https://api.vectara.io",
        vectara_verify_ssl: bool = True,
        vhc_eligible: 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, optional): 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.
                - 'filter_name': the name of the filterable attribute in Vectara.
            summarize_docs (bool, optional): Whether to summarize the retrieved documents.
            summarize_llm_name (str, optional): The name of the LLM to use for summarization.
            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.
            return_direct (bool, optional): Whether the agent should return the tool's response directly.
            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) -> list[dict]:
            """
            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)
                if summarize_docs is None
                else summarize_docs
            )
            try:
                filter_string = build_filter_string(
                    kwargs, tool_args_type, fixed_filter
                )
            except ValueError as e:
                msg = (
                    f"Building filter string failed in search tool due to invalid input or configuration ({e}). "
                    "Please verify the input arguments and ensure they meet the expected format or conditions."
                )
                return [{"text": msg, "metadata": {"args": args, "kwargs": kwargs}}]

            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 retrieve any results for the query."
                return [{"text": msg, "metadata": {"args": args, "kwargs": kwargs}}]
            unique_ids = set()
            docs = []
            doc_matches = {}
            for doc in response:
                if doc.id_ in unique_ids:
                    doc_matches[doc.id_].append(doc.node.get_content())
                    continue
                unique_ids.add(doc.id_)
                doc_matches[doc.id_] = [doc.node.get_content()]
                docs.append((doc.id_, doc.metadata))

            res = []
            if summarize:
                summaries_dict = asyncio.run(
                    summarize_documents(
                        corpus_key=self.vectara_corpus_key,
                        api_key=self.vectara_api_key,
                        llm_name=summarize_llm_name,
                        doc_ids=list(unique_ids),
                    )
                )
            else:
                summaries_dict = {}

            for doc_id, metadata in docs:
                res.append(
                    {
                        "text": summaries_dict.get(doc_id, "") if summarize else "",
                        "metadata": {
                            "document_id": doc_id,
                            "metadata": metadata,
                            "matching_text": doc_matches[doc_id],
                        },
                    }
                )

            # Create human-readable output using sequential format
            def format_search_results(results):
                if not results:
                    return "No search results found"

                # Create a sequential view for human reading
                formatted_results = []
                for i, result in enumerate(results, 1):
                    result_str = f"**Result #{i}**\n"
                    result_str += f"Document ID: {result['metadata']['document_id']}\n"
                    if summarize and result["text"]:
                        result_str += f"Summary: {result['text']}\n"

                    # Add all matching text if available
                    matches = result["metadata"]["matching_text"]
                    if matches:
                        result_str += "".join(
                            f"Match #{inx} Text: {match}\n"
                            for inx, match in enumerate(matches, 1)
                        )
                    formatted_results.append(result_str)
                return "\n".join(formatted_results)

            return create_human_readable_output(res, format_search_results)

        class SearchToolBaseParams(BaseModel):
            """Model for the base parameters of the search tool."""

            query: str = Field(
                ...,
                description="The search query to perform, in the form of a question.",
            )
            top_k: int = Field(
                default=10, description="The number of top documents to retrieve."
            )
            summarize: bool = Field(
                True,
                description="Whether to summarize the retrieved documents.",
            )

        class SearchToolBaseParamsWithoutSummarize(BaseModel):
            """Model for the base parameters of the search tool."""

            query: str = Field(
                ...,
                description="The search query to perform, in the form of a question.",
            )
            top_k: int = Field(
                default=10, description="The number of top documents to retrieve."
            )

        search_tool_extra_desc = (
            tool_description
            + "\n"
            + "Use this tool to search for relevant documents, not to ask questions."
        )

        tool = create_tool_from_dynamic_function(
            search_function,
            tool_name,
            search_tool_extra_desc,
            (
                SearchToolBaseParams
                if summarize_docs is None
                else SearchToolBaseParamsWithoutSummarize
            ),
            tool_args_schema,
            return_direct=return_direct,
            vhc_eligible=vhc_eligible,
        )
        return tool

    def create_rag_tool(
        self,
        tool_name: str,
        tool_description: str,
        tool_args_schema: type[BaseModel] = None,
        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,
        llm_name: Optional[str] = None,
        temperature: Optional[float] = None,
        frequency_penalty: Optional[float] = None,
        presence_penalty: Optional[float] = None,
        include_citations: bool = True,
        citation_pattern: str = None,
        citation_url_pattern: str = "{doc.url}",
        citation_text_pattern: str = "{doc.title}",
        save_history: bool = False,
        fcs_threshold: float = 0.0,
        return_direct: bool = False,
        verbose: bool = False,
        vectara_base_url: str = "https://api.vectara.io",
        vectara_verify_ssl: bool = True,
        vhc_eligible: 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, optional): The schema for any tool arguments for filtering.
            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.
                - 'filter_name': the name of the filterable attribute in Vectara.
            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.
            llm_name (str, optional): The name of the LLM to use for generation.
            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.
            citation_url_pattern (str, optional): The pattern for the citations in the response.
                Default is "{doc.url}" which uses the document URL.
                If include_citations is False, this parameter is ignored.
                citation_pattern (str, optional): old name for citation_url_pattern. Deprecated.
            citation_text_pattern (str, optional): The text pattern for citations in the response.
                Default is "{doc.title}" which uses the title of the document.
                If include_citations is False, this parameter is ignored.
            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.
            return_direct (bool, optional): Whether the agent should return the tool's response directly.
            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,
        )
        keys_to_ignore = ["lang", "offset", "len"]

        # Dynamically generate the RAG function
        def rag_function(*args: Any, **kwargs: Any) -> dict:
            """
            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:
                msg = (
                    f"Building filter string failed in rag tool. "
                    f"Reason: {e}. Ensure that the input arguments match the expected "
                    f"format and include all required fields. "
                )
                return {"text": msg, "metadata": {"args": args, "kwargs": kwargs}}

            citations_url_pattern = (
                (
                    citation_url_pattern
                    if citation_url_pattern is not None
                    else citation_pattern
                )
                if include_citations
                else None
            )
            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,
                llm_name=llm_name,
                temperature=temperature,
                frequency_penalty=frequency_penalty,
                presence_penalty=presence_penalty,
                citations_style="markdown" if include_citations else None,
                citations_url_pattern=citations_url_pattern,
                citations_text_pattern=(
                    citation_text_pattern 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. "
                    "Please check the arguments and try again."
                )
                kwargs["query"] = query
                return {"text": msg, "metadata": {"args": args, "kwargs": kwargs}}
            if str(response) == "None":
                msg = "Tool failed to generate a response."
                kwargs["query"] = query
                return {"text": msg, "metadata": {"args": args, "kwargs": kwargs}}

            # Extract citation metadata
            pattern = r"\[(\d+)\]"
            matches = re.findall(pattern, response.response)
            citation_numbers = sorted(set(int(match) for match in matches))
            citation_metadata = {}
            for citation_number in citation_numbers:
                metadata = {
                    k: v
                    for k, v in response.source_nodes[
                        citation_number - 1
                    ].metadata.items()
                    if k not in keys_to_ignore
                }
                citation_metadata[str(citation_number)] = metadata
            fcs = 0.0
            fcs_str = response.metadata["fcs"] if "fcs" in response.metadata else "0.0"
            if fcs_str and is_float(fcs_str):
                fcs = float(fcs_str)
                if fcs < fcs_threshold:
                    msg = f"Could not answer the query due to suspected hallucination (fcs={fcs})."
                    return {
                        "text": msg,
                        "metadata": {"args": args, "kwargs": kwargs, "fcs": fcs},
                    }
            if fcs:
                citation_metadata["fcs"] = fcs
            res = {"text": response.response, "metadata": citation_metadata}

            # Create human-readable output with citation formatting
            def format_rag_response(result):
                text = result["text"]
                return text

            return create_human_readable_output(res, format_rag_response)

        class RagToolBaseParams(BaseModel):
            """Model for the base parameters of the RAG tool."""

            query: str = Field(
                ...,
                description="The search query to perform, in the form of a question",
            )

        tool = create_tool_from_dynamic_function(
            rag_function,
            tool_name,
            tool_description,
            RagToolBaseParams,
            tool_args_schema,
            return_direct=return_direct,
            vhc_eligible=vhc_eligible,
        )
        return tool

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

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

Source code in vectara_agentic/tools.py

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

create_rag_tool(tool_name, tool_description, tool_args_schema=None, 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, llm_name=None, temperature=None, frequency_penalty=None, presence_penalty=None, include_citations=True, citation_pattern=None, citation_url_pattern='{doc.url}', citation_text_pattern='{doc.title}', save_history=False, fcs_threshold=0.0, return_direct=False, verbose=False, vectara_base_url='https://api.vectara.io', vectara_verify_ssl=True, vhc_eligible=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 any tool arguments for filtering.	None
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. - 'filter_name': the name of the filterable attribute in Vectara.	{}
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
llm_name	str	The name of the LLM to use for generation.	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
citation_url_pattern	str	The pattern for the citations in the response. Default is "{doc.url}" which uses the document URL. If include_citations is False, this parameter is ignored. citation_pattern (str, optional): old name for citation_url_pattern. Deprecated.	'{doc.url}'
citation_text_pattern	str	The text pattern for citations in the response. Default is "{doc.title}" which uses the title of the document. If include_citations is False, this parameter is ignored.	'{doc.title}'
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
return_direct	bool	Whether the agent should return the tool's response directly.	False
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] = None,
    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,
    llm_name: Optional[str] = None,
    temperature: Optional[float] = None,
    frequency_penalty: Optional[float] = None,
    presence_penalty: Optional[float] = None,
    include_citations: bool = True,
    citation_pattern: str = None,
    citation_url_pattern: str = "{doc.url}",
    citation_text_pattern: str = "{doc.title}",
    save_history: bool = False,
    fcs_threshold: float = 0.0,
    return_direct: bool = False,
    verbose: bool = False,
    vectara_base_url: str = "https://api.vectara.io",
    vectara_verify_ssl: bool = True,
    vhc_eligible: 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, optional): The schema for any tool arguments for filtering.
        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.
            - 'filter_name': the name of the filterable attribute in Vectara.
        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.
        llm_name (str, optional): The name of the LLM to use for generation.
        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.
        citation_url_pattern (str, optional): The pattern for the citations in the response.
            Default is "{doc.url}" which uses the document URL.
            If include_citations is False, this parameter is ignored.
            citation_pattern (str, optional): old name for citation_url_pattern. Deprecated.
        citation_text_pattern (str, optional): The text pattern for citations in the response.
            Default is "{doc.title}" which uses the title of the document.
            If include_citations is False, this parameter is ignored.
        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.
        return_direct (bool, optional): Whether the agent should return the tool's response directly.
        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,
    )
    keys_to_ignore = ["lang", "offset", "len"]

    # Dynamically generate the RAG function
    def rag_function(*args: Any, **kwargs: Any) -> dict:
        """
        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:
            msg = (
                f"Building filter string failed in rag tool. "
                f"Reason: {e}. Ensure that the input arguments match the expected "
                f"format and include all required fields. "
            )
            return {"text": msg, "metadata": {"args": args, "kwargs": kwargs}}

        citations_url_pattern = (
            (
                citation_url_pattern
                if citation_url_pattern is not None
                else citation_pattern
            )
            if include_citations
            else None
        )
        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,
            llm_name=llm_name,
            temperature=temperature,
            frequency_penalty=frequency_penalty,
            presence_penalty=presence_penalty,
            citations_style="markdown" if include_citations else None,
            citations_url_pattern=citations_url_pattern,
            citations_text_pattern=(
                citation_text_pattern 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. "
                "Please check the arguments and try again."
            )
            kwargs["query"] = query
            return {"text": msg, "metadata": {"args": args, "kwargs": kwargs}}
        if str(response) == "None":
            msg = "Tool failed to generate a response."
            kwargs["query"] = query
            return {"text": msg, "metadata": {"args": args, "kwargs": kwargs}}

        # Extract citation metadata
        pattern = r"\[(\d+)\]"
        matches = re.findall(pattern, response.response)
        citation_numbers = sorted(set(int(match) for match in matches))
        citation_metadata = {}
        for citation_number in citation_numbers:
            metadata = {
                k: v
                for k, v in response.source_nodes[
                    citation_number - 1
                ].metadata.items()
                if k not in keys_to_ignore
            }
            citation_metadata[str(citation_number)] = metadata
        fcs = 0.0
        fcs_str = response.metadata["fcs"] if "fcs" in response.metadata else "0.0"
        if fcs_str and is_float(fcs_str):
            fcs = float(fcs_str)
            if fcs < fcs_threshold:
                msg = f"Could not answer the query due to suspected hallucination (fcs={fcs})."
                return {
                    "text": msg,
                    "metadata": {"args": args, "kwargs": kwargs, "fcs": fcs},
                }
        if fcs:
            citation_metadata["fcs"] = fcs
        res = {"text": response.response, "metadata": citation_metadata}

        # Create human-readable output with citation formatting
        def format_rag_response(result):
            text = result["text"]
            return text

        return create_human_readable_output(res, format_rag_response)

    class RagToolBaseParams(BaseModel):
        """Model for the base parameters of the RAG tool."""

        query: str = Field(
            ...,
            description="The search query to perform, in the form of a question",
        )

    tool = create_tool_from_dynamic_function(
        rag_function,
        tool_name,
        tool_description,
        RagToolBaseParams,
        tool_args_schema,
        return_direct=return_direct,
        vhc_eligible=vhc_eligible,
    )
    return tool

create_search_tool(tool_name, tool_description, tool_args_schema=None, tool_args_type={}, summarize_docs=None, summarize_llm_name=None, 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, return_direct=False, save_history=True, verbose=False, vectara_base_url='https://api.vectara.io', vectara_verify_ssl=True, vhc_eligible=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.	None
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. - 'filter_name': the name of the filterable attribute in Vectara.	{}
summarize_docs	bool	Whether to summarize the retrieved documents.	None
summarize_llm_name	str	The name of the LLM to use for summarization.	None
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
return_direct	bool	Whether the agent should return the tool's response directly.	False
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] = None,
    tool_args_type: Dict[str, str] = {},
    summarize_docs: Optional[bool] = None,
    summarize_llm_name: Optional[str] = None,
    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,
    return_direct: bool = False,
    save_history: bool = True,
    verbose: bool = False,
    vectara_base_url: str = "https://api.vectara.io",
    vectara_verify_ssl: bool = True,
    vhc_eligible: 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, optional): 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.
            - 'filter_name': the name of the filterable attribute in Vectara.
        summarize_docs (bool, optional): Whether to summarize the retrieved documents.
        summarize_llm_name (str, optional): The name of the LLM to use for summarization.
        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.
        return_direct (bool, optional): Whether the agent should return the tool's response directly.
        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) -> list[dict]:
        """
        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)
            if summarize_docs is None
            else summarize_docs
        )
        try:
            filter_string = build_filter_string(
                kwargs, tool_args_type, fixed_filter
            )
        except ValueError as e:
            msg = (
                f"Building filter string failed in search tool due to invalid input or configuration ({e}). "
                "Please verify the input arguments and ensure they meet the expected format or conditions."
            )
            return [{"text": msg, "metadata": {"args": args, "kwargs": kwargs}}]

        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 retrieve any results for the query."
            return [{"text": msg, "metadata": {"args": args, "kwargs": kwargs}}]
        unique_ids = set()
        docs = []
        doc_matches = {}
        for doc in response:
            if doc.id_ in unique_ids:
                doc_matches[doc.id_].append(doc.node.get_content())
                continue
            unique_ids.add(doc.id_)
            doc_matches[doc.id_] = [doc.node.get_content()]
            docs.append((doc.id_, doc.metadata))

        res = []
        if summarize:
            summaries_dict = asyncio.run(
                summarize_documents(
                    corpus_key=self.vectara_corpus_key,
                    api_key=self.vectara_api_key,
                    llm_name=summarize_llm_name,
                    doc_ids=list(unique_ids),
                )
            )
        else:
            summaries_dict = {}

        for doc_id, metadata in docs:
            res.append(
                {
                    "text": summaries_dict.get(doc_id, "") if summarize else "",
                    "metadata": {
                        "document_id": doc_id,
                        "metadata": metadata,
                        "matching_text": doc_matches[doc_id],
                    },
                }
            )

        # Create human-readable output using sequential format
        def format_search_results(results):
            if not results:
                return "No search results found"

            # Create a sequential view for human reading
            formatted_results = []
            for i, result in enumerate(results, 1):
                result_str = f"**Result #{i}**\n"
                result_str += f"Document ID: {result['metadata']['document_id']}\n"
                if summarize and result["text"]:
                    result_str += f"Summary: {result['text']}\n"

                # Add all matching text if available
                matches = result["metadata"]["matching_text"]
                if matches:
                    result_str += "".join(
                        f"Match #{inx} Text: {match}\n"
                        for inx, match in enumerate(matches, 1)
                    )
                formatted_results.append(result_str)
            return "\n".join(formatted_results)

        return create_human_readable_output(res, format_search_results)

    class SearchToolBaseParams(BaseModel):
        """Model for the base parameters of the search tool."""

        query: str = Field(
            ...,
            description="The search query to perform, in the form of a question.",
        )
        top_k: int = Field(
            default=10, description="The number of top documents to retrieve."
        )
        summarize: bool = Field(
            True,
            description="Whether to summarize the retrieved documents.",
        )

    class SearchToolBaseParamsWithoutSummarize(BaseModel):
        """Model for the base parameters of the search tool."""

        query: str = Field(
            ...,
            description="The search query to perform, in the form of a question.",
        )
        top_k: int = Field(
            default=10, description="The number of top documents to retrieve."
        )

    search_tool_extra_desc = (
        tool_description
        + "\n"
        + "Use this tool to search for relevant documents, not to ask questions."
    )

    tool = create_tool_from_dynamic_function(
        search_function,
        tool_name,
        search_tool_extra_desc,
        (
            SearchToolBaseParams
            if summarize_docs is None
            else SearchToolBaseParamsWithoutSummarize
        ),
        tool_args_schema,
        return_direct=return_direct,
        vhc_eligible=vhc_eligible,
    )
    return tool

create_app(agent, config)

Create and configure the FastAPI app.

Parameters:

Name	Type	Description	Default
agent	Agent	The agent instance to handle chat/completion.	required
config	AgentConfig	Configuration containing the API key.	required

Returns:

Name	Type	Description
FastAPI	FastAPI	Configured FastAPI application.

Source code in vectara_agentic/agent_endpoint.py

def create_app(agent: Agent, config: AgentConfig) -> FastAPI:
    """
    Create and configure the FastAPI app.

    Args:
        agent (Agent): The agent instance to handle chat/completion.
        config (AgentConfig): Configuration containing the API key.

    Returns:
        FastAPI: Configured FastAPI application.
    """
    app = FastAPI()
    logger = logging.getLogger("uvicorn.error")
    logging.basicConfig(level=logging.INFO)

    api_key_header = APIKeyHeader(name="X-API-Key")

    async def _verify_api_key(api_key: str = Depends(api_key_header)):
        """
        Dependency that verifies the X-API-Key header.

        Raises:
            HTTPException(403): If the provided key does not match.

        Returns:
            bool: True if key is valid.
        """
        if api_key != config.endpoint_api_key:
            raise HTTPException(status_code=403, detail="Unauthorized")
        return True

    @app.get(
        "/chat", summary="Chat with the agent", dependencies=[Depends(_verify_api_key)]
    )
    async def chat(message: str):
        """
        Handle GET /chat requests.

        Args:
            message (str): The user's message to the agent.

        Returns:
            dict: Contains the agent's response under 'response'.

        Raises:
            HTTPException(400): If message is empty.
            HTTPException(500): On internal errors.
        """
        if not message:
            raise HTTPException(status_code=400, detail="No message provided")
        try:
            res = agent.chat(message)
            return {"response": res}
        except Exception as e:
            raise HTTPException(status_code=500, detail="Internal server error") from e

    @app.post(
        "/v1/completions",
        response_model=CompletionResponse,
        dependencies=[Depends(_verify_api_key)],
    )
    async def completions(req: CompletionRequest):
        """
        Handle POST /v1/completions requests.

        Args:
            req (CompletionRequest): The completion request payload.

        Returns:
            CompletionResponse: The generated completion and usage stats.

        Raises:
            HTTPException(400): If prompt is missing.
            HTTPException(500): On internal errors.
        """
        if not req.prompt:
            raise HTTPException(status_code=400, detail="`prompt` is required")
        raw = req.prompt if isinstance(req.prompt, str) else req.prompt[0]
        try:
            start = time.time()
            text = agent.chat(raw)
            logger.info(f"Agent returned in {time.time()-start:.2f}s")
        except Exception as e:
            raise HTTPException(status_code=500, detail="Internal server error") from e

        p_tokens = len(raw.split())
        c_tokens = len(text.split())

        return CompletionResponse(
            id=f"cmpl-{uuid.uuid4()}",
            object="text_completion",
            created=int(time.time()),
            model=req.model,
            choices=[Choice(text=text, index=0, logprobs=None, finish_reason="stop")],
            usage=CompletionUsage(
                prompt_tokens=p_tokens,
                completion_tokens=c_tokens,
                total_tokens=p_tokens + c_tokens,
            ),
        )

    @app.post(
        "/v1/chat",
        response_model=ChatCompletionResponse,
        dependencies=[Depends(_verify_api_key)],
    )
    async def chat_completion(req: ChatCompletionRequest):
        if not req.messages:
            raise HTTPException(status_code=400, detail="`messages` is required")

        # concatenate all user messages into a single prompt
        raw = " ".join(m.content for m in req.messages if m.role == "user")

        try:
            start = time.time()
            text = agent.chat(raw)
            logger.info(f"Agent returned in {time.time()-start:.2f}s")
        except Exception as e:
            raise HTTPException(status_code=500, detail="Internal server error") from e

        p_tokens = len(raw.split())
        c_tokens = len(text.split())

        return ChatCompletionResponse(
            id=f"chatcmpl-{uuid.uuid4()}",
            object="chat.completion",
            created=int(time.time()),
            model=req.model,
            choices=[
                ChatCompletionChoice(
                    index=0,
                    message=ChatMessage(role="assistant", content=text),
                    finish_reason="stop",
                )
            ],
            usage=CompletionUsage(
                prompt_tokens=p_tokens,
                completion_tokens=c_tokens,
                total_tokens=p_tokens + c_tokens,
            ),
        )

    return app

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

Launch the FastAPI application using Uvicorn.

Parameters:

Name	Type	Description	Default
agent	Agent	The agent instance for request handling.	required
host	str	Host interface. Defaults to "0.0.0.0".	'0.0.0.0'
port	int	Port number. Defaults to 8000.	8000

Source code in vectara_agentic/agent_endpoint.py

def start_app(agent: Agent, host="0.0.0.0", port=8000):
    """
    Launch the FastAPI application using Uvicorn.

    Args:
        agent (Agent): The agent instance for request handling.
        host (str, optional): Host interface. Defaults to "0.0.0.0".
        port (int, optional): Port number. Defaults to 8000.
    """
    app = create_app(agent, config=AgentConfig())
    uvicorn.run(app, host=host, port=port)