complexer worden, schieten traditionele houtkap en monitoring tekort. Wat teams eigenlijk nodig hebben is waarneembaarheid: de mogelijkheid om beslissingen van agenten te traceren, de responskwaliteit automatisch te evalueren en drift in de loop van de tijd te detecteren, zonder grote hoeveelheden aangepaste evaluatie- en telemetriecode te schrijven en te onderhouden.
Daarom moeten teams het juiste platform voor waarneembaarheid aannemen, terwijl ze zich concentreren op de kerntaak van het opbouwen en verbeteren van de orkestratie van de agenten. En integreer hun toepassing met het observatieplatform met minimale overhead voor hun functionele code. In dit artikel laat ik zien hoe je een open-source AI-observatieplatform kunt opzetten om het volgende uit te voeren met behulp van een minimale code-aanpak:
- LLM-als-rechter: Configureer vooraf gebouwde beoordelaars om antwoorden te scoren op correctheid, relevantie, hallucinatie en meer. Geef scores over runs weer met gedetailleerde logboeken en analyses.
- Testen op schaal: Zet datasets op om regressietestgevallen op te slaan voor het meten van de nauwkeurigheid ten opzichte van de verwachte grondwaarheidsreacties. Detecteer proactief LLM en agent-drift.
- MELT-gegevens: Houd statistieken bij (latentie, tokengebruik, modeldrift), gebeurtenissen (API-aanroepen, LLM-aanroepen, toolgebruik), logboeken (gebruikersinteractie, tooluitvoering, besluitvorming door agenten) met gedetailleerde traceringen – allemaal zonder gedetailleerde telemetrie- en instrumentatiecode.
We zullen Langfuse gebruiken voor waarneembaarheid. Het is open-source en framework-agnostisch en kan werken met populaire orkestratieframeworks en LLM-providers.
Toepassing met meerdere agenten
Voor deze demonstratie heb ik de LangGraph-code van een klantenservicetoepassing bijgevoegd. De applicatie accepteert tickets van de gebruiker, classificeert deze in Technisch, Facturering of Beide met behulp van een triage-agent en stuurt deze vervolgens door naar de technische ondersteuningsagent, de factureringsondersteuningsagent of naar beide. Vervolgens synthetiseert een finalizer-agent de reactie van beide agenten in een samenhangend, beter leesbaar formaat. Het stroomschema is als volgt:
De code is hier bijgevoegd
# --------------------------------------------------
# 0. Load .env
# --------------------------------------------------
from dotenv import load_dotenv
load_dotenv(override=True)
# --------------------------------------------------
# 1. Imports
# --------------------------------------------------
import os
from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import AzureChatOpenAI
from langfuse import Langfuse
from langfuse.langchain import CallbackHandler
# --------------------------------------------------
# 2. Langfuse Client (WORKING CONFIG)
# --------------------------------------------------
langfuse = Langfuse(
host="https://cloud.langfuse.com",
public_key=os.environ("LANGFUSE_PUBLIC_KEY") ,
secret_key=os.environ("LANGFUSE_SECRET_KEY")
)
langfuse_callback = CallbackHandler()
os.environ("LANGGRAPH_TRACING") = "false"
# --------------------------------------------------
# 3. Azure OpenAI Setup
# --------------------------------------------------
llm = AzureChatOpenAI(
azure_deployment=os.environ("AZURE_OPENAI_DEPLOYMENT_NAME"),
api_version=os.environ.get("AZURE_OPENAI_API_VERSION", "2025-01-01-preview"),
temperature=0.2,
callbacks=(langfuse_callback), # 🔑 enables token usage
)
# --------------------------------------------------
# 4. Shared State
# --------------------------------------------------
class AgentState(TypedDict, total=False):
ticket: str
category: str
technical_response: str
billing_response: str
final_response: str
# --------------------------------------------------
# 5. Agent Definitions
# --------------------------------------------------
def triage_agent(state: dict) -> dict:
with langfuse.start_as_current_observation(
as_type="span",
name="triage_agent",
input={"ticket": state("ticket")},
) as span:
span.update_trace(name="Customer Service Query - LangGraph Demo")
response = llm.invoke((
{
"role": "system",
"content": (
"Classify the query as one of: "
"Technical, Billing, Both. "
"Respond with only the label."
),
},
{"role": "user", "content": state("ticket")},
))
raw = response.content.strip().lower()
if "both" in raw:
category = "Both"
elif "technical" in raw:
category = "Technical"
elif "billing" in raw:
category = "Billing"
else:
category = "Technical" # ✅ safe fallback
span.update(output={"raw": raw, "category": category})
return {"category": category}
def technical_support_agent(state: dict) -> dict:
with langfuse.start_as_current_observation(
as_type="span",
name="technical_support_agent",
input={
"ticket": state("ticket"),
"category": state.get("category"),
},
) as span:
response = llm.invoke((
{
"role": "system",
"content": (
"You are a technical support specialist. "
"Provide a clear, step-by-step solution."
),
},
{"role": "user", "content": state("ticket")},
))
answer = response.content
span.update(output={"technical_response": answer})
return {"technical_response": answer}
def billing_support_agent(state: dict) -> dict:
with langfuse.start_as_current_observation(
as_type="span",
name="billing_support_agent",
input={
"ticket": state("ticket"),
"category": state.get("category"),
},
) as span:
response = llm.invoke((
{
"role": "system",
"content": (
"You are a billing support specialist. "
"Answer clearly about payments, invoices, or accounts."
),
},
{"role": "user", "content": state("ticket")},
))
answer = response.content
span.update(output={"billing_response": answer})
return {"billing_response": answer}
def finalizer_agent(state: dict) -> dict:
with langfuse.start_as_current_observation(
as_type="span",
name="finalizer_agent",
input={
"ticket": state("ticket"),
"technical": state.get("technical_response"),
"billing": state.get("billing_response"),
},
) as span:
parts = (
f"Technical:n{state('technical_response')}"
for k in ("technical_response")
if state.get(k)
) + (
f"Billing:n{state('billing_response')}"
for k in ("billing_response")
if state.get(k)
)
if not parts:
final = "Error: No agent responses available."
else:
response = llm.invoke((
{
"role": "system",
"content": (
"Combine the following agent responses into ONE clear, professional, "
"customer-facing answer. Do not mention agents or internal labels. "
f"Answer the user's query: '{state('ticket')}'."
),
},
{"role": "user", "content": "nn".join(parts)},
))
final = response.content
span.update(output={"final_response": final})
return {"final_response": final}
# --------------------------------------------------
# 6. LangGraph Construction
# --------------------------------------------------
builder = StateGraph(AgentState)
builder.add_node("triage", triage_agent)
builder.add_node("technical", technical_support_agent)
builder.add_node("billing", billing_support_agent)
builder.add_node("finalizer", finalizer_agent)
builder.set_entry_point("triage")
# Conditional routing
builder.add_conditional_edges(
"triage",
lambda state: state("category"),
{
"Technical": "technical",
"Billing": "billing",
"Both": "technical",
"__default__": "technical", # ✅ never dead-end
},
)
# Sequential resolution
builder.add_conditional_edges(
"technical",
lambda state: state("category"),
{
"Both": "billing", # Proceed to billing if Both
"__default__": "finalizer",
},
)
builder.add_edge("billing", "finalizer")
builder.add_edge("finalizer", END)
graph = builder.compile()
# --------------------------------------------------
# 9. Main
# --------------------------------------------------
if __name__ == "__main__":
print("===============================================")
print(" Conditional Multi-Agent Support System (Ready)")
print("===============================================")
print("Enter 'exit' or 'quit' to stop the program.n")
while True:
# Get user input for the ticket
ticket = input("Enter your support query (ticket): ")
# Check for exit command
if ticket.lower() in ("exit", "quit"):
print("nExiting the support system. Goodbye!")
break
if not ticket.strip():
print("Please enter a non-empty query.")
continue
try:
# --- Run the graph with the user's ticket ---
result = graph.invoke(
{"ticket": ticket},
config={"callbacks": (langfuse_callback)},
)
# --- Print Results ---
category = result.get('category', 'N/A')
print(f"n✅ Triage Classification: **{category}**")
# Check which agents were executed based on the presence of a response
executed_agents = ()
if result.get("technical_response"):
executed_agents.append("Technical")
if result.get("billing_response"):
executed_agents.append("Billing")
print(f"🛠️ Agents Executed: {', '.join(executed_agents) if executed_agents else 'None (Triage Failed)'}")
print("n================ FINAL RESPONSE ================n")
print(result("final_response"))
print("n" + "="*60 + "n")
except Exception as e:
# This is important for debugging: print the exception type and message
print(f"nAn error occurred during processing ({type(e).__name__}): {e}")
print("nPlease try another query.")
print("n" + "="*60 + "n")
Waarneembaarheidsconfiguratie
Om Langfuse in te stellen, gaat u naar https://cloud.langfuse.com/ en maakt u een account aan met een factureringsniveau (hobbyniveau met royale limieten beschikbaar) en stelt u vervolgens een project in. In de projectinstellingen kunt u de openbare en geheime sleutels genereren die aan het begin van de code moeten worden opgegeven. U moet ook de LLM-verbinding toevoegen, die wordt gebruikt voor de LLM-als-rechter-evaluatie.
LLM-als-rechter-opstelling
Dit is de kern van de prestatie-evaluatie-instellingen voor agenten. Hier kunt u verschillende vooraf gebouwde Evaluators uit de Evaluator-bibliotheek configureren, die de antwoorden op verschillende criteria beoordelen, zoals Beknoptheid, Correctheid, Hallucinatie, Antwoordcriticus enz. Deze zouden voor de meeste gebruiksscenario’s moeten volstaan, anders kunnen ook aangepaste Evaluators worden ingesteld. Hier is een weergave van de Evaluator-bibliotheek:
Selecteer de evaluator, zeg Relevantie, die u wilt gebruiken. U kunt ervoor kiezen om het uit te voeren voor nieuwe of bestaande traceringen of voor gegevenssetuitvoeringen. Controleer bovendien de evaluatieprompt om er zeker van te zijn dat deze voldoet aan uw evaluatiedoelstelling. Het belangrijkste is dat de query-, generatie- en andere variabelen correct moeten worden toegewezen aan de bron (meestal aan de invoer en uitvoer van de applicatietracering). In ons geval zijn dit respectievelijk de ticketgegevens die door de gebruiker zijn ingevoerd en het antwoord dat door de finalizer-agent is gegenereerd. Bovendien kunt u voor Dataset-uitvoeringen de gegenereerde reacties vergelijken met de Ground Truth-reacties die zijn opgeslagen als verwachte uitvoer (uitgelegd in de volgende secties).
Hier is de configuratie voor de ‘GT-nauwkeurigheid‘ evaluatie die ik heb opgezet voor nieuwe Dataset-runs, samen met de Variabele mapping. Het voorbeeld van de evaluatieprompt wordt ook weergegeven. De meeste beoordelaars scoren binnen een bereik van 0 tot 1:
Voor de klantenservicedemo heb ik 3 evaluatoren geconfigureerd: Relevantie, beknoptheid die lopen voor alle nieuwe sporen, en GT-nauwkeurigheiddie alleen wordt geïmplementeerd voor Dataset-uitvoeringen.
Gegevenssets instellen
Maak een gegevensset die u kunt gebruiken als opslagplaats voor testcases. Hier kunt u testgevallen opslaan met de invoervraag en het ideale verwachte antwoord. Om de dataset te maken, zijn er drie keuzes: maak één record tegelijk aan, upload een CSV met zoekopdrachten en verwachte antwoorden, of, heel handig, voeg invoer en uitvoer toe. rechtstreeks vanuit de toepassingssporen wier antwoorden door menselijke experts als van goede kwaliteit worden beschouwd.
Hier is de dataset die ik voor de demo heb gemaakt. Dit is een combinatie van technische, facturerings- of ‘Beide’-query’s, en ik heb alle records gemaakt op basis van applicatietraceringen:
Dat is het! De configuratie is voltooid en we zijn klaar om observatie uit te voeren.
Waarneembaarheidsresultaten
De Langfuse-startpagina is een dashboard met verschillende handige grafieken. Het toont in één oogopslag het aantal uitvoeringssporen, scores en gemiddelden, sporen op tijd, modelgebruik en kosten enz.
MELT-gegevens
De nuttigste waarneembaarheidsgegevens zijn beschikbaar in de optie ‘Tracing’, die samengevatte en gedetailleerde weergaven van alle uitvoeringen weergeeft. Hier is een weergave van het dashboard met de tijd, naam, invoer, uitvoer en de cruciale latentie- en tokengebruiksstatistieken. Houd er rekening mee dat er voor elke agentuitvoering van onze toepassing twee evaluatietraceringen worden gegenereerd voor de Beknoptheid En Relevantie beoordelaars die wij hebben opgezet.
Laten we eens kijken naar de details van een van de uitvoeringen van de Klantenservice-applicatie. In het linkerpaneel wordt de agentstroom zowel in de vorm van een boom als in een stroomdiagram weergegeven. Het toont de LangGraph-knooppunten (agenten) en de LLM-oproepen samen met het tokengebruik. Als onze agenten tooloproepen of human-in-the-loop-stappen hadden gehad, zouden ze hier ook zijn afgebeeld. Merk op dat de evaluatie scoort voor Beknoptheid En Relevantie staan ook bovenaan afgebeeld, deze zijn voor deze run respectievelijk 0,40 en 1. Als u erop klikt, wordt de reden voor de score weergegeven en een link die ons naar het evaluatortraject brengt.
Aan de rechterkant kunnen we voor elke agent, LLM en tooloproep de invoer en de gegenereerde uitvoer zien. Hier zien we bijvoorbeeld dat de zoekopdracht is gecategoriseerd als ‘Beide’ en dat daarom in het linkerdiagram wordt weergegeven dat zowel de technische als de factureringsondersteuningsmedewerkers zijn gebeld, wat bevestigt dat onze stroom werkt zoals verwacht.
Bovenaan het rechterpaneel bevindt zich de ‘Toevoegen aan datasets’ knop. Wanneer u op deze knop klikt, wordt bij elke stap in de boom een paneel geopend zoals hieronder afgebeeld, waar u de invoer en uitvoer van die stap rechtstreeks kunt toevoegen aan een testgegevensset die in de vorige sectie is gemaakt. Dit is een handige functie voor menselijke experts om veel voorkomende gebruikersquery’s en goede reacties aan de dataset toe te voegen tijdens normale agentbewerkingen, waardoor met minimale inspanning een regressietestrepository kan worden opgebouwd. Wanneer er in de toekomst een grote upgrade of release van de applicatie plaatsvindt, kan de Regressie-dataset worden uitgevoerd en kunnen de gegenereerde outputs worden gescoord aan de hand van de verwachte outputs (grondwaarheid) die hier zijn vastgelegd met behulp van de ‘GT-nauwkeurigheid‘-evaluator die we hebben gemaakt tijdens de LLM-als-rechter-opstelling. Dit helpt om LLM-drift (of agent-drift) vroegtijdig te detecteren en corrigerende maatregelen te nemen.
Hier is een van de evaluatiesporen (Beknoptheid) voor deze toepassingstrace. De evaluator motiveert de score van 0,4 die volgens hem dit antwoord was.
Scores
De optie Scores in Langfuse toont een lijst met alle evaluatieruns van de verschillende actieve beoordelaars, samen met hun scores. Relevanter is het Analytics-dashboard, waar twee scores kunnen worden geselecteerd en statistieken zoals gemiddelde en standaardafwijking samen met trendlijnen kunnen worden bekeken.
Regressie testen
Met Datasets zijn we klaar om regressietests uit te voeren met behulp van de testcase-repository met query’s en verwachte outputs. We hebben vier query’s opgeslagen in onze Regressie-dataset, met een combinatie van technische, facturerings- en ‘Beide’-query’s.
Hiervoor kunnen we de bijgevoegde code uitvoeren die de relevante dataset ophaalt en het experiment uitvoert. Alle testruns worden geregistreerd, samen met de gemiddelde scores. Het resultaat van een geselecteerde test kunnen wij bekijken met Beknoptheid, GT-nauwkeurigheid en relevantie scores voor elke testcase in één dashboard. En indien nodig kan het gedetailleerde spoor worden geopend om de redenering voor de partituur te zien.
Je kunt de code hier bekijken.
from langfuse import get_client
from langfuse.openai import OpenAI
from langchain_openai import AzureChatOpenAI
from langfuse import Langfuse
import os
# Initialize client
from dotenv import load_dotenv
load_dotenv(override=True)
langfuse = Langfuse(
host="https://cloud.langfuse.com",
public_key=os.environ("LANGFUSE_PUBLIC_KEY") ,
secret_key=os.environ("LANGFUSE_SECRET_KEY")
)
llm = AzureChatOpenAI(
azure_deployment=os.environ.get("AZURE_OPENAI_DEPLOYMENT_NAME"),
api_version=os.environ.get("AZURE_OPENAI_API_VERSION", "2025-01-01-preview"),
temperature=0.2,
)
# Define your task function
def my_task(*, item, **kwargs):
question = item.input('ticket')
response = llm.invoke(({"role": "user", "content": question}))
raw = response.content.strip().lower()
return raw
# Get dataset from Langfuse
dataset = langfuse.get_dataset("Regression")
# Run experiment directly on the dataset
result = dataset.run_experiment(
name="Production Model Test",
description="Monthly evaluation of our production model",
task=my_task # see above for the task definition
)
# Use format method to display results
print(result.format())
Belangrijkste afhaalrestaurants
- De waarneembaarheid van AI hoeft niet veel code te vereisen.
De meeste evaluatie-, tracerings- en regressietestmogelijkheden voor LLM-agents kunnen worden ingeschakeld via configuratie in plaats van aangepaste code, waardoor de ontwikkelings- en onderhoudsinspanningen aanzienlijk worden verminderd. - Uitgebreide evaluatieworkflows kunnen declaratief worden gedefinieerd.
Mogelijkheden zoals LLM-as-a-Judge-scores (relevantie, beknoptheid, hallucinatie, nauwkeurigheid van de grondwaarheid), variabelentoewijzing en evaluatieprompts worden rechtstreeks in het observatieplatform geconfigureerd, zonder op maat gemaakte evaluatielogica te schrijven. - Datasets en regressietesten zijn configuratie-eerste functies.
Testcase-repository’s, dataset-runs en ground-truth-vergelijkingen kunnen worden opgezet en hergebruikt via de gebruikersinterface of een eenvoudige configuratie, waardoor teams regressietests kunnen uitvoeren over agentversies met minimale extra code. - Volledige MELT-waarneembaarheid komt ‘out of the box’.
Metrieken (latentie, tokengebruik, kosten), gebeurtenissen (LLM en tool-oproepen), logboeken en traceringen worden automatisch vastgelegd en gecorreleerd, waardoor de noodzaak van handmatige instrumentatie in de workflows van agenten wordt vermeden. - Minimale instrumentatie, maximale zichtbaarheid.
Met lichtgewicht SDK-integratie krijgen teams diepgaand inzicht in de uitvoeringspaden van meerdere agenten, evaluatieresultaten en prestatietrends, waardoor ontwikkelaars zich kunnen concentreren op agentlogica in plaats van op observatie-analyses.
Conclusie
Naarmate LLM-agenten complexer worden, Waarneembaarheid is niet langer optioneel. Zonder dit systeem veranderen multi-agentsystemen snel in zwarte dozen die moeilijk te evalueren, te debuggen en te verbeteren zijn.
Een AI-observatieplatform verschuift deze last weg van ontwikkelaars en applicatiecode. Met behulp van een minimale code, configuratie-eerste aanpakkunnen teams LLM-als-een-Judge-evaluatie, regressietesten en volledige MELT-observatie mogelijk maken zonder aangepaste pijplijnen te bouwen en te onderhouden. Dit vermindert niet alleen de technische inspanningen, maar versnelt ook het traject van prototype naar productie.
Door een open-source, framework-agnostisch platform zoals Langfuse te adopteren, krijgen teams een enige bron van waarheid voor de prestaties van agenten, waardoor AI-systemen gemakkelijker te vertrouwen, te ontwikkelen en op schaal te gebruiken zijn.
Meer weten? De hier gepresenteerde Customer Service-agenttoepassing volgt een manager-werknemer-architectuurpatroon niet werk in CrewAI. Lees hoe waarneembaarheid heeft me geholpen dit bekende probleem met het hiërarchische manager-werknemerproces van CrewAI op te lossen, door de reacties van agenten bij elke stap te traceren en deze te verfijnen om de orkestratie naar behoren te laten werken. Volledige analyse hier: Waarom de Manager-Worker-architectuur van CrewAI faalt – en hoe u dit kunt oplossen
Neem contact met mij op en deel uw opmerkingen op www.linkedin.com/in/partha-sarkar-lets-talk-AI
Alle afbeeldingen en gegevens die in dit artikel worden gebruikt, zijn synthetisch gegenereerd. Figuren en code door mij gemaakt
