Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -360,6 +360,21 @@ prompt = Netra.prompts.get_prompt("my-prompt", use_cache=True, cache_ttl=300)
Netra.prompts.clear_cache()
```

### Linking prompts to traces

Wrap your LLM calls in `use_prompt` to tag every span started inside the block with the prompt's identity (`netra.prompt.id`, `netra.prompt.name`, `netra.prompt.version`, `netra.prompt.label`):

```python
prompt = Netra.prompts.get_prompt("my-prompt", label="production")

with Netra.prompts.use_prompt(prompt):
client.chat.completions.create(model="gpt-4o", messages=prompt["messages"])
```

- Pass the value returned by `get_prompt` as-is. A failed fetch (empty result) links nothing and does not raise.
- Spans in `threading.Thread`, `Timer` and `ThreadPoolExecutor` workers are linked while Netra's threading instrumentation is enabled (the default). With it disabled, only `asyncio` tasks inherit the prompt.
- Requires a Netra backend that returns prompt identity from `get_prompt`; older backends link nothing.

Caching notes:

- `use_cache` defaults to `False`; enable it per call when you want caching.
Expand Down
3 changes: 3 additions & 0 deletions netra/processors/session_span_processor.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@

from netra.config import Config
from netra.netra_logger import get_logger
from netra.prompts.context import get_active_prompt_attributes
from netra.session_manager import ATTR_SESSION_ID, ATTR_TENANT_ID, ATTR_USER_ID, SessionManager

logger = get_logger(__name__)
Expand Down Expand Up @@ -78,6 +79,8 @@ def on_start(self, span: trace.Span, parent_context: Optional[otel_context.Conte
span.set_attribute("library.name", Config.LIBRARY_NAME)
span.set_attribute("library.version", Config.LIBRARY_VERSION)
span.set_attribute("sdk.name", Config.SDK_NAME)
for key, value in get_active_prompt_attributes().items():
span.set_attribute(key, value)

if session_id:
span.set_attribute(ATTR_SESSION_ID, session_id)
Expand Down
3 changes: 3 additions & 0 deletions netra/prompts/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
from netra.config import Config
from netra.netra_logger import NoticeLevel, get_logger
from netra.prompts.client import PromptsHttpClient
from netra.prompts.context import use_prompt

logger = get_logger(__name__)

Expand All @@ -26,6 +27,8 @@ def __init__(self, cfg: Config) -> None:
self._client = PromptsHttpClient(cfg)
self._cache: TTLCache[Any] = TTLCache(default_ttl=PROMPT_CACHE_TTL_SECONDS)

use_prompt = staticmethod(use_prompt)

def clear_cache(self) -> None:
"""Clear all cached prompt entries."""
self._cache.clear()
Expand Down
65 changes: 65 additions & 0 deletions netra/prompts/context.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,65 @@
import logging
from contextlib import contextmanager
from typing import Any, Dict, Iterator, Mapping, Optional

from opentelemetry import context as context_api

from netra.config import Config

logger = logging.getLogger(__name__)

ATTR_PROMPT_ID = f"{Config.LIBRARY_NAME}.prompt.id"
ATTR_PROMPT_NAME = f"{Config.LIBRARY_NAME}.prompt.name"
ATTR_PROMPT_VERSION = f"{Config.LIBRARY_NAME}.prompt.version"
ATTR_PROMPT_LABEL = f"{Config.LIBRARY_NAME}.prompt.label"

# Prompt identity tagged onto every span started inside `use_prompt`. Stored in the
# OTel context (not a bare ContextVar) so the threading instrumentation carries it
# into ThreadPoolExecutor workers.
_ACTIVE_PROMPT_KEY = "netra.active_prompt"


def get_active_prompt_attributes() -> Dict[str, Any]:
"""
Build span attributes for the prompt set by `use_prompt`.

Returns:
Span attributes, or an empty dict when no prompt is active or it carries no identity
"""
prompt = context_api.get_value(_ACTIVE_PROMPT_KEY)
if not isinstance(prompt, Mapping):
return {}

prompt_id = prompt.get("promptId")
version = prompt.get("versionNumber")
if not prompt_id or version is None:
logger.debug("Active prompt has no promptId/versionNumber; spans will not be linked: %s", list(prompt))
return {}
Comment thread
sarathms-kv marked this conversation as resolved.
Resolved

attributes: Dict[str, Any] = {ATTR_PROMPT_ID: prompt_id, ATTR_PROMPT_VERSION: version}
if prompt.get("promptName"):
attributes[ATTR_PROMPT_NAME] = prompt["promptName"]
if prompt.get("label"):
attributes[ATTR_PROMPT_LABEL] = prompt["label"]
return attributes


@contextmanager
def use_prompt(prompt: Optional[Mapping[str, Any]]) -> Iterator[None]:
"""
Link every span started inside the block to a prompt version.

The prompt lives in the OTel context, so it reaches worker threads only when
the threading instrumentation propagates that context (e.g. `ThreadPoolExecutor`).

Args:
prompt: The value returned by `Netra.prompts.get_prompt`. A missing or
empty value is accepted and links nothing.
"""
if not isinstance(prompt, Mapping):
logger.debug("use_prompt called without a prompt mapping (%s); nothing linked", type(prompt).__name__)
token = context_api.attach(context_api.set_value(_ACTIVE_PROMPT_KEY, prompt))
try:
yield
finally:
context_api.detach(token)
102 changes: 102 additions & 0 deletions tests/test_prompts_context.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,102 @@
import pytest
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter

from netra.processors.session_span_processor import SessionSpanProcessor
from netra.prompts.context import (
ATTR_PROMPT_ID,
ATTR_PROMPT_LABEL,
ATTR_PROMPT_NAME,
ATTR_PROMPT_VERSION,
get_active_prompt_attributes,
use_prompt,
)

PROMPT = {
"messages": [{"role": "user", "content": "hi"}],
"promptId": "p1",
"promptName": "welcome",
"versionNumber": 3,
"label": "production",
}


class TestUsePrompt:
def test_no_active_prompt_yields_no_attributes(self) -> None:
assert get_active_prompt_attributes() == {}

def test_active_prompt_yields_identity_attributes(self) -> None:
with use_prompt(PROMPT):
assert get_active_prompt_attributes() == {
ATTR_PROMPT_ID: "p1",
ATTR_PROMPT_NAME: "welcome",
ATTR_PROMPT_VERSION: 3,
ATTR_PROMPT_LABEL: "production",
}
assert get_active_prompt_attributes() == {}

@pytest.mark.parametrize("prompt", [None, {}, {"promptId": "p1"}, {"versionNumber": 1}, "not-a-mapping"])
def test_untrackable_prompt_links_nothing(self, prompt: object) -> None:
with use_prompt(prompt): # type: ignore[arg-type]
assert get_active_prompt_attributes() == {}

def test_version_zero_is_kept(self) -> None:
with use_prompt({"promptId": "p1", "versionNumber": 0}):
assert get_active_prompt_attributes()[ATTR_PROMPT_VERSION] == 0

def test_nested_blocks_restore_outer_prompt(self) -> None:
inner = {**PROMPT, "promptId": "p2"}
with use_prompt(PROMPT):
with use_prompt(inner):
assert get_active_prompt_attributes()[ATTR_PROMPT_ID] == "p2"
assert get_active_prompt_attributes()[ATTR_PROMPT_ID] == "p1"

def test_prompt_reset_when_block_raises(self) -> None:
with pytest.raises(RuntimeError):
with use_prompt(PROMPT):
raise RuntimeError("boom")
assert get_active_prompt_attributes() == {}


class TestSpanTagging:
def test_spans_inside_block_are_tagged_and_outside_are_not(self) -> None:
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SessionSpanProcessor())
provider.add_span_processor(SimpleSpanProcessor(exporter))
tracer = provider.get_tracer("test")

with use_prompt(PROMPT):
with tracer.start_as_current_span("inside"):
pass
with tracer.start_as_current_span("outside"):
pass

spans = {s.name: s.attributes for s in exporter.get_finished_spans()}
assert spans["inside"][ATTR_PROMPT_ID] == "p1"
assert spans["inside"][ATTR_PROMPT_VERSION] == 3
assert ATTR_PROMPT_ID not in spans["outside"]


def test_prompt_propagates_into_threadpool_with_threading_instrumentation() -> None:
from concurrent.futures import ThreadPoolExecutor

from opentelemetry.instrumentation.threading import ThreadingInstrumentor

ThreadingInstrumentor().instrument()
try:
with use_prompt(PROMPT), ThreadPoolExecutor(1) as pool:
attrs = pool.submit(get_active_prompt_attributes).result()
assert attrs[ATTR_PROMPT_ID] == "p1"

import threading

out: dict = {}
with use_prompt(PROMPT):
t = threading.Thread(target=lambda: out.update(get_active_prompt_attributes()))
t.start()
t.join()
assert out[ATTR_PROMPT_ID] == "p1"
finally:
ThreadingInstrumentor().uninstrument()
Loading