steeds complexer wordt, werken traditionele houtkap en monitoring niet goed. Wat het team echt nodig heeft, is observatievermogen: de mogelijkheid om beslissingen van agenten te volgen, de responskwaliteit automatisch te evalueren en afwijkingen 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 observatie hanteren en zich tegelijkertijd concentreren op de kerntaak van het opbouwen en verbeteren van agentorkestratie. En integreer de applicatie in het observatieplatform met minimale overhead voor de functionele code. In dit artikel laat ik zien hoe je een open source AI-observatieplatform kunt opzetten om het volgende te doen met behulp van een minimale code-aanpak:
- LLM-als-rechter: Configureer vooraf gebouwde beoordelaars om reacties te beoordelen op waarheid, relevantie, hallucinatie en meer. Toon scores gedurende het hele proces met gedetailleerde logboeken en analyses.
- Testen op schaal: Bereid een dataset voor om regressietestgevallen op te slaan om de nauwkeurigheid te meten ten opzichte van de verwachte grondwaarheidsreacties. Detecteer proactief LLM- en agentafwijkingen.
- MELT-gegevens: Houd statistieken bij (latentie, tokengebruik, modeldrift), gebeurtenissen (API-aanroepen, LLM-aanroepen, toolgebruik), logboeken (gebruikersinteracties, tooluitvoering, besluitvorming door agenten) met gedetailleerde traceringen – allemaal zonder gedetailleerde telemetrie- en instrumentatiecode.
We zullen Langfuse gebruiken voor observaties. 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 de klantenservicetoepassing bijgevoegd. De applicatie ontvangt tickets van gebruikers, classificeert ze in Technisch, Facturering of Beide met behulp van de Triage-agent en stuurt ze vervolgens door naar een technische ondersteuningsagent, Billing Support-agent of beide. Vervolgens synthetiseert de finalizer-agent de antwoorden van beide agenten in een samenhangend, beter leesbaar formaat. Het stroomdiagram is als volgt:
Code 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 (hobbyniveaus met grote limieten zijn beschikbaar) en stelt u vervolgens een project in. In de projectinstellingen kunt u openbare en geheime sleutels maken die aan het begin van de code moeten worden opgegeven. U moet ook een LLM-verbinding toevoegen die zal worden gebruikt voor LLM-evaluatie als rechter.
Instelling van LLM als rechter
Dit is de essentie van het instellen van prestatie-evaluaties voor agenten. Hier kunt u verschillende ingebouwde Evaluators uit de Evaluator-bibliotheek configureren die de antwoorden beoordelen op basis van verschillende criteria, zoals Beknopt, Correct, Hallucinatie, Antwoordkritiek, enz. Dit zou voor de meeste gebruiksscenario’s voldoende moeten zijn, anders kan ook een Aangepaste Evaluator worden ingesteld. Zo ziet de Evaluator-bibliotheek eruit:
Selecteer de evaluator, bijvoorbeeld Relevantie, die u wilt gebruiken. U kunt ervoor kiezen om het uit te voeren voor nieuwe of bestaande traceringen, of om gegevenssets uit te voeren. Controleer bovendien de evaluatieopdracht om er zeker van te zijn dat deze aan uw evaluatiedoelen voldoet. Het allerbelangrijkste is dat query’s, builds en andere variabelen op de juiste manier moeten worden toegewezen aan hun bronnen (meestal aan de invoer en uitvoer van de applicatietracering). In ons geval zijn dit de ticketgegevens die door de gebruiker zijn ingevoerd en het antwoord dat door de finalisatieagent is gegenereerd. Als u een gegevensset wilt uitvoeren, kunt u bovendien het resulterende antwoord vergelijken met het Ground Truth-antwoord dat is opgeslagen als de verwachte uitvoer (uitgelegd in de volgende sectie).
Hier is de configuratie voor ‘GT-nauwkeurigheid‘ evaluatie Ik heb een nieuwe dataset voorbereid, samen met het in kaart brengen van variabelen. Ook wordt een korte preview van de evaluatie weergegeven. De meeste beoordelaars scoorden tussen 0 en 1:
Voor de klantenservicedemo heb ik 3 evaluatoren geconfigureerd: Relevantie, beknoptheid die loopt voor alle nieuwe sporen, en GT-nauwkeurigheidalleen geïmplementeerd voor het uitvoeren van gegevenssets.
Voorbereiding van de dataset
Maak een gegevensset die u kunt gebruiken als opslagplaats voor testcases. Hier kunt u testgevallen opslaan met invoerquery’s en de ideale verwachte antwoorden. Om een dataset aan te maken zijn er 3 opties: maak één record tegelijk aan, upload een CSV met de vraag en het verwachte antwoord, of voeg eenvoudigweg invoer en uitvoer toe rechtstreeks vanuit de applicatietracering wier antwoorden door menselijke experts als van goede kwaliteit werden beoordeeld.
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 aantekeningen gemaakt van de applicatietracering:
Alleen dat! 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 observatiegegevens zijn beschikbaar in de optie ‘Tracking’, die een samenvatting en gedetailleerd overzicht van alle uitvoeringen weergeeft. Hieronder ziet u een dashboardweergave met tijd, naam, invoer en uitvoer, evenals belangrijke latentie- en tokengebruiksstatistieken. Houd er rekening mee dat er voor elke agentuitvoering van onze toepassing twee evaluatietraceringen worden gegenereerd Beknoptheid En Relevantie beoordelaar die we hebben opgesteld.
Laten we eens kijken naar de details van een van de uitvoeringen van een klantenservicetoepassing. In het linkerpaneel wordt de agentstroom weergegeven als een boom en als een stroomdiagram. Het toont LangGraph-knooppunten (agenten) en LLM-oproepen samen met tokengebruik. Als onze agent tooloproepen of human-in-the-loop-stappen heeft, worden deze hier ook weergegeven. Merk op dat de evaluatiescore voor Beknoptheid En Relevantie ook hierboven weergegeven, namelijk respectievelijk 0,40 en 1 voor dit proces. Als u erop klikt, ziet u de reden voor de score en een link die ons naar het beoordelaarstraject brengt.
Aan de rechterkant kunnen we voor elke agent, LLM en tooloproep de resulterende invoer en uitvoer zien. Hier zien we bijvoorbeeld dat de zoekopdracht is gecategoriseerd als ‘Beide’ en daarom is in het linkerdiagram te zien dat zowel de technische ondersteuning als de factureringsagenten zijn gebeld, wat bevestigt dat onze stroom werkt zoals verwacht.
Bovenaan het rechterpaneel staat ‘Toevoegen aan dataset’ knop. Bij elke stap in de boomstructuur opent deze knop, wanneer erop wordt geklikt, een paneel zoals hieronder weergegeven, waar u de invoer en uitvoer van die stap rechtstreeks kunt toevoegen aan de testgegevensset die in de vorige sectie is gemaakt. Dit is een handige functie voor menselijke experts om veel voorkomende gebruikersquery’s en goede reacties op datasets toe te voegen tijdens normale agentoperaties, waardoor met minimale inspanning een regressietestrepository kan worden opgebouwd. In de toekomst, wanneer er een grote upgrade of release van de applicatie plaatsvindt, kan de Regressie-dataset worden uitgevoerd en kan de resulterende output worden beoordeeld aan de hand van de verwachte output (grondwaarheid) die hier is vastgelegd met behulp van ‘GT-nauwkeurigheid‘ beoordelaars die we tijdens de LLM-opstelling als juryleden hebben gecreëerd. Dit helpt LLM-afwijkingen (of agentafwijkingen) vroegtijdig te detecteren en corrigerende maatregelen te nemen.
Hier is een spoor van de evaluatie (Beknoptheid) voor sporen van deze toepassing. De evaluator motiveerde de score van 0,4 die hij als reactie beoordeelde.
Scoren
De optie Scores in Langfuse geeft een lijst weer met alle evaluaties van verschillende actieve beoordelaars, samen met hun scores. Relevanter is het Analytics-dashboard, waar twee scores kunnen worden geselecteerd en statistieken zoals gemiddelde en standaarddeviatie en trendlijnen kunnen worden bekeken.
Regressie testen
Met Datasets zijn we klaar om regressietests uit te voeren met behulp van een opslagplaats van querytestcases en verwachte output. 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 testritten worden geregistreerd, samen met de gemiddelde score. We kunnen de resultaten van de geselecteerde test zien met Kortom, nauwkeurigheid en relevantie van GT scores voor elke testcase in één dashboard. En indien nodig is er toegang tot gedetailleerde details om de reden voor de score te zien.
Je kunt de code hier zien.
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())
Belangrijke punten
- Voor AI-waarneembaarheid is niet veel code nodig.
De meeste evaluatie-, tracerings- en regressietestmogelijkheden voor de LLM-agent kunnen worden ingeschakeld via configuratie in plaats van via aangepaste code, waardoor de ontwikkelings- en onderhoudsinspanningen aanzienlijk worden verminderd. - Uitgebreide evaluatieworkflows kunnen declaratief worden gedefinieerd.
Vaardigheden zoals LLM-beoordeling als rechter (relevantie, beknoptheid, hallucinatie, nauwkeurigheid van de grondwaarheid), variabelentoewijzing en evaluatieopdrachten worden rechtstreeks in het observatieplatform geconfigureerd, zonder op maat gemaakte evaluatielogica te schrijven. - Gegevensverzameling en regressietesten zijn functies waarbij de configuratie voorop staat.
Testcase-repository’s, uitvoeringen van datasets en vergelijkingen van de grondwaarheid kunnen worden georkestreerd en hergebruikt via een eenvoudige gebruikersinterface of configuratie, zodat teams met minimale extra code regressietests kunnen uitvoeren voor verschillende agentversies. - Volledige MELT-waarneembaarheid komt ‘out of the box’.
Metrieken (latentie, tokengebruik, kosten), gebeurtenissen (LLM en tooloproepen), logboeken en traceringen worden automatisch vastgelegd en gecorreleerd, waardoor de noodzaak van handmatige instrumentatie in de workflows van agenten wordt geëlimineerd. - Minimale instrumentatie, maximale zichtbaarheid.
Met lichtgewicht SDK-integratie krijgen teams diepgaand inzicht in de uitvoeringspijplijnen van meerdere agenten, evaluatieresultaten en prestatietrends, waardoor ontwikkelaars zich kunnen concentreren op agentlogica in plaats van op observatiepijplijnen.
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.
AI-observatieplatforms verschuiven deze last weg van ontwikkelaars en applicatiecode. Met behulp van een Minimale code, configuratie eerste benaderingteams kunnen LLM-evaluatie als rechter, regressietesten en volledige MELT-observatie mogelijk maken zonder aangepaste pijplijnen te bouwen en te onderhouden. Dit vermindert niet alleen de engineeringinspanning, maar versnelt ook het proces van prototype tot productie.
Door een open source, raamwerkvrij platform zoals Langfuse te adopteren, profiteren teams de 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 toepassing voor klantenserviceagenten volgt het architectuurpatroon van de manager en de werknemer Nee werkt bij CrewAI Lees hoe observatievermogen heeft me geholpen dit veelvoorkomende probleem met het manager-werknemer-hiërarchieproces in CrewAI op te lossen, door de reactie van de agent bij elke stap te volgen en deze zo af te stemmen dat de orkestratie werkt zoals bedoeld. Volledige analyse hier: Waarom de Worker-Manager-architectuur van CrewAI faalde – en hoe dit te verhelpen
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. Afbeeldingen en code die ik heb gemaakt
