diff --git a/examples/agent/langchain-agent-http/.env.example b/examples/agent/langchain-agent-http/.env.example new file mode 100644 index 00000000..f07e8497 --- /dev/null +++ b/examples/agent/langchain-agent-http/.env.example @@ -0,0 +1,17 @@ +# Splunk Agent Observability — copy this file to .env and fill in your values. + +SPLUNK_AO_API_KEY=your-splunk-ao-api-key +SPLUNK_AO_API_URL=https://agent-observability-api.us0.signalfx.com +SPLUNK_AO_CONSOLE_URL=https://agent-observability.us0.signalfx.com +SPLUNK_AO_USE_DIRECT_API=true +SPLUNK_AO_PROJECT=your-project-name +SPLUNK_AO_LOG_STREAM=your-log-stream + +# OpenAI (standard) +OPENAI_API_KEY=your-openai-api-key +OPENAI_MODEL=gpt-4o-mini + +# Azure OpenAI (optional — overrides OPENAI_API_KEY / OPENAI_MODEL if set) +# OPENAI_API_KEY=your-azure-openai-api-key +# OPENAI_BASE_URL=https://your-resource.cognitiveservices.azure.com/openai/v1/ +# AZURE_OPENAI_API_VERSION=2024-02-01 diff --git a/examples/agent/langchain-agent-http/.gitignore b/examples/agent/langchain-agent-http/.gitignore new file mode 100644 index 00000000..fc36f7d8 --- /dev/null +++ b/examples/agent/langchain-agent-http/.gitignore @@ -0,0 +1,4 @@ +.env +__pycache__/ +*.pyc +.venv/ diff --git a/examples/agent/langchain-agent-http/app.py b/examples/agent/langchain-agent-http/app.py new file mode 100644 index 00000000..5805ed56 --- /dev/null +++ b/examples/agent/langchain-agent-http/app.py @@ -0,0 +1,177 @@ +"""FastAPI HTTP wrapper around the langchain-agent example. + +Exposes POST /invoke and POST /invoke/nested endpoints that accept a JSON body +with a `prompt` field, pass it to a LangGraph ReAct agent, and return the response. + +This mirrors the "Scenarios" test layout from splunk-otel-python-contrib PR #236: + • HTTP span (FastAPI request) → GenAI Workflow span (LangChain/LangGraph agent) + • The Workflow span has no GenAI parent, so `gen_ai.conversation_root` is set + to True via the auto-detection logic in SplunkAOLogger.add_workflow_span() + (native path) and start_splunk_ao_span() (OTel path). + +Verification: + 1. Start the server: + uvicorn app:app --reload --port 8080 + 2. Send a request: + curl -X POST http://localhost:8080/invoke \\ + -H "Content-Type: application/json" \\ + -d '{"prompt": "Say hello to Erin"}' + 3. Observe the trace in the Splunk AO console at SPLUNK_AO_CONSOLE_URL and + confirm the root WorkflowSpan carries gen_ai.conversation_root=true in its + user_metadata (and on the LoggedWorkflowSpan.conversation_root field). + +Environment variables (see .env.example): + SPLUNK_AO_API_KEY, SPLUNK_AO_API_URL, SPLUNK_AO_PROJECT, SPLUNK_AO_LOG_STREAM, + OPENAI_API_KEY (or OPENAI_BASE_URL + AZURE_OPENAI_* for Azure). +""" + +import logging +import os + +from dotenv import load_dotenv +from fastapi import FastAPI, HTTPException +from langchain.tools import tool +from langchain_openai import ChatOpenAI +from langgraph.prebuilt import create_react_agent +from pydantic import BaseModel + +from splunk_ao import splunk_ao_context +from splunk_ao.handlers.langchain import SplunkAOCallback + +load_dotenv() + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +app = FastAPI( + title="LangChain Agent HTTP — gen_ai.conversation_root demo", + description="Wraps a LangGraph ReAct agent in FastAPI for e2e conversation-root tracing verification.", + version="0.1.0", +) + + +# --------------------------------------------------------------------------- +# Agent setup — built once at startup, reused across requests. +# --------------------------------------------------------------------------- + +_PROJECT = os.getenv("SPLUNK_AO_PROJECT", "langchain-http-demo") +_LOG_STREAM = os.getenv("SPLUNK_AO_LOG_STREAM", "langchain-http") + + +@tool +def greet(name: str) -> str: + """Say hello to someone by name.""" + return f"Hello, {name}! 👋" + + +@tool +def get_weather(city: str) -> str: + """Return a made-up weather report for a city (demo tool).""" + return f"It's sunny and 22 °C in {city} right now." + + +_llm = ChatOpenAI( + model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"), + temperature=0.7, +) + +_agent = create_react_agent( + model=_llm, + tools=[greet, get_weather], +) + + +def _invoke_with_tracing(prompt: str) -> str: + """Run the agent under a splunk_ao_context so each invocation is traced.""" + callback = SplunkAOCallback() + with splunk_ao_context(project=_PROJECT, log_stream=_LOG_STREAM): + result = _agent.invoke( + {"messages": [{"role": "user", "content": prompt}]}, + config={"callbacks": [callback]}, + ) + # Last message in the graph output is the final AI response. + messages = result.get("messages", []) + return messages[-1].content if messages else str(result) + + +# --------------------------------------------------------------------------- +# Request / response models +# --------------------------------------------------------------------------- + + +class InvokeRequest(BaseModel): + """Payload accepted by POST /invoke.""" + + prompt: str + """The user prompt to send to the agent (e.g. 'Say hello to Erin').""" + + +class InvokeResponse(BaseModel): + """Response returned by POST /invoke.""" + + response: str + """The agent's final text output.""" + note: str = ( + "The root WorkflowSpan for this request should carry " + "gen_ai.conversation_root=true (user_metadata) and " + "LoggedWorkflowSpan.conversation_root=True." + ) + + +# --------------------------------------------------------------------------- +# Routes +# --------------------------------------------------------------------------- + + +@app.get("/health") +def health() -> dict: + """Liveness probe.""" + return {"status": "ok"} + + +@app.post("/invoke", response_model=InvokeResponse) +def invoke_agent(body: InvokeRequest) -> InvokeResponse: + """Invoke the LangGraph ReAct agent with a user prompt. + + The HTTP span (this FastAPI handler) is *not* a GenAI span, so the first + GenAI WorkflowSpan created by the SplunkAOCallback will be detected as the + conversation root (gen_ai.conversation_root = True). + + Scenario mirrors PR #236 test cases: + - single_agent_under_http: HTTP request → WorkflowSpan (root = True) + """ + logger.info("Received prompt: %s", body.prompt) + try: + output = _invoke_with_tracing(body.prompt) + except Exception as exc: + logger.exception("Agent invocation failed") + raise HTTPException(status_code=500, detail=str(exc)) from exc + + logger.info("Agent response: %s", output) + return InvokeResponse(response=output) + + +@app.post("/invoke/nested", response_model=InvokeResponse) +def invoke_nested(body: InvokeRequest) -> InvokeResponse: + """Invoke two sequential agent calls in one HTTP request. + + Mirrors the 'two_sequential_agents' scenario from PR #236: + each _invoke_with_tracing() call produces a separate trace, and each + trace's root WorkflowSpan gets conversation_root=True independently. + """ + logger.info("Received nested prompt: %s", body.prompt) + try: + out1 = _invoke_with_tracing(body.prompt) + out2 = _invoke_with_tracing(f"Summarise in one sentence: {out1}") + except Exception as exc: + logger.exception("Nested agent invocation failed") + raise HTTPException(status_code=500, detail=str(exc)) from exc + + output = f"[Pass 1] {out1} | [Pass 2] {out2}" + return InvokeResponse(response=output) + + +if __name__ == "__main__": + import uvicorn + + uvicorn.run("app:app", host="0.0.0.0", port=8080, reload=True) diff --git a/examples/agent/langchain-agent-http/requirements.txt b/examples/agent/langchain-agent-http/requirements.txt new file mode 100644 index 00000000..472ce2f8 --- /dev/null +++ b/examples/agent/langchain-agent-http/requirements.txt @@ -0,0 +1,7 @@ +fastapi==0.139.2 +uvicorn[standard]==0.38.0 +python-dotenv==1.2.1 +langchain==1.2.15 +langchain-openai==1.2.0 +openai==2.41.0 +splunk-ao>=0.1.0 diff --git a/src/splunk_ao/logger/logger.py b/src/splunk_ao/logger/logger.py index 5879c938..71bb2cb6 100644 --- a/src/splunk_ao/logger/logger.py +++ b/src/splunk_ao/logger/logger.py @@ -1664,6 +1664,11 @@ def add_workflow_span( id=uuid.uuid4(), step_number=step_number, ) + # Auto-mark as conversation root when this workflow is the first GenAI span + # in the trace (its direct parent is the LoggedTrace, not another GenAI span). + if isinstance(self.current_parent(), LoggedTrace): + span.conversation_root = True + span.user_metadata = {**(span.user_metadata or {}), "gen_ai.conversation_root": "true"} return self._attach_parentable_span(span, status_code) @nop_sync @@ -1747,6 +1752,11 @@ def add_agent_span( id=uuid.uuid4(), step_number=step_number, ) + # Auto-mark as conversation root when this agent span is the first GenAI span + # in the trace (its direct parent is the LoggedTrace, not another GenAI span). + if isinstance(self.current_parent(), LoggedTrace): + span.conversation_root = True + span.user_metadata = {**(span.user_metadata or {}), "gen_ai.conversation_root": "true"} return self._attach_parentable_span(span, status_code) @nop_sync diff --git a/src/splunk_ao/otel.py b/src/splunk_ao/otel.py index f60f32ee..03c08a3f 100644 --- a/src/splunk_ao/otel.py +++ b/src/splunk_ao/otel.py @@ -9,7 +9,7 @@ from requests import Session -from galileo_core.schemas.logging.span import RetrieverSpan, ToolSpan, WorkflowSpan +from galileo_core.schemas.logging.span import AgentSpan, RetrieverSpan, ToolSpan, WorkflowSpan from galileo_core.schemas.logging.span import Span as GalileoSpan from splunk_ao.config import SplunkAOConfig from splunk_ao.decorator import ( @@ -26,6 +26,9 @@ logger = logging.getLogger(__name__) +# Semantic-convention attribute key for marking the invocation-level GenAI root span. +# Mirrors the constant from opentelemetry-util-genai (splunk-otel-python-contrib PR #236). +GEN_AI_CONVERSATION_ROOT = "gen_ai.conversation_root" INSTALL_ERR_MSG = ( "OpenTelemetry packages are not installed. " @@ -407,9 +410,18 @@ def start_splunk_ao_span(galileo_span: GalileoSpan) -> Generator[trace.Span, Any tracer_provider = trace.get_tracer_provider() _TRACE_PROVIDER_CONTEXT_VAR.set(cast(TracerProvider, tracer_provider)) tracer = tracer_provider.get_tracer("galileo-tracer") + # Capture root status BEFORE entering the span context so we see the caller's + # OTel parent (or absence of one). A Workflow/Agent span with no valid OTel + # parent is, by definition, the conversation root for this trace. + _is_conversation_root = ( + not trace.get_current_span().get_span_context().is_valid + and isinstance(galileo_span, (WorkflowSpan, AgentSpan)) + ) with tracer.start_as_current_span(galileo_span.name) as span: yield span span.set_attribute("gen_ai.system", "galileo-otel") + if _is_conversation_root: + span.set_attribute(GEN_AI_CONVERSATION_ROOT, True) # Set dataset attributes for ground truth/reference output support _apply_dataset_attributes( span, galileo_span.dataset_input, galileo_span.dataset_output, galileo_span.dataset_metadata diff --git a/src/splunk_ao/schema/logged.py b/src/splunk_ao/schema/logged.py index 6f1b90bb..7eee4752 100644 --- a/src/splunk_ao/schema/logged.py +++ b/src/splunk_ao/schema/logged.py @@ -61,6 +61,9 @@ class LoggedWorkflowSpan(WorkflowSpan): output: IngestOutputType | None = _OUTPUT_FIELD redacted_output: IngestOutputType | None = _REDACTED_OUTPUT_FIELD spans: list["LoggedSpan"] = Field(default_factory=list) + # When True, marks this span as the invocation-level GenAI root for the trace + # (mirrors gen_ai.conversation_root from the OTel semantic conventions). + conversation_root: bool | None = Field(default=None) class LoggedAgentSpan(AgentSpan): @@ -71,6 +74,9 @@ class LoggedAgentSpan(AgentSpan): output: IngestOutputType | None = _OUTPUT_FIELD redacted_output: IngestOutputType | None = _REDACTED_OUTPUT_FIELD spans: list["LoggedSpan"] = Field(default_factory=list) + # When True, marks this span as the invocation-level GenAI root for the trace + # (mirrors gen_ai.conversation_root from the OTel semantic conventions). + conversation_root: bool | None = Field(default=None) class LoggedLlmSpan(LlmSpan):