Custom agents — monitoring¶

#!/usr/bin/env python3
"""Publish one agent and send one trace, to prove the whole path works."""

import base64
import json
import os
import sys

from langchain_openai import ChatOpenAI
from langfuse import Langfuse
from langfuse.langchain import CallbackHandler
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from trust3_ai_collector_sdk import (
    Asset,
    AssetKind,
    InventoryClient,
    InventoryClientConfig,
    InventorySnapshot,
    Relationship,
    RelationshipKind,
)
from trust3_ai_collector_sdk.errors import (
    InventoryTransportError,
    InventoryValidationError,
)

AGENT_NAME = "Expense Report Auditor"   # the one string both halves must agree on
AGENT_ID = "expense_report_auditor"
MODEL = "gpt-4o-mini"


def env(name: str) -> str:
    """Read a required setting, or stop with a message naming what is missing."""
    value = os.environ.get(name, "").strip()
    if not value:
        sys.exit(f"FAILED: {name} is not set. See 'Export your settings' in the docs.")
    return value


snapshot = InventorySnapshot(
    source="finance_platform",
    account="prod.example.com",
    vendor="Acme Corp",
    product_name="Finance Platform",
    environment="prod",
)
snapshot.assets += [
    Asset(
        kind=AssetKind.AGENT,
        id=AGENT_ID,
        name=AGENT_NAME,
        description="Checks submitted expense reports against travel policy",
        owners=["finance-eng@example.com"],
        provider="openai",
        model_name=MODEL,
        properties={
            "provenance_json": json.dumps({"categories": ["expenses", "travel_policy"]}),
        },
    ),
    Asset(kind=AssetKind.MODEL, id=f"openai:{MODEL}", name=MODEL, provider="openai"),
]
snapshot.relationships += [
    Relationship(
        kind=RelationshipKind.USES_MODEL,
        from_id=AGENT_ID,
        from_kind=AssetKind.AGENT,
        to_id=f"openai:{MODEL}",
        to_kind=AssetKind.MODEL,
    ),
]

# ---- 1. discovery: publish the agent so the trace has something to link to --
config = InventoryClientConfig(api_key=env("TRUST3_API_KEY"))
print(f"[1/3] publishing inventory to {config.api_server_host} ...")
try:
    with InventoryClient(config) as client:
        result = client.submit(snapshot)
except InventoryValidationError as exc:
    sys.exit(f"FAILED: governance rejected the inventory (HTTP {exc.status_code}): {exc.response_text}")
except InventoryTransportError as exc:
    sys.exit(f"FAILED: could not reach {config.api_server_host} after retries: {exc}")

print(
    f"      ok: {result.assets_pushed} assets, "
    f"{result.relationships_pushed} relationships, "
    f"governance={result.governance_status}"
)

# ---- 2. monitoring: point tracing at the Trust3 endpoint --------------------
print("[2/3] configuring tracing ...")
endpoint = env("TRUST3_TRACE_ENDPOINT")
token = base64.b64encode(
    f'{env("TRUST3_TRACE_USERNAME")}:{env("TRUST3_TRACE_PASSWORD")}'.encode()
).decode()

provider = TracerProvider(resource=Resource.create({"service.name": "finance-platform"}))
provider.add_span_processor(
    BatchSpanProcessor(
        OTLPSpanExporter(endpoint=endpoint, headers={"Authorization": f"Basic {token}"})
    )
)

# Placeholder keys + should_export_span=False: Langfuse instruments but never
# exports, leaving the Trust3 collector as the only destination.
langfuse = Langfuse(
    public_key="pk-local",
    secret_key="sk-local",
    tracer_provider=provider,
    should_export_span=lambda span: False,
)
print(f"      ok: exporting to {endpoint}")

# ---- 3. one call, named so it links to the agent published above ------------
print(f'[3/3] calling {MODEL} as "{AGENT_NAME}" ...')
llm = ChatOpenAI(model=MODEL, temperature=0)

# The agent span is the root of the trace, and its name is what Trust3 links
# against. The model call goes inside it. Keep this shape — see the note below.
with langfuse.start_as_current_observation(as_type="span", name=AGENT_NAME):
    reply = llm.invoke(
        "Does a 45 dollar airport taxi need a receipt under our travel policy?",
        config={
            "callbacks": [CallbackHandler()],
            "run_name": AGENT_NAME,      # names the model call within the trace
            "metadata": {"user.id": "alice@example.com"},
        },
    )
print(f"      ok: {len(reply.content)} characters returned")

# Spans are batched — without this a short script exits having exported nothing.
print("      flushing spans ...")
provider.shutdown()

print()
print("[done] script finished without errors.")
print(f'       Open AI Assets in the portal and find "Expense Report Auditor".')
print("       Traces are picked up on the collector's scan interval, 15 minutes by default,")
print("       so activity can take that long to appear.")
print("       If the endpoint or credentials were wrong, the OpenTelemetry SDK logs an export")
print("       error above — the LLM call itself still succeeds, so watch for that.")

#!/usr/bin/env python3
"""Publish an agent and a sub-agent, then trace a real LangGraph run."""

import base64
import json
import os
import sys
from typing import Annotated, TypedDict

from langchain_openai import ChatOpenAI
from langfuse import Langfuse
from langfuse.langchain import CallbackHandler
from langgraph.graph import START, StateGraph
from langgraph.graph.message import add_messages
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from trust3_ai_collector_sdk import (
    Asset,
    AssetKind,
    InventoryClient,
    InventoryClientConfig,
    InventorySnapshot,
    Relationship,
    RelationshipKind,
)
from trust3_ai_collector_sdk.errors import (
    InventoryTransportError,
    InventoryValidationError,
)

AGENT_NAME = "Loan Application Reviewer"
AGENT_ID = "loan_application_reviewer"
SUB_AGENT_NAME = "Credit Bureau Lookup"
SUB_AGENT_ID = "credit_bureau_lookup"
MODEL = "gpt-4o-mini"


def env(name: str) -> str:
    """Read a required setting, or stop with a message naming what is missing."""
    value = os.environ.get(name, "").strip()
    if not value:
        sys.exit(f"FAILED: {name} is not set. See 'Export your settings' in the docs.")
    return value


snapshot = InventorySnapshot(
    source="lending_platform",
    account="prod.example.com",
    vendor="Acme Corp",
    product_name="Lending Platform",
    environment="prod",
)
snapshot.assets += [
    Asset(
        kind=AssetKind.AGENT,
        id=AGENT_ID,
        name=AGENT_NAME,
        description="Reviews a loan application against underwriting criteria",
        owners=["lending-eng@example.com"],
        provider="openai",
        model_name=MODEL,
        properties={
            "provenance_json": json.dumps(
                {"categories": ["lending", "underwriting", "credit_risk"]}
            ),
        },
    ),
    Asset(
        kind=AssetKind.AGENT,
        id=SUB_AGENT_ID,
        name=SUB_AGENT_NAME,
        description="Fetches and explains a credit bureau report",
        owners=["lending-eng@example.com"],
        provider="openai",
        model_name=MODEL,
    ),
    Asset(kind=AssetKind.MODEL, id=f"openai:{MODEL}", name=MODEL, provider="openai"),
]
snapshot.relationships += [
    Relationship(
        kind=RelationshipKind.USES_MODEL,
        from_id=AGENT_ID,
        from_kind=AssetKind.AGENT,
        to_id=f"openai:{MODEL}",
        to_kind=AssetKind.MODEL,
    ),
    Relationship(
        kind=RelationshipKind.CAN_INVOKE,
        from_id=AGENT_ID,
        from_kind=AssetKind.AGENT,
        to_id=SUB_AGENT_ID,
        to_kind=AssetKind.AGENT,
    ),
]

# ---- 1. discovery: publish the agent so the trace has something to link to --
config = InventoryClientConfig(api_key=env("TRUST3_API_KEY"))
print(f"[1/3] publishing inventory to {config.api_server_host} ...")
try:
    with InventoryClient(config) as client:
        result = client.submit(snapshot)
except InventoryValidationError as exc:
    sys.exit(f"FAILED: governance rejected the inventory (HTTP {exc.status_code}): {exc.response_text}")
except InventoryTransportError as exc:
    sys.exit(f"FAILED: could not reach {config.api_server_host} after retries: {exc}")

print(
    f"      ok: {result.assets_pushed} assets, "
    f"{result.relationships_pushed} relationships, "
    f"governance={result.governance_status}"
)

# ---- 2. monitoring: point tracing at the Trust3 endpoint --------------------
print("[2/3] configuring tracing ...")
endpoint = env("TRUST3_TRACE_ENDPOINT")
token = base64.b64encode(
    f'{env("TRUST3_TRACE_USERNAME")}:{env("TRUST3_TRACE_PASSWORD")}'.encode()
).decode()

provider = TracerProvider(resource=Resource.create({"service.name": "lending-platform"}))
provider.add_span_processor(
    BatchSpanProcessor(
        OTLPSpanExporter(endpoint=endpoint, headers={"Authorization": f"Basic {token}"})
    )
)

# Placeholder keys + should_export_span=False: Langfuse instruments but never
# exports, leaving the Trust3 collector as the only destination.
Langfuse(
    public_key="pk-local",
    secret_key="sk-local",
    tracer_provider=provider,
    should_export_span=lambda span: False,
)
print(f"      ok: exporting to {endpoint}")

# ---- 3. a minimal graph, standing in for your real one ---------------------
print("[3/3] building the graph ...")
llm = ChatOpenAI(model=MODEL, temperature=0)


class State(TypedDict):
    messages: Annotated[list, add_messages]


def review(state: State) -> dict:
    return {"messages": [llm.invoke(state["messages"])]}


builder = StateGraph(State)
builder.add_node("review", review)
builder.add_edge(START, "review")
graph = builder.compile()
handler = CallbackHandler()

# config goes on the OUTERMOST invocation, so run_name becomes the root span of
# the whole run and every model and tool call nests beneath it.
print(f'      running "{AGENT_NAME}" ...')
graph.invoke(
    {"messages": [{"role": "user",
                   "content": "Applicant earns 90k, wants 250k over 25 years. Proceed?"}]},
    config={
        "callbacks": [handler],
        "run_name": AGENT_NAME,          # must equal the published agent name
        "metadata": {"user.id": "alice@example.com"},
    },
)
print("      ok")

# A delegated sub-agent is a trace in its own right, so it needs its own run_name —
# the root span name is what Trust3 matches, for sub-agents exactly as for agents.
# In a real system this would be its own graph; the same one is reused here.
print(f'      running "{SUB_AGENT_NAME}" as a delegated sub-agent ...')
graph.invoke(
    {"messages": [{"role": "user",
                   "content": "Summarise this applicant's credit history."}]},
    config={
        "callbacks": [handler],
        "run_name": SUB_AGENT_NAME,      # must equal the published sub-agent name
        "metadata": {"user.id": "alice@example.com"},
    },
)
print("      ok")

# Spans are batched — without this a short script exits having exported nothing.
print("      flushing spans ...")
provider.shutdown()

print()
print("[done] script finished without errors.")
print(f'       Open AI Assets and find "Loan Application Reviewer" and "Credit Bureau Lookup".')
print("       Traces are picked up on the collector's scan interval, 15 minutes by default,")
print("       so activity can take that long to appear.")
print("       If the endpoint or credentials were wrong, the OpenTelemetry SDK logs an export")
print("       error above — the LLM call itself still succeeds, so watch for that.")

#!/usr/bin/env python3
"""Publish one agent, then trace a direct model call with no agent or graph."""

import base64
import json
import os
import sys

from langchain_openai import ChatOpenAI
from langfuse import Langfuse
from langfuse.langchain import CallbackHandler
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from trust3_ai_collector_sdk import (
    Asset,
    AssetKind,
    InventoryClient,
    InventoryClientConfig,
    InventorySnapshot,
    Relationship,
    RelationshipKind,
)
from trust3_ai_collector_sdk.errors import (
    InventoryTransportError,
    InventoryValidationError,
)

AGENT_NAME = "Support Ticket Summarizer"
AGENT_ID = "support_ticket_summarizer"
MODEL = "gpt-4o-mini"


def env(name: str) -> str:
    """Read a required setting, or stop with a message naming what is missing."""
    value = os.environ.get(name, "").strip()
    if not value:
        sys.exit(f"FAILED: {name} is not set. See 'Export your settings' in the docs.")
    return value


snapshot = InventorySnapshot(
    source="support_desk",
    account="prod.example.com",
    vendor="Acme Corp",
    product_name="Support Desk",
    environment="prod",
)
snapshot.assets += [
    Asset(
        kind=AssetKind.AGENT,
        id=AGENT_ID,
        name=AGENT_NAME,
        description="Condenses a support ticket thread into a one-line summary",
        owners=["support-eng@example.com"],
        provider="openai",
        model_name=MODEL,
        properties={
            "provenance_json": json.dumps({"categories": ["customer_support", "ticket_triage"]}),
        },
    ),
    Asset(kind=AssetKind.MODEL, id=f"openai:{MODEL}", name=MODEL, provider="openai"),
]
snapshot.relationships += [
    Relationship(
        kind=RelationshipKind.USES_MODEL,
        from_id=AGENT_ID,
        from_kind=AssetKind.AGENT,
        to_id=f"openai:{MODEL}",
        to_kind=AssetKind.MODEL,
    ),
]

# ---- 1. discovery: publish the agent so the trace has something to link to --
config = InventoryClientConfig(api_key=env("TRUST3_API_KEY"))
print(f"[1/3] publishing inventory to {config.api_server_host} ...")
try:
    with InventoryClient(config) as client:
        result = client.submit(snapshot)
except InventoryValidationError as exc:
    sys.exit(f"FAILED: governance rejected the inventory (HTTP {exc.status_code}): {exc.response_text}")
except InventoryTransportError as exc:
    sys.exit(f"FAILED: could not reach {config.api_server_host} after retries: {exc}")

print(
    f"      ok: {result.assets_pushed} assets, "
    f"{result.relationships_pushed} relationships, "
    f"governance={result.governance_status}"
)

# ---- 2. monitoring: point tracing at the Trust3 endpoint --------------------
print("[2/3] configuring tracing ...")
endpoint = env("TRUST3_TRACE_ENDPOINT")
token = base64.b64encode(
    f'{env("TRUST3_TRACE_USERNAME")}:{env("TRUST3_TRACE_PASSWORD")}'.encode()
).decode()

provider = TracerProvider(resource=Resource.create({"service.name": "support-desk"}))
provider.add_span_processor(
    BatchSpanProcessor(
        OTLPSpanExporter(endpoint=endpoint, headers={"Authorization": f"Basic {token}"})
    )
)

# Placeholder keys + should_export_span=False: Langfuse instruments but never
# exports, leaving the Trust3 collector as the only destination.
langfuse = Langfuse(
    public_key="pk-local",
    secret_key="sk-local",
    tracer_provider=provider,
    should_export_span=lambda span: False,
)
print(f"      ok: exporting to {endpoint}")

# ---- 3. a plain call, wrapped in the span that makes it linkable -----------
print(f'[3/3] calling {MODEL} as "{AGENT_NAME}" ...')
llm = ChatOpenAI(model=MODEL, temperature=0)

# There is no agent or graph here to produce a root span, so open one yourself
# and name it after the published asset. The model call goes inside it.
with langfuse.start_as_current_observation(as_type="span", name=AGENT_NAME):
    reply = llm.invoke(
        "Summarise in one line: customer cannot log in after the 9.2 upgrade, "
        "cleared cache, still failing, needs it today.",
        config={
            "callbacks": [CallbackHandler()],
            "run_name": AGENT_NAME,      # names the model call within the trace
            "metadata": {"user.id": "alice@example.com"},
        },
    )
print(f"      ok: {len(reply.content)} characters returned")

# Spans are batched — without this a short script exits having exported nothing.
print("      flushing spans ...")
provider.shutdown()

print()
print("[done] script finished without errors.")
print(f'       Open AI Assets in the portal and find "Support Ticket Summarizer".')
print("       Traces are picked up on the collector's scan interval, 15 minutes by default,")
print("       so activity can take that long to appear.")
print("       If the endpoint or credentials were wrong, the OpenTelemetry SDK logs an export")
print("       error above — the LLM call itself still succeeds, so watch for that.")