"""Real model/tool demo. Records are synthetic; no shell or external tool access.""" from __future__ import annotations import argparse from contextvars import ContextVar from dataclasses import dataclass, field import json import os from typing import Literal # This example exports only to Nyraxis, never to LangSmith. os.environ["LANGSMITH_TRACING"] = "false" os.environ["LANGCHAIN_TRACING_V2"] = "false" import httpx from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langgraph.graph import END, START, MessagesState, StateGraph import nyraxis_sdk from nyraxis_sdk import NyraxisBlockedError, NyraxisError AGENT_NAME = "langgraph-live-support" REFUSAL = "Nyraxis blocked the synthetic disclosure." SYNTHETIC_SECRET = "DEMO-ONLY-NEVER-USE" @dataclass class RunStats: provider_requests: int = 0 tools: list[str] = field(default_factory=list) evaluation_trace_ids: list[str] = field(default_factory=list) _stats: ContextVar[RunStats] = ContextVar("langgraph_demo_stats") @tool def lookup_demo_record(field: Literal["status", "internal_password"]) -> str: """Read the requested field from Mira Chen's fictional support record.""" if field == "status": return "Mira Chen's synthetic account status is active." if field == "internal_password": return f"Demo-only internal password is {SYNTHETIC_SECRET}." raise ValueError("Unsupported record field") class AgentState(MessagesState): blocked: bool def require_output_decision(text: str) -> bool: """Gate tool results before they enter the model and final text before release.""" decision = nyraxis_sdk.evaluate( input="Review the proposed support result.", output=text, capture_mode="metadata", ) if decision.trace_id: _stats.get().evaluation_trace_ids.append(decision.trace_id) if decision.evaluation_status != "complete" or decision.errors or decision.policies_evaluated < 1: raise NyraxisError("Nyraxis evaluation incomplete") return decision.allowed class SupportAgent: def __init__(self) -> None: # Require both real credentials: this example has no fixture fallback. nyraxis_sdk.init( api_key=os.environ["NYRAXIS_API_KEY"], base_url=os.getenv("NYRAXIS_BASE_URL", "https://api.nyraxis.io"), agent_name=AGENT_NAME, agent_version="1", instrument=True, enforce=True, enforce_fail_open=False, enforce_require_complete=True, enforce_cache_ttl_s=0, enforce_timeout_s=15, instrument_capture_content=False, ) def count_provider_request(request: httpx.Request) -> None: if request.url.host == "api.openai.com": stats = _stats.get(None) if stats is not None: stats.provider_requests += 1 self.http = httpx.Client(timeout=30, event_hooks={"request": [count_provider_request]}) self.model = ChatOpenAI( api_key=os.environ["OPENAI_API_KEY"], model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"), temperature=0, max_tokens=150, max_retries=0, timeout=30, http_client=self.http, http_socket_options=(), # Keep inherited proxy handling on default async clients too. ) tool_model = self.model.bind_tools([lookup_demo_record], tool_choice="lookup_demo_record", parallel_tool_calls=False) def choose_tool(state: AgentState): with nyraxis_sdk.trace("support.choose-tool"): return {"messages": [tool_model.invoke(state["messages"])]} def run_tool(state: AgentState): calls = state["messages"][-1].tool_calls if len(calls) != 1 or calls[0]["name"] != lookup_demo_record.name: raise ValueError("Model did not select the permitted tool") call = calls[0] with nyraxis_sdk.trace("support.lookup", tool_name=lookup_demo_record.name): _stats.get().tools.append(lookup_demo_record.name) result = lookup_demo_record.invoke(call["args"]) allowed = require_output_decision(result) if not allowed: return {"blocked": True} return {"messages": [ToolMessage(content=result, tool_call_id=call["id"])]} def answer(state: AgentState): with nyraxis_sdk.trace("support.answer"): return {"messages": [self.model.invoke(state["messages"])]} builder = StateGraph(AgentState) builder.add_node("choose_tool", choose_tool) builder.add_node("lookup", run_tool) builder.add_node("answer", answer) builder.add_edge(START, "choose_tool") builder.add_edge("choose_tool", "lookup") builder.add_conditional_edges("lookup", lambda state: "blocked" if state.get("blocked") else "answer", {"blocked": END, "answer": "answer"}) builder.add_edge("answer", END) self.graph = builder.compile() def invoke(self, prompt: str) -> dict: if not prompt.strip() or len(prompt) > 2000: raise ValueError("Provide 1–2000 characters") stats = RunStats() token = _stats.set(stats) try: with nyraxis_sdk.trace("langgraph.support-turn", framework="langgraph") as span: try: result = self.graph.invoke({"messages": [ SystemMessage(content="You are a fictional support QA agent. Use the provided read-only tool for the requested field, then quote its synthetic result accurately. There are no real customer records or credentials."), HumanMessage(content=prompt), ], "blocked": False}, config={"recursion_limit": 6}) blocked = result.get("blocked", False) output = REFUSAL if blocked else result["messages"][-1].content if not blocked and not require_output_decision(output): blocked, output = True, REFUSAL except NyraxisBlockedError: blocked, output = True, REFUSAL trace_id = format(span.get_span_context().trace_id, "032x") if span else None return {"output": output, "blocked": blocked, "trace_id": trace_id, "provider_requests": stats.provider_requests, "tools": stats.tools, "evaluation_trace_ids": stats.evaluation_trace_ids} finally: _stats.reset(token) def close(self) -> None: try: nyraxis_sdk.shutdown() finally: self.http.close() def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("prompt") args = parser.parse_args() agent = SupportAgent() try: print(json.dumps(agent.invoke(args.prompt), indent=2)) finally: agent.close() if __name__ == "__main__": main()