Da die Systeme immer komplexer werden, sind herkömmliche Protokollierung und Überwachung nicht mehr ausreichend. Was Groups tatsächlich brauchen, ist Beobachtbarkeit: die Möglichkeit, Agentenentscheidungen zu verfolgen, die Antwortqualität automatisch zu bewerten und Abweichungen im Laufe der Zeit zu erkennen – ohne große Mengen an benutzerdefiniertem Auswertungs- und Telemetriecode schreiben und pflegen zu müssen.
Daher müssen Groups die richtige Plattform für Observability einführen, während sie sich auf die Kernaufgabe konzentrieren, die Orchestrierung der Agenten aufzubauen und zu verbessern. Und integrieren Sie ihre Anwendung mit minimalem Overhead für ihren Funktionscode in die Observability-Plattform. In diesem Artikel werde ich zeigen, wie Sie eine Open-Supply-KI-Beobachtbarkeitsplattform einrichten können, um mithilfe eines Minimalcode-Ansatzes Folgendes auszuführen:
- LLM-als-Richter: Konfigurieren Sie vorgefertigte Evaluatoren, um Antworten auf Korrektheit, Relevanz, Halluzination und mehr zu bewerten. Zeigen Sie die Ergebnisse aller Läufe mit detaillierten Protokollen und Analysen an.
- Testen im großen Maßstab: Richten Sie Datensätze zum Speichern von Regressionstestfällen ein, um die Genauigkeit anhand der erwarteten Floor-Fact-Antworten zu messen. Erkennen Sie LLM und Agent-Drift proaktiv.
- MELT-Daten: Verfolgen Sie Metriken (Latenz, Token-Nutzung, Modelldrift), Ereignisse (API-Aufrufe, LLM-Aufrufe, Device-Nutzung), Protokolle (Benutzerinteraktion, Device-Ausführung, Agenten-Entscheidungsfindung) mit detaillierten Traces – alles ohne detaillierte Telemetrie und Instrumentierungscode.
Wir werden Langfuse zur Beobachtbarkeit verwenden. Es ist Open-Supply und Framework-unabhängig und kann mit gängigen Orchestrierungs-Frameworks und LLM-Anbietern zusammenarbeiten.
Multi-Agent-Anwendung
Für diese Demonstration habe ich den LangGraph-Code einer Kundendienstanwendung angehängt. Die Anwendung nimmt Tickets vom Benutzer entgegen, klassifiziert sie mithilfe eines Triage-Agenten in „Technisch“, „Abrechnung“ oder „Beide“ und leitet sie dann an den Mitarbeiter des Technischen Helps, den Agenten des Abrechnungssupports oder an beide weiter. Anschließend synthetisiert ein Finalizer-Agent die Antwort beider Agenten in ein kohärentes, besser lesbares Format. Das Flussdiagramm sieht wie folgt aus:
Der Code ist hier angehängt
# --------------------------------------------------
# 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 Consumer (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), # 🔑 permits token utilization
)
# --------------------------------------------------
# 4. Shared State
# --------------------------------------------------
class AgentState(TypedDict, whole=False):
ticket: str
class: 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",
title="triage_agent",
enter={"ticket": state("ticket")},
) as span:
span.update_trace(title="Buyer Service Question - LangGraph Demo")
response = llm.invoke((
{
"position": "system",
"content material": (
"Classify the question as considered one of: "
"Technical, Billing, Each. "
"Reply with solely the label."
),
},
{"position": "person", "content material": state("ticket")},
))
uncooked = response.content material.strip().decrease()
if "each" in uncooked:
class = "Each"
elif "technical" in uncooked:
class = "Technical"
elif "billing" in uncooked:
class = "Billing"
else:
class = "Technical" # ✅ secure fallback
span.replace(output={"uncooked": uncooked, "class": class})
return {"class": class}
def technical_support_agent(state: dict) -> dict:
with langfuse.start_as_current_observation(
as_type="span",
title="technical_support_agent",
enter={
"ticket": state("ticket"),
"class": state.get("class"),
},
) as span:
response = llm.invoke((
{
"position": "system",
"content material": (
"You're a technical assist specialist. "
"Present a transparent, step-by-step resolution."
),
},
{"position": "person", "content material": state("ticket")},
))
reply = response.content material
span.replace(output={"technical_response": reply})
return {"technical_response": reply}
def billing_support_agent(state: dict) -> dict:
with langfuse.start_as_current_observation(
as_type="span",
title="billing_support_agent",
enter={
"ticket": state("ticket"),
"class": state.get("class"),
},
) as span:
response = llm.invoke((
{
"position": "system",
"content material": (
"You're a billing assist specialist. "
"Reply clearly about funds, invoices, or accounts."
),
},
{"position": "person", "content material": state("ticket")},
))
reply = response.content material
span.replace(output={"billing_response": reply})
return {"billing_response": reply}
def finalizer_agent(state: dict) -> dict:
with langfuse.start_as_current_observation(
as_type="span",
title="finalizer_agent",
enter={
"ticket": state("ticket"),
"technical": state.get("technical_response"),
"billing": state.get("billing_response"),
},
) as span:
elements = (
f"Technical:n{state('technical_response')}"
for ok in ("technical_response")
if state.get(ok)
) + (
f"Billing:n{state('billing_response')}"
for ok in ("billing_response")
if state.get(ok)
)
if not elements:
ultimate = "Error: No agent responses accessible."
else:
response = llm.invoke((
{
"position": "system",
"content material": (
"Mix the next agent responses into ONE clear, skilled, "
"customer-facing reply. Don't point out brokers or inner labels. "
f"Reply the person's question: '{state('ticket')}'."
),
},
{"position": "person", "content material": "nn".be part of(elements)},
))
ultimate = response.content material
span.replace(output={"final_response": ultimate})
return {"final_response": ultimate}
# --------------------------------------------------
# 6. LangGraph Development
# --------------------------------------------------
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("class"),
{
"Technical": "technical",
"Billing": "billing",
"Each": "technical",
"__default__": "technical", # ✅ by no means dead-end
},
)
# Sequential decision
builder.add_conditional_edges(
"technical",
lambda state: state("class"),
{
"Each": "billing", # Proceed to billing if Each
"__default__": "finalizer",
},
)
builder.add_edge("billing", "finalizer")
builder.add_edge("finalizer", END)
graph = builder.compile()
# --------------------------------------------------
# 9. Most important
# --------------------------------------------------
if __name__ == "__main__":
print("===============================================")
print(" Conditional Multi-Agent Assist System (Prepared)")
print("===============================================")
print("Enter 'exit' or 'give up' to cease this system.n")
whereas True:
# Get person enter for the ticket
ticket = enter("Enter your assist question (ticket): ")
# Verify for exit command
if ticket.decrease() in ("exit", "give up"):
print("nExiting the assist system. Goodbye!")
break
if not ticket.strip():
print("Please enter a non-empty question.")
proceed
attempt:
# --- Run the graph with the person's ticket ---
end result = graph.invoke(
{"ticket": ticket},
config={"callbacks": (langfuse_callback)},
)
# --- Print Outcomes ---
class = end result.get('class', 'N/A')
print(f"n✅ Triage Classification: **{class}**")
# Verify which brokers have been executed based mostly on the presence of a response
executed_agents = ()
if end result.get("technical_response"):
executed_agents.append("Technical")
if end result.get("billing_response"):
executed_agents.append("Billing")
print(f"🛠️ Brokers Executed: {', '.be part of(executed_agents) if executed_agents else 'None (Triage Failed)'}")
print("n================ FINAL RESPONSE ================n")
print(end result("final_response"))
print("n" + "="*60 + "n")
besides Exception as e:
# That is vital for debugging: print the exception sort and message
print(f"nAn error occurred throughout processing ({sort(e).__name__}): {e}")
print("nPlease attempt one other question.")
print("n" + "="*60 + "n")
Observability-Konfiguration
Um Langfuse einzurichten, gehen Sie zu https://cloud.langfuse.com/und richten Sie ein Konto mit einer Abrechnungsstufe (Hobbystufe mit großzügigen Limits) ein und richten Sie dann ein Projekt ein. In den Projekteinstellungen können Sie die öffentlichen und geheimen Schlüssel generieren, die am Anfang des Codes angegeben werden müssen. Sie müssen außerdem die LLM-Verbindung hinzufügen, die für die LLM-Bewertung als Richter verwendet wird.
LLM-als-Richter-Setup
Dies ist der Kern des Leistungsbewertungs-Setups für Agenten. Hier können Sie verschiedene vorgefertigte Evaluatoren aus der Evaluator-Bibliothek konfigurieren, die die Antworten nach verschiedenen Kriterien wie Prägnanz, Korrektheit, Halluzination, Antwortkritik usw. bewerten. Diese sollten für die meisten Anwendungsfälle ausreichen, ansonsten können auch benutzerdefinierte Evaluatoren eingerichtet werden. Hier ist eine Ansicht der Evaluator-Bibliothek:
Wählen Sie den Evaluator, beispielsweise Relevanz, aus, den Sie verwenden möchten. Sie können es für neue oder vorhandene Traces oder für Dataset-Läufe ausführen. Überprüfen Sie außerdem die Bewertungsaufforderung, um sicherzustellen, dass sie Ihrem Bewertungsziel entspricht. Am wichtigsten ist, dass die Abfrage-, Generierungs- und anderen Variablen korrekt der Quelle zugeordnet werden (normalerweise der Eingabe und Ausgabe des Anwendungs-Hint). In unserem Fall sind dies die vom Benutzer eingegebenen Ticketdaten bzw. die vom Finalizer-Agenten generierte Antwort. Darüber hinaus können Sie bei Dataset-Läufen die generierten Antworten mit den Floor Fact-Antworten vergleichen, die als erwartete Ausgaben gespeichert sind (wird in den nächsten Abschnitten erläutert).
Hier ist die Konfiguration für das ‚GT-Genauigkeit‚ Auswertung, die ich für neue Datensatzläufe eingerichtet habe, zusammen mit der Variablenzuordnung. Die Vorschau der Bewertungsaufforderung wird ebenfalls angezeigt. Die meisten Bewerter erreichen eine Wertspanne von 0 bis 1:
Für die Kundenservice-Demo habe ich 3 Bewerter konfiguriert – Relevanz, Prägnanz die für alle neuen Spuren laufen, und GT-Genauigkeitdas nur für Dataset-Ausführungen bereitgestellt wird.
Einrichtung von Datensätzen
Erstellen Sie einen Datensatz zur Verwendung als Testfall-Repository. Hier können Sie Testfälle mit der Eingabeabfrage und der perfect erwarteten Antwort hinterlegen. Um den Datensatz zu erstellen, gibt es drei Möglichkeiten: Erstellen Sie jeweils einen Datensatz, laden Sie eine CSV-Datei mit Abfragen und erwarteten Antworten hoch oder fügen Sie ganz bequem Ein- und Ausgaben hinzu direkt aus den Anwendungsspuren deren Antworten von menschlichen Experten als qualitativ hochwertig beurteilt werden.
Hier ist der Datensatz, den ich für die Demo erstellt habe. Dabei handelt es sich um eine Mischung aus technischen, Abrechnungs- oder „beides“-Anfragen, und ich habe alle Datensätze aus Anwendungsablaufverfolgungen erstellt:
Das ist es! Die Konfiguration ist abgeschlossen und wir sind bereit, Observability auszuführen.
Beobachtbarkeitsergebnisse
Die Langfuse-Startseite ist ein Dashboard mit mehreren nützlichen Diagrammen. Es zeigt auf einen Blick die Anzahl der Ausführungsspuren, Ergebnisse und Durchschnittswerte, Spuren nach Zeit, Modellnutzung und Kosten usw.
MELT-Daten
Die nützlichsten Beobachtbarkeitsdaten sind in der Possibility „Tracing“ verfügbar, die zusammengefasste und detaillierte Ansichten aller Ausführungen anzeigt. Hier ist eine Ansicht des Dashboards mit Zeit, Identify, Eingabe, Ausgabe sowie den entscheidenden Latenz- und Token-Nutzungsmetriken. Beachten Sie, dass für jede Agentenausführung unserer Anwendung zwei Auswertungsspuren generiert werden Prägnanz Und Relevanz Gutachter, die wir eingerichtet haben.
Schauen wir uns die Particulars einer der Ausführungen der Kundendienstanwendung an. Im linken Bereich wird der Agentenfluss sowohl als Baum als auch als Flussdiagramm dargestellt. Es zeigt die LangGraph-Knoten (Agenten) und die LLM-Aufrufe zusammen mit der Token-Nutzung. Hätten unsere Agenten Device-Aufrufe oder Human-in-the-Loop-Schritte gehabt, wären diese auch hier dargestellt worden. Beachten Sie, dass die Bewertung für Folgendes punktet Prägnanz Und Relevanz Oben sind auch die Werte dargestellt, die für diesen Lauf 0,40 bzw. 1 betragen. Wenn Sie darauf klicken, wird der Grund für die Bewertung angezeigt und ein Hyperlink, der uns zum Evaluator-Hint führt.
Rechts sehen wir für jeden Agenten, LLM und Toolaufruf die Eingabe und die generierte Ausgabe. Hier sehen wir beispielsweise, dass die Abfrage als „Beide“ kategorisiert wurde. Daher wird im linken Diagramm angezeigt, dass sowohl der technische als auch der Abrechnungssupport angerufen wurden, was bestätigt, dass unser Ablauf wie erwartet funktioniert.
Oben auf der rechten Seite befindet sich das „Zu Datensätzen hinzufügen‘ Style. Wenn Sie auf diese Schaltfläche klicken, wird in jedem Schritt des Baums ein Bereich wie der unten abgebildete geöffnet, in dem Sie die Eingabe und Ausgabe dieses Schritts direkt zu einem im vorherigen Abschnitt erstellten Testdatensatz hinzufügen können. Dies ist eine nützliche Funktion für menschliche Experten, um während des normalen Agentenbetriebs häufig auftretende Benutzeranfragen und gute Antworten zum Datensatz hinzuzufügen und so mit minimalem Aufwand ein Regressionstest-Repository aufzubauen. Wenn es in Zukunft ein größeres Improve oder eine größere Veröffentlichung der Anwendung gibt, kann der Regressionsdatensatz ausgeführt werden und die generierten Ausgaben können mit den hier aufgezeichneten erwarteten Ausgaben (Floor Fact) verglichen werden, indem die Possibility „GT-Genauigkeit‚Gutachter, den wir während der LLM-as-a-a-a-judge-Einrichtung erstellt haben. Dies hilft, LLM-Drift (oder Agent-Drift) frühzeitig zu erkennen und Korrekturmaßnahmen einzuleiten.
Hier ist eine der Auswertungsspuren (Prägnanz) für diesen Anwendungstrace. Der Bewerter gibt die Begründung für die Bewertung dieser Antwort mit 0,4 an.
Partituren
Die Possibility „Bewertungen“ in Langfuse zeigt eine Liste aller Bewertungsläufe der verschiedenen aktiven Bewerter zusammen mit ihren Bewertungen an. Relevanter ist das Analytics-Dashboard, in dem zwei Scores ausgewählt und Metriken wie Mittelwert und Standardabweichung sowie Trendlinien angezeigt werden können.
Regressionstests
Mit Datasets sind wir bereit, Regressionstests unter Verwendung des Testfall-Repositorys mit Abfragen und erwarteten Ausgaben durchzuführen. Wir haben in unserem Regressionsdatensatz vier Abfragen gespeichert, mit einer Mischung aus technischen Abfragen, Abrechnungsabfragen und „Beiden“ Abfragen.
Dazu können wir den beigefügten Code ausführen, der den relevanten Datensatz abruft und das Experiment ausführt. Alle Testläufe werden zusammen mit den Durchschnittsergebnissen protokolliert. Das Ergebnis eines ausgewählten Checks können wir mit einsehen Prägnanz, GT-Genauigkeit und Relevanz Ergebnisse für jeden Testfall in einem Dashboard. Und bei Bedarf kann auf die detaillierte Ablaufverfolgung zugegriffen werden, um die Begründung für die Bewertung einzusehen.
Den Code können Sie hier einsehen.
from langfuse import get_client
from langfuse.openai import OpenAI
from langchain_openai import AzureChatOpenAI
from langfuse import Langfuse
import os
# Initialize consumer
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,
)
# Outline your activity operate
def my_task(*, merchandise, **kwargs):
query = merchandise.enter('ticket')
response = llm.invoke(({"position": "person", "content material": query}))
uncooked = response.content material.strip().decrease()
return uncooked
# Get dataset from Langfuse
dataset = langfuse.get_dataset("Regression")
# Run experiment instantly on the dataset
end result = dataset.run_experiment(
title="Manufacturing Mannequin Take a look at",
description="Month-to-month analysis of our manufacturing mannequin",
activity=my_task # see above for the duty definition
)
# Use format methodology to show outcomes
print(end result.format())
Wichtige Erkenntnisse
- KI-Beobachtbarkeit muss nicht codelastig sein.
Die meisten Evaluierungs-, Ablaufverfolgungs- und Regressionstestfunktionen für LLM-Agenten können durch Konfiguration statt durch benutzerdefinierten Code aktiviert werden, wodurch der Entwicklungs- und Wartungsaufwand erheblich reduziert wird. - Umfangreiche Auswertungsworkflows können deklarativ definiert werden.
Funktionen wie LLM-as-a-Decide-Bewertung (Relevanz, Prägnanz, Halluzination, Floor-Fact-Genauigkeit), Variablenzuordnung und Bewertungsaufforderungen werden direkt in der Observability-Plattform konfiguriert – ohne das Schreiben einer maßgeschneiderten Bewertungslogik. - Datensätze und Regressionstests sind Konfigurationsfunktionen.
Testfall-Repositorys, Datensatzläufe und Floor-Fact-Vergleiche können über die Benutzeroberfläche oder durch einfache Konfiguration eingerichtet und wiederverwendet werden, sodass Groups mit minimalem zusätzlichen Code Regressionstests über Agentenversionen hinweg ausführen können. - Die vollständige MELT-Beobachtbarkeit ist „out of the field“.
Metriken (Latenz, Token-Nutzung, Kosten), Ereignisse (LLM- und Device-Aufrufe), Protokolle und Traces werden automatisch erfasst und korreliert, sodass keine manuelle Instrumentierung über Agenten-Workflows hinweg erforderlich ist. - Minimale Instrumentierung, maximale Sicht.
Durch die einfache SDK-Integration erhalten Groups einen umfassenden Einblick in die Ausführungspfade, Bewertungsergebnisse und Leistungstrends mehrerer Agenten, sodass sich Entwickler auf die Agentenlogik statt auf Observability-Pläne konzentrieren können.
Abschluss
Da LLM-Agenten immer komplexer werden, Beobachtbarkeit ist nicht länger elective. Ohne sie verwandeln sich Multi-Agenten-Systeme schnell in Black Packing containers, die schwer zu bewerten, zu debuggen und zu verbessern sind.
Eine KI-Beobachtbarkeitsplattform entlastet Entwickler und Anwendungscode von dieser Belastung. Mit a Ansatz mit minimalem Code und Konfiguration zuerstkönnen Groups die LLM-as-a-Decide-Bewertung, Regressionstests und vollständige MELT-Beobachtbarkeit ermöglichen, ohne benutzerdefinierte Pipelines erstellen und warten zu müssen. Dies reduziert nicht nur den Engineering-Aufwand, sondern beschleunigt auch den Weg vom Prototyp zur Produktion.
Durch die Einführung einer Open-Supply-, Framework-unabhängigen Plattform wie Langfuse gewinnen Groups einen einzige Quelle der Wahrheit für die Agentenleistung – wodurch KI-Systeme leichter vertrauenswürdig, weiterentwickelt und im großen Maßstab betrieben werden können.
Möchten Sie mehr wissen? Die hier vorgestellte Kundenservice-Agentenanwendung folgt einem Supervisor-Mitarbeiter-Architekturmuster nicht Arbeit in CrewAI. Lesen Sie, wie Beobachtbarkeit hat mir geholfen, dieses bekannte Downside mit dem hierarchischen Supervisor-Mitarbeiter-Prozess von CrewAI zu beheben, indem ich die Antworten der Agenten bei jedem Schritt verfolgt und verfeinert habe, damit die Orchestrierung so funktioniert, wie sie sollte. Vollständige Analyse hier: Warum die Supervisor-Employee-Architektur von CrewAI scheitert – und wie man sie behebt
Vernetzen Sie sich mit mir und teilen Sie Ihre Kommentare unter www.linkedin.com/in/partha-sarkar-lets-talk-AI
Alle in diesem Artikel verwendeten Bilder und Daten werden synthetisch generiert. Von mir erstellte Abbildungen und Code