Skip to content

API Reference

chat_history

get_shell_history(count)

Return the last count commands from the current shell's history file.

Source code in cli/chat_history.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def get_shell_history(count: int) -> str:
    """Return the last `count` commands from the current shell's history file."""
    try:
        # zsh metafies non-ASCII bytes, so the file is not always valid UTF-8.
        with open(history_file(), errors="replace") as f:
            lines = f.read().splitlines()
    except OSError:
        return ""

    commands: list[str] = []
    for line in lines:
        if commands and commands[-1].endswith("\\"):
            # zsh stores a multi-line command as backslash-continued lines.
            commands[-1] = commands[-1][:-1] + "\n" + line
        else:
            commands.append(ZSH_ENTRY.sub("", line))
    return "".join(f"{command}\n" for command in commands[-count:])

history_file()

Path to the current shell's history file.

Source code in cli/chat_history.py 
46
47
48
49
50
def history_file() -> str:
    """Path to the current shell's history file."""
    shell = os.path.basename(os.environ.get("SHELL", ""))
    default = "~/.zsh_history" if shell == "zsh" else "~/.bash_history"
    return os.path.expanduser(os.environ.get("HISTFILE") or default)

config_manager

Configuration manager for AI models.

ConfigurationManager

Manages model configurations stored in JSON.

Source code in cli/config_manager.py 
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
class ConfigurationManager:
    """Manages model configurations stored in JSON."""

    def __init__(self, file_path: Optional[str] = None):
        if file_path is None:
            config_dir = os.path.join(os.path.expanduser("~"), ".config", "ai_model_manager")
            os.makedirs(config_dir, mode=0o700, exist_ok=True)
            file_path = os.path.join(config_dir, "models_config.json")

        self.file_path = file_path

        if not os.path.exists(self.file_path):
            self._create_default_config()

    def _create_default_config(self) -> None:
        """Create an empty configuration file."""
        self._write_config({"selected_model": None, "models": {}})
        print(f"Configuration file created at: {self.file_path}")

    def _read_config(self) -> Dict[str, Any]:
        """Read configuration from file."""
        try:
            with open(self.file_path, "r") as f:
                return json.load(f)
        except (FileNotFoundError, json.JSONDecodeError):
            return {"selected_model": None, "models": {}}

    def _write_config(self, data: Dict[str, Any]) -> None:
        """Write configuration to file."""
        with open(self.file_path, "w") as f:
            json.dump(data, f, indent=4)
        os.chmod(self.file_path, 0o600)  # the file holds API keys

    def get_model_config(self, model_name: str) -> Optional[Dict[str, Any]]:
        """Get configuration for a specific model."""
        config = self._read_config()
        return config.get("models", {}).get(model_name)

    def set_model_config(
        self,
        model_name: str,
        provider: str,
        api_key: str,
        temperature: float = 0.7,
        max_tokens: Optional[int] = None,
    ) -> None:
        """Set or update configuration for a model."""
        config = self._read_config()

        if "models" not in config:
            config["models"] = {}

        config["models"][model_name] = {
            "provider": provider,
            "api_key": api_key,
            "temperature": temperature,
            "max_tokens": max_tokens,
        }

        self._write_config(config)
        print(f"Model '{model_name}' configured successfully.")

    def remove_model(self, model_name: str) -> None:
        """Remove a model from configuration."""
        config = self._read_config()

        if model_name not in config.get("models", {}):
            raise ValueError(f"Model '{model_name}' not found in configuration.")

        del config["models"][model_name]

        # Clear selected model if it was the removed one
        if config.get("selected_model") == model_name:
            config["selected_model"] = None

        self._write_config(config)
        print(f"Model '{model_name}' removed successfully.")

    def get_selected_model(self) -> Optional[str]:
        """Get the currently selected model name."""
        config = self._read_config()
        return config.get("selected_model")

    def set_selected_model(self, model_name: str) -> None:
        """Set the active model."""
        config = self._read_config()
        model_config = config.get("models", {}).get(model_name)

        if not model_config:
            raise ValueError(f"Model '{model_name}' is not configured.")

        if not model_config.get("api_key"):
            raise ValueError(f"Model '{model_name}' has no API key configured.")

        config["selected_model"] = model_name
        self._write_config(config)
        print(f"Selected model: {model_name}")

    def list_models(self) -> Dict[str, Dict[str, Any]]:
        """Get all configured models."""
        config = self._read_config()
        return config.get("models", {})

get_model_config(model_name)

Get configuration for a specific model.

Source code in cli/config_manager.py 
41
42
43
44
def get_model_config(self, model_name: str) -> Optional[Dict[str, Any]]:
    """Get configuration for a specific model."""
    config = self._read_config()
    return config.get("models", {}).get(model_name)

get_selected_model()

Get the currently selected model name.

Source code in cli/config_manager.py 
86
87
88
89
def get_selected_model(self) -> Optional[str]:
    """Get the currently selected model name."""
    config = self._read_config()
    return config.get("selected_model")

list_models()

Get all configured models.

Source code in cli/config_manager.py 
106
107
108
109
def list_models(self) -> Dict[str, Dict[str, Any]]:
    """Get all configured models."""
    config = self._read_config()
    return config.get("models", {})

remove_model(model_name)

Remove a model from configuration.

Source code in cli/config_manager.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def remove_model(self, model_name: str) -> None:
    """Remove a model from configuration."""
    config = self._read_config()

    if model_name not in config.get("models", {}):
        raise ValueError(f"Model '{model_name}' not found in configuration.")

    del config["models"][model_name]

    # Clear selected model if it was the removed one
    if config.get("selected_model") == model_name:
        config["selected_model"] = None

    self._write_config(config)
    print(f"Model '{model_name}' removed successfully.")

set_model_config(model_name, provider, api_key, temperature=0.7, max_tokens=None)

Set or update configuration for a model.

Source code in cli/config_manager.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def set_model_config(
    self,
    model_name: str,
    provider: str,
    api_key: str,
    temperature: float = 0.7,
    max_tokens: Optional[int] = None,
) -> None:
    """Set or update configuration for a model."""
    config = self._read_config()

    if "models" not in config:
        config["models"] = {}

    config["models"][model_name] = {
        "provider": provider,
        "api_key": api_key,
        "temperature": temperature,
        "max_tokens": max_tokens,
    }

    self._write_config(config)
    print(f"Model '{model_name}' configured successfully.")

set_selected_model(model_name)

Set the active model.

Source code in cli/config_manager.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def set_selected_model(self, model_name: str) -> None:
    """Set the active model."""
    config = self._read_config()
    model_config = config.get("models", {}).get(model_name)

    if not model_config:
        raise ValueError(f"Model '{model_name}' is not configured.")

    if not model_config.get("api_key"):
        raise ValueError(f"Model '{model_name}' has no API key configured.")

    config["selected_model"] = model_name
    self._write_config(config)
    print(f"Selected model: {model_name}")

llm

LLM interaction module.

generate_response(prompt, llm_service, history)

Generate and display AI response.

Source code in cli/llm.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def generate_response(prompt: str, llm_service: LLMService, history: ChatHistory):
    """Generate and display AI response."""
    system_instruction = (
        "System prompt: Give response in short and MD format, "
        "if asked for commands then give commands and don't explain too much"
    )

    full_prompt = f"{prompt}\n{system_instruction}"
    history.append("user", full_prompt)

    # Get conversation history as context
    flat_prompt = history.get_prompt()

    # Generate response using LLM service
    response = llm_service.generate(flat_prompt)

    # Save response to history
    history.append("assistant", response or "")

    # Display formatted output
    prettify_llm_output(response)

llm_service

Service for LLM generation and orchestration.

LLMService

Service for generating LLM responses.

Source code in cli/llm_service.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
class LLMService:
    """Service for generating LLM responses."""

    def __init__(self, config_manager: Optional[ConfigurationManager] = None):
        """
        Initialize LLM service.

        Args:
            config_manager: Configuration manager instance
        """
        self.config_manager = config_manager or ConfigurationManager()
        self._provider_cache: dict[str, ILLMProvider] = {}

    def _get_provider(self, model_name: str) -> ILLMProvider:
        """Get or create a provider instance for the model."""
        if model_name in self._provider_cache:
            return self._provider_cache[model_name]

        # Get model configuration
        model_config_dict = self.config_manager.get_model_config(model_name)
        if not model_config_dict:
            raise ValueError(f"Model '{model_name}' is not configured.")

        if not model_config_dict.get("api_key"):
            raise ValueError(f"Model '{model_name}' has no API key configured.")

        # Create ModelConfig object
        model_config = ModelConfig(
            name=model_name,
            provider=model_config_dict["provider"],
            api_key=model_config_dict["api_key"],
            temperature=model_config_dict.get("temperature", 0.7),
            max_tokens=model_config_dict.get("max_tokens"),
        )

        # Create provider using factory
        provider = LLMProviderFactory.create(model_config)

        # Cache the provider
        self._provider_cache[model_name] = provider

        return provider

    def generate(
        self,
        prompt: str,
        model_name: Optional[str] = None,
        show_spinner: bool = True,
        **kwargs,
    ) -> str:
        """Generate response from LLM."""
        if model_name is None:
            model_name = self.config_manager.get_selected_model()
            if not model_name:
                raise ValueError(
                    "No model selected. Run 'tex config <model> <api_key>', "
                    "then 'tex select <model>'."
                )

        # Get provider
        provider = self._get_provider(model_name)

        # Generate with optional spinner
        if show_spinner:
            stop_spinner = threading.Event()
            spinner_thread = threading.Thread(target=spin_loader, args=(stop_spinner,))
            spinner_thread.start()

            try:
                response = provider.generate(prompt, **kwargs)
            finally:
                stop_spinner.set()
                spinner_thread.join()
        else:
            response = provider.generate(prompt, **kwargs)

        return response

__init__(config_manager=None)

Initialize LLM service.

Parameters:

Name Type Description Default
config_manager Optional[ConfigurationManager]

Configuration manager instance

 None
Source code in cli/llm_service.py 
14
15
16
17
18
19
20
21
22
def __init__(self, config_manager: Optional[ConfigurationManager] = None):
    """
    Initialize LLM service.

    Args:
        config_manager: Configuration manager instance
    """
    self.config_manager = config_manager or ConfigurationManager()
    self._provider_cache: dict[str, ILLMProvider] = {}

generate(prompt, model_name=None, show_spinner=True, **kwargs)

Generate response from LLM.

Source code in cli/llm_service.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def generate(
    self,
    prompt: str,
    model_name: Optional[str] = None,
    show_spinner: bool = True,
    **kwargs,
) -> str:
    """Generate response from LLM."""
    if model_name is None:
        model_name = self.config_manager.get_selected_model()
        if not model_name:
            raise ValueError(
                "No model selected. Run 'tex config <model> <api_key>', "
                "then 'tex select <model>'."
            )

    # Get provider
    provider = self._get_provider(model_name)

    # Generate with optional spinner
    if show_spinner:
        stop_spinner = threading.Event()
        spinner_thread = threading.Thread(target=spin_loader, args=(stop_spinner,))
        spinner_thread.start()

        try:
            response = provider.generate(prompt, **kwargs)
        finally:
            stop_spinner.set()
            spinner_thread.join()
    else:
        response = provider.generate(prompt, **kwargs)

    return response

models

Models package for LLM abstraction layer. Provides interfaces and implementations for different LLM providers.

ILLMProvider

Bases: ABC

Interface for LLM providers.

Source code in cli/models/base.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class ILLMProvider(ABC):
    """Interface for LLM providers."""

    @abstractmethod
    def generate(self, prompt: str, **kwargs) -> str:
        """Generate a response from the LLM."""
        pass

    @abstractmethod
    def get_provider_name(self) -> str:
        """Get the name of the provider."""
        pass

    @abstractmethod
    def validate_config(self) -> bool:
        """Validate the model configuration."""
        pass

generate(prompt, **kwargs) abstractmethod

Generate a response from the LLM.

Source code in cli/models/base.py 
22
23
24
25
@abstractmethod
def generate(self, prompt: str, **kwargs) -> str:
    """Generate a response from the LLM."""
    pass

get_provider_name() abstractmethod

Get the name of the provider.

Source code in cli/models/base.py 
27
28
29
30
@abstractmethod
def get_provider_name(self) -> str:
    """Get the name of the provider."""
    pass

validate_config() abstractmethod

Validate the model configuration.

Source code in cli/models/base.py 
32
33
34
35
@abstractmethod
def validate_config(self) -> bool:
    """Validate the model configuration."""
    pass

LLMProviderFactory

Factory for creating LLM provider instances.

Source code in cli/models/factory.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
class LLMProviderFactory:
    """Factory for creating LLM provider instances."""

    _providers: Dict[str, Type[ILLMProvider]] = {
        "google": GeminiProvider,
        "gemini": GeminiProvider,
        "openai": OpenAIProvider,
        "anthropic": AnthropicProvider,
        "claude": AnthropicProvider,
    }

    _model_patterns = {
        "gemini": "google",
        "gpt": "openai",
        "claude": "anthropic",
    }

    @classmethod
    def create(cls, config: ModelConfig) -> ILLMProvider:
        """Create an LLM provider instance based on configuration."""
        provider_key = config.provider.lower()

        if provider_key in cls._providers:
            return cls._providers[provider_key](config)

        inferred_provider = cls._infer_provider_from_model(config.name)
        if inferred_provider:
            return cls._providers[inferred_provider](config)

        raise ValueError(
            f"Unsupported provider: {config.provider}. "
            f"Available: {', '.join(cls.get_supported_providers())}"
        )

    @classmethod
    def _infer_provider_from_model(cls, model_name: str) -> str | None:
        """Infer provider from model name."""
        model_lower = model_name.lower()
        for pattern, provider in cls._model_patterns.items():
            if pattern in model_lower:
                return provider
        return None

    @classmethod
    def get_supported_providers(cls) -> list[str]:
        """Get list of supported provider names."""
        return list(set(cls._providers.keys()))

create(config) classmethod

Create an LLM provider instance based on configuration.

Source code in cli/models/factory.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@classmethod
def create(cls, config: ModelConfig) -> ILLMProvider:
    """Create an LLM provider instance based on configuration."""
    provider_key = config.provider.lower()

    if provider_key in cls._providers:
        return cls._providers[provider_key](config)

    inferred_provider = cls._infer_provider_from_model(config.name)
    if inferred_provider:
        return cls._providers[inferred_provider](config)

    raise ValueError(
        f"Unsupported provider: {config.provider}. "
        f"Available: {', '.join(cls.get_supported_providers())}"
    )

get_supported_providers() classmethod

Get list of supported provider names.

Source code in cli/models/factory.py 
53
54
55
56
@classmethod
def get_supported_providers(cls) -> list[str]:
    """Get list of supported provider names."""
    return list(set(cls._providers.keys()))

ModelConfig dataclass

Configuration for an LLM model.

Source code in cli/models/base.py 
 8
 9
10
11
12
13
14
15
16
@dataclass
class ModelConfig:
    """Configuration for an LLM model."""

    name: str
    provider: str
    api_key: str
    temperature: float = 0.7
    max_tokens: Optional[int] = None

anthropic_provider

Anthropic Claude provider implementation using LangChain.

AnthropicProvider

Bases: BaseLLMProvider

Anthropic Claude LLM provider.

Source code in cli/models/anthropic_provider.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class AnthropicProvider(BaseLLMProvider):
    """Anthropic Claude LLM provider."""

    def _create_llm(self) -> BaseChatModel:
        ChatAnthropic = load_backend("langchain_anthropic", "ChatAnthropic", "anthropic")
        kwargs = {
            "model": self.config.name,
            "anthropic_api_key": self.config.api_key,
            "temperature": self.config.temperature,
        }
        if self.config.max_tokens:
            kwargs["max_tokens"] = self.config.max_tokens
        return ChatAnthropic(**kwargs)

    def get_provider_name(self) -> str:
        return "Anthropic Claude"

base

Base interfaces and abstractions for LLM providers.

ILLMProvider

Bases: ABC

Interface for LLM providers.

Source code in cli/models/base.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class ILLMProvider(ABC):
    """Interface for LLM providers."""

    @abstractmethod
    def generate(self, prompt: str, **kwargs) -> str:
        """Generate a response from the LLM."""
        pass

    @abstractmethod
    def get_provider_name(self) -> str:
        """Get the name of the provider."""
        pass

    @abstractmethod
    def validate_config(self) -> bool:
        """Validate the model configuration."""
        pass
generate(prompt, **kwargs) abstractmethod

Generate a response from the LLM.

Source code in cli/models/base.py 
22
23
24
25
@abstractmethod
def generate(self, prompt: str, **kwargs) -> str:
    """Generate a response from the LLM."""
    pass
get_provider_name() abstractmethod

Get the name of the provider.

Source code in cli/models/base.py 
27
28
29
30
@abstractmethod
def get_provider_name(self) -> str:
    """Get the name of the provider."""
    pass
validate_config() abstractmethod

Validate the model configuration.

Source code in cli/models/base.py 
32
33
34
35
@abstractmethod
def validate_config(self) -> bool:
    """Validate the model configuration."""
    pass

ModelConfig dataclass

Configuration for an LLM model.

Source code in cli/models/base.py 
 8
 9
10
11
12
13
14
15
16
@dataclass
class ModelConfig:
    """Configuration for an LLM model."""

    name: str
    provider: str
    api_key: str
    temperature: float = 0.7
    max_tokens: Optional[int] = None

base_provider

Base provider implementation with common functionality.

BaseLLMProvider

Bases: ILLMProvider

Base implementation for LLM providers.

Source code in cli/models/base_provider.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class BaseLLMProvider(ILLMProvider):
    """Base implementation for LLM providers."""

    def __init__(self, config: ModelConfig):
        self.config = config
        self.validate_config()
        self.llm: BaseChatModel = self._create_llm()

    def _create_llm(self) -> BaseChatModel:
        """Create LLM instance. Must be implemented by subclasses."""
        raise NotImplementedError("Subclasses must implement _create_llm")

    def generate(self, prompt: str, **kwargs) -> str:
        """Generate response from LLM."""
        if "temperature" in kwargs:
            self.llm.temperature = kwargs["temperature"]
        # LangChain 1.x returns content as a list of blocks; .text flattens it to a string.
        return self.llm.invoke(prompt).text

    def validate_config(self) -> bool:
        """Validate configuration."""
        if not self.config.api_key:
            raise ValueError(f"API key is required for {self.get_provider_name()}")
        if not self.config.name:
            raise ValueError("Model name is required")
        return True
generate(prompt, **kwargs)

Generate response from LLM.

Source code in cli/models/base_provider.py 
37
38
39
40
41
42
def generate(self, prompt: str, **kwargs) -> str:
    """Generate response from LLM."""
    if "temperature" in kwargs:
        self.llm.temperature = kwargs["temperature"]
    # LangChain 1.x returns content as a list of blocks; .text flattens it to a string.
    return self.llm.invoke(prompt).text
validate_config()

Validate configuration.

Source code in cli/models/base_provider.py 
44
45
46
47
48
49
50
def validate_config(self) -> bool:
    """Validate configuration."""
    if not self.config.api_key:
        raise ValueError(f"API key is required for {self.get_provider_name()}")
    if not self.config.name:
        raise ValueError("Model name is required")
    return True

load_backend(module, name, extra)

Import an optional provider backend, or say how to install it.

Source code in cli/models/base_provider.py 
14
15
16
17
18
19
20
21
22
def load_backend(module: str, name: str, extra: str):
    """Import an optional provider backend, or say how to install it."""
    try:
        return getattr(import_module(module), name)
    except ImportError:
        raise ImportError(
            f"Support for '{extra}' models is not installed. "
            f"Run: pip install 'Vertex-CLI[{extra}]'"
        ) from None

factory

Factory for creating LLM provider instances.

LLMProviderFactory

Factory for creating LLM provider instances.

Source code in cli/models/factory.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
class LLMProviderFactory:
    """Factory for creating LLM provider instances."""

    _providers: Dict[str, Type[ILLMProvider]] = {
        "google": GeminiProvider,
        "gemini": GeminiProvider,
        "openai": OpenAIProvider,
        "anthropic": AnthropicProvider,
        "claude": AnthropicProvider,
    }

    _model_patterns = {
        "gemini": "google",
        "gpt": "openai",
        "claude": "anthropic",
    }

    @classmethod
    def create(cls, config: ModelConfig) -> ILLMProvider:
        """Create an LLM provider instance based on configuration."""
        provider_key = config.provider.lower()

        if provider_key in cls._providers:
            return cls._providers[provider_key](config)

        inferred_provider = cls._infer_provider_from_model(config.name)
        if inferred_provider:
            return cls._providers[inferred_provider](config)

        raise ValueError(
            f"Unsupported provider: {config.provider}. "
            f"Available: {', '.join(cls.get_supported_providers())}"
        )

    @classmethod
    def _infer_provider_from_model(cls, model_name: str) -> str | None:
        """Infer provider from model name."""
        model_lower = model_name.lower()
        for pattern, provider in cls._model_patterns.items():
            if pattern in model_lower:
                return provider
        return None

    @classmethod
    def get_supported_providers(cls) -> list[str]:
        """Get list of supported provider names."""
        return list(set(cls._providers.keys()))
create(config) classmethod

Create an LLM provider instance based on configuration.

Source code in cli/models/factory.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@classmethod
def create(cls, config: ModelConfig) -> ILLMProvider:
    """Create an LLM provider instance based on configuration."""
    provider_key = config.provider.lower()

    if provider_key in cls._providers:
        return cls._providers[provider_key](config)

    inferred_provider = cls._infer_provider_from_model(config.name)
    if inferred_provider:
        return cls._providers[inferred_provider](config)

    raise ValueError(
        f"Unsupported provider: {config.provider}. "
        f"Available: {', '.join(cls.get_supported_providers())}"
    )
get_supported_providers() classmethod

Get list of supported provider names.

Source code in cli/models/factory.py 
53
54
55
56
@classmethod
def get_supported_providers(cls) -> list[str]:
    """Get list of supported provider names."""
    return list(set(cls._providers.keys()))

gemini_provider

Google Gemini provider implementation using LangChain.

GeminiProvider

Bases: BaseLLMProvider

Google Gemini LLM provider.

Source code in cli/models/gemini_provider.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
class GeminiProvider(BaseLLMProvider):
    """Google Gemini LLM provider."""

    def _create_llm(self) -> BaseChatModel:
        # Deferred: importing this package costs ~780ms, so commands that never
        # talk to a model (list, config, select, --help) must not pay for it.
        from langchain_google_genai import ChatGoogleGenerativeAI

        kwargs = {
            "model": self.config.name,
            "google_api_key": self.config.api_key,
            "temperature": self.config.temperature,
        }
        if self.config.max_tokens:
            kwargs["max_output_tokens"] = self.config.max_tokens
        return ChatGoogleGenerativeAI(**kwargs)

    def get_provider_name(self) -> str:
        return "Google Gemini"

openai_provider

OpenAI provider implementation using LangChain.

OpenAIProvider

Bases: BaseLLMProvider

OpenAI LLM provider.

Source code in cli/models/openai_provider.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class OpenAIProvider(BaseLLMProvider):
    """OpenAI LLM provider."""

    def _create_llm(self) -> BaseChatModel:
        ChatOpenAI = load_backend("langchain_openai", "ChatOpenAI", "openai")
        kwargs = {
            "model": self.config.name,
            "openai_api_key": self.config.api_key,
            "temperature": self.config.temperature,
        }
        if self.config.max_tokens:
            kwargs["max_tokens"] = self.config.max_tokens
        return ChatOpenAI(**kwargs)

    def get_provider_name(self) -> str:
        return "OpenAI"

prompt

CLI entry point for Vertex-CLI.

main()

Report failures as messages instead of tracebacks.

Source code in cli/prompt.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
def main():
    """Report failures as messages instead of tracebacks."""
    # Provider libraries warn about their own internals (ignored sampling params,
    # automatic function calling). None of it is actionable for a CLI user.
    warnings.filterwarnings("ignore")
    logging.getLogger("google_genai").setLevel(logging.ERROR)
    try:
        _run()
    except KeyboardInterrupt:
        sys.exit(130)
    except Exception as exc:
        print(f"Error: {exc}", file=sys.stderr)
        sys.exit(1)

utils

install_requirements()

Install the dependencies declared in pyproject.toml.

Source code in cli/utils.py 
20
21
22
23
24
def install_requirements():
    """Install the dependencies declared in pyproject.toml."""
    for package in requires("Vertex-CLI") or []:
        if "extra ==" not in package:  # skip optional [dev] extras
            subprocess.run([sys.executable, "-m", "pip", "install", package])

prettify_llm_output(response)

Prettify LLM output using Rich markdown.

Source code in cli/utils.py 
27
28
29
30
31
32
33
34
def prettify_llm_output(response):
    """Prettify LLM output using Rich markdown."""
    from rich.console import Console  # deferred, only needed to render a reply
    from rich.markdown import Markdown

    console = Console()
    md = Markdown(response.strip())
    console.print("\n", md, "\n")

spin_loader(stop_event)

Display a spinning loader.

Source code in cli/utils.py 
 8
 9
10
11
12
13
14
15
16
17
def spin_loader(stop_event):
    """Display a spinning loader."""
    spinner = itertools.cycle(["-", "/", "|", "\\"])
    while not stop_event.is_set():
        sys.stdout.write(next(spinner))
        sys.stdout.flush()
        time.sleep(0.1)
        sys.stdout.write("\b")
    sys.stdout.write(" ")
    sys.stdout.flush()