Skip to content

manager

Class info

Classes

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

              🛈 DocStrings

              PromptManager

              Bases: TaskManagerMixin

              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 
               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
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
              class PromptManager(TaskManagerMixin):
    """Manages multiple prompt providers.

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

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

        Args:
            config: Prompt configuration including providers
        """
        super().__init__()
        self.config = config
        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 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
        """
        result = {}
        providers = {provider: self.providers[provider]} if provider else self.providers

        for name, p in providers.items():
            try:
                prompts = await p.list_prompts()
                result[name] = prompts
            except Exception:
                logger.exception("Failed to list prompts from %s", name)
                continue

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

              Initialize prompt manager.

              Parameters:

              Name Type Description Default
              config PromptConfig

              Prompt configuration including providers

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

    Args:
        config: Prompt configuration including providers
    """
    super().__init__()
    self.config = config
    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 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 
              198
199
200
              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 
              147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
              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 
              202
203
204
205
206
207
              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 
              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
              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 
              163
164
165
166
167
168
169
170
171
172
173
174
              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 
              176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
              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
    """
    result = {}
    providers = {provider: self.providers[provider]} if provider else self.providers

    for name, p in providers.items():
        try:
            prompts = await p.list_prompts()
            result[name] = prompts
        except Exception:
            logger.exception("Failed to list prompts from %s", name)
            continue

    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 
              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
              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 = {}

    # 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