Skip to content

manager

Class info

Classes

Name Children Inherits
BuiltinPromptProvider
llmling_agent.prompts.builtin_provider
Default provider using system prompts.
    LangfuseConfig
    llmling_agent_config.prompt_hubs
    Configuration for Langfuse prompt provider.
      • BasePromptHubConfig
      OpenLITConfig
      llmling_agent_config.prompt_hubs
      Configuration for OpenLIT prompt provider.
        • BasePromptHubConfig
        PromptLayerConfig
        llmling_agent_config.prompt_hubs
        Configuration for PromptLayer prompt provider.
          • BasePromptHubConfig
          PromptManager
          llmling_agent.prompts.manager
          Manages multiple prompt providers.
            TaskManager
            llmling_agent.utils.tasks
            Mixin for managing async tasks.
              TraceloopConfig
              llmling_agent_config.prompt_hubs
              Configuration for Traceloop prompt provider.
                • BasePromptHubConfig

                🛈 DocStrings

                Prompt Manager.

                PromptManager

                Manages multiple prompt providers.

                Handles: - Provider initialization and cleanup - Prompt reference parsing and resolution - Access to prompts from different sources

                Source code in src/llmling_agent/prompts/manager.py 
                 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
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
                class PromptManager:
    """Manages multiple prompt providers.

    Handles:
    - Provider initialization and cleanup
    - Prompt reference parsing and resolution
    - Access to prompts from different sources
    """

    def __init__(self, config: PromptLibraryConfig):
        """Initialize prompt manager.

        Args:
            config: Prompt configuration including providers
        """
        super().__init__()
        self.config = config
        self.task_manager = TaskManager()
        self.providers: dict[str, BasePromptProvider] = {}

        # Always register builtin provider
        self.providers["builtin"] = BuiltinPromptProvider(config.system_prompts)

        # Register configured providers
        for provider_config in config.providers:
            match provider_config:
                case LangfuseConfig():
                    from llmling_agent_prompts.langfuse_hub import LangfusePromptHub

                    self.providers["langfuse"] = LangfusePromptHub(provider_config)
                case OpenLITConfig():
                    from llmling_agent_prompts.openlit_provider import OpenLITProvider

                    self.providers["openlit"] = OpenLITProvider(provider_config)
                case PromptLayerConfig():
                    from llmling_agent_prompts.promptlayer import PromptLayerProvider

                    self.providers["promptlayer"] = PromptLayerProvider(provider_config)
                case TraceloopConfig():
                    from llmling_agent_prompts.traceloop_provider import (
                        TraceloopPromptHub,
                    )

                    self.providers["traceloop"] = TraceloopPromptHub(provider_config)

    async def get_from(
        self,
        identifier: str,
        *,
        provider: str | None = None,
        version: str | None = None,
        variables: dict[str, Any] | None = None,
    ) -> str:
        """Get a prompt.

        Args:
            identifier: Prompt identifier/name
            provider: Provider name (None = builtin)
            version: Optional version string
            variables: Optional template variables

        Examples:
            await prompts.get_from("code_review", variables={"language": "python"})
            await prompts.get_from(
                "expert",
                provider="langfuse",
                version="v2",
                variables={"domain": "ML"}
            )
        """
        provider_name = provider or "builtin"
        if provider_name not in self.providers:
            msg = f"Unknown prompt provider: {provider_name}"
            raise KeyError(msg)

        provider_instance = self.providers[provider_name]
        try:
            kwargs: dict[str, Any] = {}
            if provider_instance.supports_versions and version:
                kwargs["version"] = version
            if provider_instance.supports_variables and variables:
                kwargs["variables"] = variables

            return await provider_instance.get_prompt(identifier, **kwargs)
        except Exception as e:
            msg = f"Failed to get prompt {identifier!r} from {provider_name}"
            raise RuntimeError(msg) from e

    async def get(self, reference: str) -> str:
        """Get a prompt using identifier syntax.

        Args:
            reference: Prompt identifier in format:
                      [provider:]name[@version][?var1=val1,var2=val2]

        Examples:
            await prompts.get("error_handler")  # Builtin prompt
            await prompts.get("langfuse:code_review@v2?style=detailed")
            await prompts.get("openlit:explain?context=web,detail=high")
        """
        provider_name, id_, version, vars_ = parse_prompt_reference(reference)
        prov = provider_name if provider_name != "builtin" else None
        return await self.get_from(id_, provider=prov, version=version, variables=vars_)

    def get_sync(self, reference: str) -> str:
        """Synchronous wrapper for get().

        Note: This is a temporary solution for initialization contexts.
        Should be used only when async operation is not possible.
        """
        try:
            loop = asyncio.get_running_loop()
            return loop.run_until_complete(self.get(reference))
        except RuntimeError:
            # No running loop - create new one
            return asyncio.run(self.get(reference))

    async def list_prompts(self, provider: str | None = None) -> dict[str, list[str]]:
        """List available prompts.

        Args:
            provider: Optional provider name to filter by

        Returns:
            Dict mapping provider names to their available prompts
        """
        providers = {provider: self.providers[provider]} if provider else self.providers

        if not providers:
            return {}
        # Get prompts from providers concurrently
        result = {}
        coros = [p.list_prompts() for p in providers.values()]
        gathered_results = await asyncio.gather(*coros, return_exceptions=True)
        for (name, _), prompts in zip(providers.items(), gathered_results, strict=False):
            if isinstance(prompts, BaseException):
                logger.exception("Failed to list prompts from %s", name)
                continue
            result[name] = prompts
        return result

    def cleanup(self):
        """Clean up providers."""
        self.providers.clear()

    def get_default_provider(self) -> str:
        """Get default provider name.

        Returns configured default or "builtin"
        """
        return "builtin"

                __init__ 

                __init__(config: PromptLibraryConfig)

                Initialize prompt manager.

                Parameters:

                Name Type Description Default
                config PromptLibraryConfig

                Prompt configuration including providers

                		 required
                Source code in src/llmling_agent/prompts/manager.py 
                 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
                def __init__(self, config: PromptLibraryConfig):
    """Initialize prompt manager.

    Args:
        config: Prompt configuration including providers
    """
    super().__init__()
    self.config = config
    self.task_manager = TaskManager()
    self.providers: dict[str, BasePromptProvider] = {}

    # Always register builtin provider
    self.providers["builtin"] = BuiltinPromptProvider(config.system_prompts)

    # Register configured providers
    for provider_config in config.providers:
        match provider_config:
            case LangfuseConfig():
                from llmling_agent_prompts.langfuse_hub import LangfusePromptHub

                self.providers["langfuse"] = LangfusePromptHub(provider_config)
            case OpenLITConfig():
                from llmling_agent_prompts.openlit_provider import OpenLITProvider

                self.providers["openlit"] = OpenLITProvider(provider_config)
            case PromptLayerConfig():
                from llmling_agent_prompts.promptlayer import PromptLayerProvider

                self.providers["promptlayer"] = PromptLayerProvider(provider_config)
            case TraceloopConfig():
                from llmling_agent_prompts.traceloop_provider import (
                    TraceloopPromptHub,
                )

                self.providers["traceloop"] = TraceloopPromptHub(provider_config)

                cleanup 

                cleanup()

                Clean up providers.

                Source code in src/llmling_agent/prompts/manager.py 
                203
204
205
                def cleanup(self):
    """Clean up providers."""
    self.providers.clear()

                get async 

                get(reference: str) -> str

                Get a prompt using identifier syntax.

                Parameters:

                Name Type Description Default
                reference str

                Prompt identifier in format: [provider:]name[@version][?var1=val1,var2=val2]

                		 required

                Examples:

                await prompts.get("error_handler") # Builtin prompt await prompts.get("langfuse:code_review@v2?style=detailed") await prompts.get("openlit:explain?context=web,detail=high")

                Source code in src/llmling_agent/prompts/manager.py 
                150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
                async def get(self, reference: str) -> str:
    """Get a prompt using identifier syntax.

    Args:
        reference: Prompt identifier in format:
                  [provider:]name[@version][?var1=val1,var2=val2]

    Examples:
        await prompts.get("error_handler")  # Builtin prompt
        await prompts.get("langfuse:code_review@v2?style=detailed")
        await prompts.get("openlit:explain?context=web,detail=high")
    """
    provider_name, id_, version, vars_ = parse_prompt_reference(reference)
    prov = provider_name if provider_name != "builtin" else None
    return await self.get_from(id_, provider=prov, version=version, variables=vars_)

                get_default_provider 

                get_default_provider() -> str

                Get default provider name.

                Returns configured default or "builtin"

                Source code in src/llmling_agent/prompts/manager.py 
                207
208
209
210
211
212
                def get_default_provider(self) -> str:
    """Get default provider name.

    Returns configured default or "builtin"
    """
    return "builtin"

                get_from async 

                get_from(
    identifier: str,
    *,
    provider: str | None = None,
    version: str | None = None,
    variables: dict[str, Any] | None = None,
) -> str

                Get a prompt.

                Parameters:

                Name Type Description Default
                identifier str

                Prompt identifier/name

                		 required
                provider str | None

                Provider name (None = builtin)

                		 None
                version str | None

                Optional version string

                		 None
                variables dict[str, Any] | None

                Optional template variables

                		 None

                Examples:

                await prompts.get_from("code_review", variables={"language": "python"}) await prompts.get_from( "expert", provider="langfuse", version="v2", variables={"domain": "ML"} )

                Source code in src/llmling_agent/prompts/manager.py 
                107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
                async def get_from(
    self,
    identifier: str,
    *,
    provider: str | None = None,
    version: str | None = None,
    variables: dict[str, Any] | None = None,
) -> str:
    """Get a prompt.

    Args:
        identifier: Prompt identifier/name
        provider: Provider name (None = builtin)
        version: Optional version string
        variables: Optional template variables

    Examples:
        await prompts.get_from("code_review", variables={"language": "python"})
        await prompts.get_from(
            "expert",
            provider="langfuse",
            version="v2",
            variables={"domain": "ML"}
        )
    """
    provider_name = provider or "builtin"
    if provider_name not in self.providers:
        msg = f"Unknown prompt provider: {provider_name}"
        raise KeyError(msg)

    provider_instance = self.providers[provider_name]
    try:
        kwargs: dict[str, Any] = {}
        if provider_instance.supports_versions and version:
            kwargs["version"] = version
        if provider_instance.supports_variables and variables:
            kwargs["variables"] = variables

        return await provider_instance.get_prompt(identifier, **kwargs)
    except Exception as e:
        msg = f"Failed to get prompt {identifier!r} from {provider_name}"
        raise RuntimeError(msg) from e

                get_sync 

                get_sync(reference: str) -> str

                Synchronous wrapper for get().

                Note: This is a temporary solution for initialization contexts. Should be used only when async operation is not possible.

                Source code in src/llmling_agent/prompts/manager.py 
                166
167
168
169
170
171
172
173
174
175
176
177
                def get_sync(self, reference: str) -> str:
    """Synchronous wrapper for get().

    Note: This is a temporary solution for initialization contexts.
    Should be used only when async operation is not possible.
    """
    try:
        loop = asyncio.get_running_loop()
        return loop.run_until_complete(self.get(reference))
    except RuntimeError:
        # No running loop - create new one
        return asyncio.run(self.get(reference))

                list_prompts async 

                list_prompts(provider: str | None = None) -> dict[str, list[str]]

                List available prompts.

                Parameters:

                Name Type Description Default
                provider str | None

                Optional provider name to filter by

                		 None

                Returns:

                Type Description
                dict[str, list[str]]

                Dict mapping provider names to their available prompts
                Source code in src/llmling_agent/prompts/manager.py 
                179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
                async def list_prompts(self, provider: str | None = None) -> dict[str, list[str]]:
    """List available prompts.

    Args:
        provider: Optional provider name to filter by

    Returns:
        Dict mapping provider names to their available prompts
    """
    providers = {provider: self.providers[provider]} if provider else self.providers

    if not providers:
        return {}
    # Get prompts from providers concurrently
    result = {}
    coros = [p.list_prompts() for p in providers.values()]
    gathered_results = await asyncio.gather(*coros, return_exceptions=True)
    for (name, _), prompts in zip(providers.items(), gathered_results, strict=False):
        if isinstance(prompts, BaseException):
            logger.exception("Failed to list prompts from %s", name)
            continue
        result[name] = prompts
    return result

                parse_prompt_reference 

                parse_prompt_reference(reference: str) -> tuple[str, str, str | None, dict[str, str]]

                Parse a prompt reference string.

                Parameters:

                Name Type Description Default
                reference str

                Format [provider:]name[@version][?var1=val1,var2=val2]

                		 required

                Returns:

                Type Description
                str

                Tuple of (provider, identifier, version, variables)
                str

                Provider defaults to "builtin" if not specified
                Source code in src/llmling_agent/prompts/manager.py 
                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
                def parse_prompt_reference(reference: str) -> tuple[str, str, str | None, dict[str, str]]:
    """Parse a prompt reference string.

    Args:
        reference: Format [provider:]name[@version][?var1=val1,var2=val2]

    Returns:
        Tuple of (provider, identifier, version, variables)
        Provider defaults to "builtin" if not specified
    """
    provider = "builtin"
    version = None
    variables: dict[str, str] = {}

    # Split provider and rest
    if ":" in reference:
        provider, identifier = reference.split(":", 1)
    else:
        identifier = reference

    # Handle version
    if "@" in identifier:
        identifier, version = identifier.split("@", 1)

    # Handle query parameters
    if "?" in identifier:
        identifier, query = identifier.split("?", 1)
        for pair in query.split(","):
            if "=" not in pair:
                continue
            key, value = pair.split("=", 1)
            variables[key.strip()] = value.strip()

    return provider, identifier.strip(), version, variables