Jenseits der Kontextüberlastung: Wie Parlant × Milvus dem Verhalten von LLM-Agenten Kontrolle und Klarheit verschafft

Stellen Sie sich vor, Sie sollen eine Aufgabe erledigen, die 200 Geschäftsregeln, 50 Tools und 30 Demos umfasst, und Sie haben dafür nur eine Stunde Zeit. Das ist einfach unmöglich. Dennoch erwarten wir oft, dass große Sprachmodelle genau das tun, wenn wir sie zu "Agenten" machen und sie mit Anweisungen überfrachten.

In der Praxis scheitert dieser Ansatz jedoch schnell. Herkömmliche Agenten-Frameworks wie LangChain oder LlamaIndex injizieren alle Regeln und Werkzeuge auf einmal in den Kontext des Modells, was zu Regelkonflikten, Kontextüberlastung und unvorhersehbarem Verhalten in der Produktion führt.

Um dieses Problem zu lösen, hat ein Open-Source-Agenten-Framework namens Parlant kürzlich auf GitHub an Zugkraft gewonnen. Es führt einen neuen Ansatz namens Alignment Modeling ein, zusammen mit einem Überwachungsmechanismus und bedingten Übergängen, die das Agentenverhalten viel besser kontrollierbar und erklärbar machen.

In Verbindung mit Milvus, einer Open-Source-Vektordatenbank, wird Parlant noch leistungsfähiger. Milvus fügt semantische Intelligenz hinzu, die es den Agenten ermöglicht, dynamisch und in Echtzeit die relevantesten Regeln und Kontexte abzurufen - so bleiben sie präzise, effizient und produktionsbereit.

In diesem Beitrag erfahren Sie, wie Parlant im Verborgenen arbeitet und wie die Integration mit Milvus eine produktionsreife Lösung ermöglicht.

Warum traditionelle Agenten-Frameworks scheitern

Herkömmliche Agenten-Frameworks lieben es, groß aufzutreten: Hunderte von Regeln, Dutzende von Tools und eine Handvoll Demos - alles in einer einzigen, überladenen Eingabeaufforderung zusammengefasst. Das mag in einer Demo oder einem kleinen Sandkastentest gut aussehen, aber sobald man es in die Produktion überführt, zeigen sich schnell die Risse.

• Widersprüchliche Regeln bringen Chaos: Wenn zwei oder mehr Regeln gleichzeitig gelten, haben diese Frameworks keine eingebaute Möglichkeit zu entscheiden, welche von ihnen gewinnt. Manchmal wird eine davon ausgewählt. Manchmal werden beide gemischt. Manchmal wird etwas völlig Unvorhersehbares getan.

• Randfälle machen die Lücken deutlich: Sie können unmöglich alles vorhersagen, was ein Benutzer sagen könnte. Und wenn Ihr Modell auf etwas stößt, das nicht in seinen Trainingsdaten enthalten ist, gibt es nur allgemeine, unverbindliche Antworten.

• Die Fehlersuche ist mühsam und kostspielig: Wenn sich ein Agent nicht korrekt verhält, ist es fast unmöglich festzustellen, welche Regel das Problem verursacht hat. Da sich alles in einer riesigen System-Eingabeaufforderung befindet, besteht die einzige Möglichkeit, das Problem zu beheben, darin, die Eingabeaufforderung neu zu schreiben und alles von Grund auf neu zu testen.

Was ist Parlant und wie funktioniert es?

Parlant ist eine quelloffene Ausrichtungsmaschine für LLM-Agenten. Sie können genau steuern, wie sich ein Agent in verschiedenen Szenarien verhält, indem Sie seinen Entscheidungsprozess auf strukturierte, regelbasierte Weise modellieren.

Um die Probleme traditioneller Agenten-Frameworks zu lösen, führt Parlant einen neuen, leistungsstarken Ansatz ein: Alignment Modeling. Die Kernidee besteht darin, die Regeldefinition von der Regelausführung zu trennen, um sicherzustellen, dass nur die relevantesten Regeln zu einem bestimmten Zeitpunkt in den LLM-Kontext eingespeist werden.

Granulare Richtlinien: Der Kern des Alignment Modeling

Das Herzstück des Alignment-Modells von Parlant ist das Konzept der Granularen Richtlinien. Anstatt einen riesigen Systemprompt voller Regeln zu schreiben, definieren Sie kleine, modulare Richtlinien, die jeweils beschreiben, wie der Agent mit einer bestimmten Art von Situation umgehen sollte.

Jede Leitlinie besteht aus drei Teilen:

• Bedingung - Eine Beschreibung in natürlicher Sprache, wann die Regel gelten soll. Parlant wandelt diese Bedingung in einen semantischen Vektor um und gleicht sie mit den Eingaben des Benutzers ab, um herauszufinden, ob sie relevant ist.

• Aktion - Eine klare Anweisung, die definiert, wie der Agent reagieren soll, wenn die Bedingung erfüllt ist. Diese Aktion wird nur bei Auslösung in den LLM-Kontext injiziert.

• Tools - Alle externen Funktionen oder APIs, die mit dieser spezifischen Regel verbunden sind. Diese werden dem Agenten nur zugänglich gemacht, wenn die Richtlinie aktiv ist, so dass die Verwendung der Tools kontrolliert und kontextabhängig bleibt.

await agent.create_guideline(
    condition="The user asks about a refund and the order amount exceeds 500 RMB",
    action="First call the order status check tool to confirm whether the refund conditions are met, then provide a detailed explanation of the refund process",
    tools=[check_order_status, calculate_refund_amount]
)

Jedes Mal, wenn ein Benutzer mit dem Agenten interagiert, führt Parlant einen leichtgewichtigen Abgleichschritt durch, um die drei bis fünf relevantesten Richtlinien zu finden. Nur diese Regeln werden in den Kontext des Modells eingefügt, so dass die Eingabeaufforderungen kurz und konzentriert bleiben und gleichzeitig sichergestellt wird, dass der Agent stets die richtigen Regeln befolgt.

Überwachungsmechanismen für Genauigkeit und Konsistenz

Um die Genauigkeit und Konsistenz weiter zu gewährleisten, führt Parlant einen Überwachungsmechanismus ein, der als zweite Ebene der Qualitätskontrolle fungiert. Der Prozess läuft in drei Schritten ab:

1. Generierung eines Antwortkandidaten - Der Agent erstellt eine erste Antwort basierend auf den übereinstimmenden Richtlinien und dem aktuellen Gesprächskontext.

2. Überprüfung der Übereinstimmung - Die Antwort wird mit den aktiven Richtlinien verglichen, um zu überprüfen, ob alle Anweisungen korrekt befolgt wurden.

3. Überarbeitung oder Bestätigung - Werden Probleme festgestellt, korrigiert das System die Ausgabe; ist alles in Ordnung, wird die Antwort genehmigt und an den Benutzer gesendet.

Dieser Überwachungsmechanismus stellt sicher, dass der Agent die Regeln nicht nur versteht, sondern sie auch tatsächlich einhält, bevor er antwortet - das verbessert sowohl die Zuverlässigkeit als auch die Kontrolle.

Bedingte Übergänge für Kontrolle und Sicherheit

In traditionellen Agenten-Frameworks ist jedes verfügbare Werkzeug jederzeit für den LLM zugänglich. Dieser "alles auf den Tisch"-Ansatz führt oft zu überladenen Aufforderungen und unbeabsichtigten Tool-Aufrufen. Parlant löst dieses Problem durch bedingte Übergänge. Ähnlich wie bei Zustandsautomaten wird eine Aktion oder ein Werkzeug nur dann ausgelöst, wenn eine bestimmte Bedingung erfüllt ist. Jedes Werkzeug ist eng an die entsprechende Richtlinie gebunden und wird nur verfügbar, wenn die Bedingung der Richtlinie aktiviert ist.

# The balance inquiry tool is exposed only when the condition "the user wants to make a transfer" is met
await agent.create_guideline(
    condition="The user wants to make a transfer",
    action="First check the account balance. If the balance is below 500 RMB, remind the user that an overdraft fee may apply.",
    tools=[get_user_account_balance]
)

Durch diesen Mechanismus wird der Aufruf eines Werkzeugs zu einem bedingten Übergang - Werkzeuge werden nur dann von "inaktiv" zu "aktiv", wenn ihre Auslösebedingungen erfüllt sind. Durch diese Strukturierung der Ausführung stellt Parlant sicher, dass jede Aktion bewusst und kontextabhängig erfolgt, um Missbrauch zu verhindern und gleichzeitig die Effizienz und Systemsicherheit zu verbessern.

Wie Milvus Parlant antreibt

Wenn wir einen Blick unter die Haube des Parlant-Richtlinienabgleichs werfen, wird eine zentrale technische Herausforderung deutlich: Wie kann das System die drei bis fünf relevantesten Regeln aus hunderten - oder sogar tausenden - Optionen in nur wenigen Millisekunden finden? Genau hier kommt eine Vektordatenbank ins Spiel. Das semantische Retrieval macht dies möglich.

Wie Milvus den Richtlinienabgleich von Parlant unterstützt

Der Richtlinienabgleich funktioniert über semantische Ähnlichkeit. Das Feld "Bedingung" jeder Leitlinie wird in eine Vektoreinbettung umgewandelt, die die Bedeutung der Leitlinie und nicht nur den wörtlichen Text erfasst. Wenn ein Benutzer eine Nachricht sendet, vergleicht Parlant die Semantik dieser Nachricht mit allen gespeicherten Richtlinieneinbettungen, um die relevantesten zu finden.

So funktioniert der Prozess Schritt für Schritt:

1. Kodierung der Abfrage - Die Nachricht des Benutzers und der aktuelle Gesprächsverlauf werden in einen Abfragevektor umgewandelt.

2. Suche nach Ähnlichkeit - Das System führt eine Ähnlichkeitssuche im Leitfadenvektor durch, um die nächstliegenden Übereinstimmungen zu finden.

3. Abrufen der Top-K-Ergebnisse - Die drei bis fünf semantisch relevantesten Richtlinien werden zurückgegeben.

4. Einfügen in den Kontext - Diese übereinstimmenden Leitlinien werden dann dynamisch in den Kontext des LLM eingefügt, damit das Modell nach den richtigen Regeln handeln kann.

Um diesen Arbeitsablauf zu ermöglichen, muss die Vektordatenbank drei entscheidende Funktionen bieten: eine leistungsstarke ANN-Suche (Approximate Nearest Neighbor), eine flexible Filterung von Metadaten und Vektoraktualisierungen in Echtzeit. Milvus, die quelloffene, cloud-native Vektordatenbank, bietet in allen drei Bereichen eine Leistung auf Produktionsniveau.

Um zu verstehen, wie Milvus in realen Szenarien funktioniert, lassen Sie uns als Beispiel einen Finanzdienstleistungsagenten betrachten.

Nehmen wir an, das System definiert 800 Geschäftsrichtlinien, die Aufgaben wie Kontoabfragen, Fondsüberweisungen und Beratungen zu Vermögensverwaltungsprodukten abdecken. In diesem Fall fungiert Milvus als Speicher- und Abrufschicht für alle Richtliniendaten.

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType
import parlant.sdk as p
# Connect to Milvus
connections.connect(host=“localhost”, port=“19530”)
# Define the schema for the guideline collection
fields = [
FieldSchema(name=“guideline_id”, dtype=DataType.VARCHAR, max_length=100, is_primary=True),
FieldSchema(name=“condition_vector”, dtype=DataType.FLOAT_VECTOR, dim=768),
FieldSchema(name=“condition_text”, dtype=DataType.VARCHAR, max_length=1000),
FieldSchema(name=“action_text”, dtype=DataType.VARCHAR, max_length=2000),
FieldSchema(name=“priority”, dtype=DataType.INT64),
FieldSchema(name=“business_domain”, dtype=DataType.VARCHAR, max_length=50)
]
schema = CollectionSchema(fields=fields, description=“Agent Guidelines”)
guideline_collection = Collection(name=“agent_guidelines”, schema=schema)
# Create an HNSW index for high-performance retrieval
index_params = {
“index_type”: “HNSW”,
“metric_type”: “COSINE”,
“params”: {“M”: 16, “efConstruction”: 200}
}
guideline_collection.create_index(field_name=“condition_vector”, index_params=index_params)

Wenn nun ein Benutzer sagt: "Ich möchte 100.000 RMB auf das Konto meiner Mutter überweisen", sieht der Laufzeitablauf folgendermaßen aus:

1. Rektorisierung der Abfrage - Umwandlung der Benutzereingabe in einen 768-dimensionalen Vektor.

2. Hybrides Retrieval - Durchführung einer Vektorähnlichkeitssuche in Milvus mit Metadatenfilterung (z.B. business_domain="transfer").

3. Rangfolge der Ergebnisse - Rangfolge der in Frage kommenden Leitlinien auf der Grundlage von Ähnlichkeitsbewertungen in Kombination mit ihren Prioritätswerten.

4. Kontextinjektion - Injektion der Top-3 übereinstimmenden Leitlinien action_text in den Kontext des Parlant-Agenten.

In dieser Konfiguration liefert Milvus eine P99-Latenzzeit von unter 15 ms, selbst wenn die Richtlinienbibliothek auf 100.000 Einträge skaliert. Im Vergleich dazu führt die Verwendung einer herkömmlichen relationalen Datenbank mit Stichwortabgleich in der Regel zu einer Latenzzeit von über 200 ms und einer deutlich geringeren Treffergenauigkeit.

Wie Milvus Langzeitgedächtnis und Personalisierung ermöglicht

Milvus bietet mehr als nur den Abgleich von Richtlinien. In Szenarien, in denen Agenten ein Langzeitgedächtnis und personalisierte Antworten benötigen, kann Milvus als Gedächtnisebene dienen, die vergangene Interaktionen von Benutzern als Vektoreinbettungen speichert und abruft und dem Agenten hilft, sich zu erinnern, was zuvor besprochen wurde.

# store user’s past interactions
user_memory_fields = [
    FieldSchema(name="interaction_id", dtype=DataType.VARCHAR, max_length=100, is_primary=True),
    FieldSchema(name="user_id", dtype=DataType.VARCHAR, max_length=50),
    FieldSchema(name="interaction_vector", dtype=DataType.FLOAT_VECTOR, dim=768),
    FieldSchema(name="interaction_summary", dtype=DataType.VARCHAR, max_length=500),
    FieldSchema(name="timestamp", dtype=DataType.INT64)
]
memory_collection = Collection(name="user_memory", schema=CollectionSchema(user_memory_fields))

Wenn derselbe Benutzer zurückkehrt, kann der Agent die relevantesten historischen Interaktionen von Milvus abrufen und sie verwenden, um eine vernetztere, menschenähnliche Erfahrung zu erzeugen. Wenn ein Nutzer beispielsweise letzte Woche nach einem Investmentfonds gefragt hat, kann sich der Agent an diesen Kontext erinnern und proaktiv reagieren: "Willkommen zurück! Haben Sie noch Fragen zu dem Fonds, den wir beim letzten Mal besprochen haben?"

Wie man die Leistung von Milvus-gestützten Agentensystemen optimiert

Wenn ein von Milvus betriebenes Agentensystem in einer Produktionsumgebung eingesetzt wird, ist die Leistungsoptimierung von entscheidender Bedeutung. Um niedrige Latenzzeiten und einen hohen Durchsatz zu erreichen, müssen mehrere Schlüsselparameter beachtet werden:

1. Auswahl des richtigen Index-Typs

Es ist wichtig, die richtige Indexstruktur zu wählen. HNSW (Hierarchical Navigable Small World) ist beispielsweise ideal für Szenarien mit hohem Abrufaufkommen, wie z. B. im Finanz- oder Gesundheitswesen, wo es auf Genauigkeit ankommt. IVF_FLAT eignet sich besser für groß angelegte Anwendungen wie E-Commerce-Empfehlungen, bei denen eine etwas geringere Wiederauffindbarkeit im Gegenzug für eine schnellere Leistung und einen geringeren Speicherbedarf akzeptabel ist.

2. Sharding-Strategie

Wenn die Anzahl der gespeicherten Richtlinien eine Million Einträge übersteigt, empfiehlt sich eine Partitionierung, um die Daten nach Geschäftsbereichen oder Anwendungsfällen aufzuteilen. Durch die Partitionierung wird der Suchraum pro Abfrage verkleinert, wodurch sich die Abrufgeschwindigkeit erhöht und die Latenzzeit auch bei wachsendem Datenbestand stabil bleibt.

3. Cache-Konfiguration

Für Richtlinien, auf die häufig zugegriffen wird, wie z. B. Standard-Kundenabfragen oder Workflows mit hohem Datenverkehr, können Sie das Caching der Milvus-Abfrageergebnisse nutzen. Dadurch kann das System frühere Ergebnisse wiederverwenden und die Latenzzeit bei wiederholten Abfragen auf unter 5 Millisekunden senken.

Hands-on-Demo: Wie man ein intelligentes Q&A-System mit Parlant und Milvus Lite aufbaut

Milvus Lite ist eine leichtgewichtige Version von Milvus - eine Python-Bibliothek, die einfach in Ihre Anwendungen eingebettet werden kann. Sie ist ideal für schnelles Prototyping in Umgebungen wie Jupyter Notebooks oder für die Ausführung auf Edge- und Smart-Geräten mit begrenzten Rechenressourcen. Trotz seines geringen Platzbedarfs unterstützt Milvus Lite die gleichen APIs wie andere Milvus-Implementierungen. Das bedeutet, dass der clientseitige Code, den Sie für Milvus Lite schreiben, später nahtlos mit einer vollständigen Milvus- oder Zilliz-Cloud-Instanz verbunden werden kann - ohne Refactoring.

In dieser Demo verwenden wir Milvus Lite in Verbindung mit Parlant, um zu demonstrieren, wie man ein intelligentes Q&A-System aufbaut, das schnelle, kontextbezogene Antworten bei minimaler Einrichtung liefert.

Voraussetzungen：

1) Parlant GitHub: https://github.com/emcie-co/parlant

2.die Parlant-Dokumentation: https://parlant.io/docs

3.python3.10+

4.OpenAI_key

5.MlivusLite

Schritt 1: Abhängigkeiten installieren

# Install required Python packages
pip install pymilvus parlant openai
# Or, if you’re using a Conda environment:
conda activate your_env_name
pip install pymilvus parlant openai

Schritt 2: Umgebungsvariablen konfigurieren

# Set your OpenAI API key
export OPENAI_API_KEY="your_openai_api_key_here"
# Verify that the variable is set correctly
echo $OPENAI_API_KEY

Schritt 3: Implementieren des Kerncodes

• Erstellen Sie einen benutzerdefinierten OpenAI Embedder

class OpenAIEmbedder(p.Embedder):
    # Converts text into vector embeddings with built-in timeout and retry
    # Dimension: 1536 (text-embedding-3-small)
    # Timeout: 60 seconds; Retries: up to 2 times

• Initialisieren der Wissensbasis

1. eine Milvus-Sammlung mit dem Namen kb_articles erstellen.

2. fügen Sie Beispieldaten ein (z.B. Rückerstattungsrichtlinien, Umtauschrichtlinien, Versanddauer).

3. einen HNSW-Index erstellen, um das Auffinden zu beschleunigen.

• Erstellen Sie das Vektorsuchwerkzeug

@p.tool
async def vector_search(query: str, top_k: int = 5, min_score: float = 0.35):
    # 1. Convert user query into a vector
    # 2. Perform similarity search in Milvus
    # 3. Return results with relevance above threshold

• Konfigurieren Sie den Parlant Agent

Leitlinie 1: Bei sachlichen oder richtlinienbezogenen Fragen muss der Agent zunächst eine Vektorsuche durchführen.

Leitlinie 2: Wenn Beweise gefunden werden, muss der Agent anhand einer strukturierten Vorlage antworten (Zusammenfassung + Schlüsselpunkte + Quellen).

# Guideline 1: Run vector search for factual or policy-related questions
await agent.create_guideline(
            condition="User asks a factual question about policy, refund, exchange, or shipping",
            action=(
                "Call vector_search with the user's query. "
                "If evidence is found, synthesize an answer by quoting key sentences and cite doc_id/title. "
                "If evidence is insufficient, ask a clarifying question before answering."
            ),
            tools=[vector_search],
# Guideline 2: Use a standardized, structured response when evidence is available
await agent.create_guideline(
condition=“Evidence is available”,
action=(
“Answer with the following template:\n”
“Summary: provide a concise conclusion.\n”
“Key points: 2-3 bullets distilled from evidence.\n”
“Sources: list doc_id and title.\n”
“Note: if confidence is low, state limitations and ask for clarification.”
),
tools=[],
)
tools=[],

)

• Schreiben Sie den vollständigen Code

import os
import asyncio
import json
from typing import List, Dict, Any
import parlant.sdk as p
from pymilvus import MilvusClient, DataType
# 1) Environment variables: using OpenAI (as both the default generation model and embedding service)
# Make sure the OPENAI_API_KEY is set
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY")
if not OPENAI_API_KEY:
    raise RuntimeError("Please set OPENAI_API_KEY environment variable")
# 2) Initialize Milvus Lite (runs locally, no standalone service required)
# MilvusClient runs in Lite mode using a local file path (requires pymilvus >= 2.x)
client = MilvusClient("./milvus_demo.db")  # Lite mode uses a local file path
COLLECTION = "kb_articles"
# 3) Example data: three policy or FAQ entries (in practice, you can load and chunk data from files)
DOCS = [
    {"doc_id": "POLICY-001", "title": "Refund Policy", "chunk": "Refunds are available within 30 days of purchase if the product is unused."},
    {"doc_id": "POLICY-002", "title": "Exchange Policy", "chunk": "Exchanges are permitted within 15 days; original packaging required."},
    {"doc_id": "FAQ-101", "title": "Shipping Time", "chunk": "Standard shipping usually takes 3–5 business days within the country."},
]
# 4) Generate embeddings using OpenAI (you can replace this with another embedding service)
# Here we use Parlant’s built-in OpenAI embedder for simplicity, but you could also call the OpenAI SDK directly.
class OpenAIEmbedder(p.Embedder):
    async def embed(self, texts: List[str], hints: Dict[str, Any] = {}) -> p.EmbeddingResult:
        # Generate text embeddings using the OpenAI API, with timeout and retry handling
        import openai
        try:
            client = openai.AsyncOpenAI(
                api_key=OPENAI_API_KEY,
                timeout=60.0,  # 60-second timeout
                max_retries=2  # Retry up to 2 times
            )
            print(f"Generating embeddings for {len(texts)} texts...")
            response = await client.embeddings.create(
                model="text-embedding-3-small",
                input=texts
            )
            vectors = [data.embedding for data in response.data]
            print(f"Successfully generated {len(vectors)} embeddings.")
            return p.EmbeddingResult(vectors=vectors)
        except Exception as e:
            print(f"OpenAI API call failed: {e}")
            # Return mock vectors for testing Milvus connectivity
            print("Using mock vectors for testing...")
            import random
            vectors = [[random.random() for _ in range(1536)] for _ in texts]
            return p.EmbeddingResult(vectors=vectors)
    @property
    def id(self) -> str:
        return "text-embedding-3-small"
    @property
    def max_tokens(self) -> int:
        return 8192
    @property
    def tokenizer(self) -> p.EstimatingTokenizer:
        from parlant.core.nlp.tokenization import ZeroEstimatingTokenizer
        return ZeroEstimatingTokenizer()
    @property
    def dimensions(self) -> int:
        return 1536
embedder = OpenAIEmbedder()
async def ensure_collection_and_load():
    # Create the collection (schema: primary key, vector field, additional fields)
    if not client.has_collection(COLLECTION):
        client.create_collection(
            collection_name=COLLECTION,
            dimension=len((await embedder.embed(["dimension_probe"])).vectors[0]),
            # Default metric: COSINE (can be changed with metric_type="COSINE")
            auto_id=True,
        )
        # Create an index to speed up retrieval (HNSW used here as an example)
        client.create_index(
            collection_name=COLLECTION,
            field_name="vector",
            index_type="HNSW",
            metric_type="COSINE",
            params={"M": 32, "efConstruction": 200}
        )
    # Insert data (skip if already exists; simple idempotent logic for the demo)
    # Generate embeddings
    chunks = [d["chunk"] for d in DOCS]
    embedding_result = await embedder.embed(chunks)
    vectors = embedding_result.vectors
    # Check if the same doc_id already exists; this is for demo purposes only — real applications should use stricter deduplication
    # Here we insert directly. In production, use an upsert operation or an explicit primary key
    client.insert(
        COLLECTION,
        data=[
            {"vector": vectors[i], "doc_id": DOCS[i]["doc_id"], "title": DOCS[i]["title"], "chunk": DOCS[i]["chunk"]}
            for i in range(len(DOCS))
        ],
    )
    # Load into memory
    client.load_collection(COLLECTION)
# 5) Define the vector search tool (Parlant Tool)
@p.tool
async def vector_search(context: p.ToolContext, query: str, top_k: int = 5, min_score: float = 0.35) -> p.ToolResult:
    # 5.1 Generate the query vector
    embed_res = await embedder.embed([query])
    qvec = embed_res.vectors[0]
    # 5.2 Search Milvus
    results = client.search(
        collection_name=COLLECTION,
        data=[qvec],
        limit=top_k,
        output_fields=["doc_id", "title", "chunk"],
        search_params={"metric_type": "COSINE", "params": {"ef": 128}},
    )
    # 5.3 Assemble structured evidence and filter by score threshold
    hits = []
    for hit in results[0]:
        score = hit["distance"] if "distance" in hit else hit.get("score", 0.0)
        if score >= min_score:
            hits.append({
                "doc_id": hit["entity"]["doc_id"],
                "title": hit["entity"]["title"],
                "chunk": hit["entity"]["chunk"],
                "score": float(score),
            })
    return p.ToolResult({"evidence": hits})
# 6) Run Parlant Server and create the Agent + Guidelines
async def main():
    await ensure_collection_and_load()
    async with p.Server() as server:
        agent = await server.create_agent(
            name="Policy Assistant",
            description="Rule-controlled RAG assistant with Milvus Lite",
        )
        # Example variable: current time (can be used in templates or logs)
        @p.tool
        async def get_datetime(context: p.ToolContext) -> p.ToolResult:
            from datetime import datetime
            return p.ToolResult({"now": datetime.now().isoformat()})
        await agent.create_variable(name="current-datetime", tool=get_datetime)
        # Core Guideline 1: Run vector search for factual or policy-related questions
        await agent.create_guideline(
            condition="User asks a factual question about policy, refund, exchange, or shipping",
            action=(
                "Call vector_search with the user's query. "
                "If evidence is found, synthesize an answer by quoting key sentences and cite doc_id/title. "
                "If evidence is insufficient, ask a clarifying question before answering."
            ),
            tools=[vector_search],
        )
        # Core Guideline 2: Use a standardized, structured response when evidence is available
        await agent.create_guideline(
            condition="Evidence is available",
            action=(
                "Answer with the following template:\\n"
                "Summary: provide a concise conclusion.\\n"
                "Key points: 2-3 bullets distilled from evidence.\\n"
                "Sources: list doc_id and title.\\n"
                "Note: if confidence is low, state limitations and ask for clarification."
            ),
            tools=[],
        )
        # Hint: Local Playground URL
        print("Playground: <http://localhost:8800>")
if __name__ == "__main__":
    asyncio.run(main())

Schritt 4: Ausführen des Codes

# Run the main program
python main.py

• Besuchen Sie den Playground:

<http://localhost:8800>

Sie haben nun erfolgreich ein intelligentes Q&A-System mit Parlant und Milvus erstellt.

Parlant vs. LangChain/LlamaIndex: Wie sie sich unterscheiden und wie sie zusammenarbeiten

Wie unterscheidet sich Parlant von bestehenden Agenten-Frameworks wie LangChain oder LlamaIndex?

LangChain und LlamaIndex sind Allzweck-Frameworks. Sie bieten eine breite Palette von Komponenten und Integrationen und sind daher ideal für Rapid Prototyping und Forschungsexperimente. Wenn es jedoch um den Einsatz in der Produktion geht, müssen Entwickler oft selbst zusätzliche Schichten aufbauen, wie z.B. Regelmanagement, Konformitätsprüfungen und Zuverlässigkeitsmechanismen, um die Agenten konsistent und vertrauenswürdig zu halten.

Parlant bietet ein integriertes Richtlinienmanagement, Selbstkritik-Mechanismen und Erklärungswerkzeuge, die Entwicklern helfen, das Verhalten, die Reaktionen und die Gründe eines Agenten zu verwalten. Dadurch eignet sich Parlant besonders für anspruchsvolle, kundenorientierte Anwendungsfälle, bei denen es auf Genauigkeit und Verantwortlichkeit ankommt, wie z. B. im Finanz-, Gesundheits- und Rechtswesen.

Diese Frameworks können sogar zusammenarbeiten:

• Verwenden Sie LangChain, um komplexe Datenverarbeitungspipelines oder Abrufworkflows zu erstellen.

• Verwenden Sie Parlant, um die letzte Interaktionsschicht zu verwalten und sicherzustellen, dass die Ausgaben den Geschäftsregeln entsprechen und interpretierbar bleiben.

• Verwenden Sie Milvus als Grundlage für die Vektordatenbank, um die semantische Suche, das Gedächtnis und den Wissensabruf im gesamten System in Echtzeit zu ermöglichen.

Schlussfolgerung

Wenn LLM-Agenten von der Erprobung in die Produktion übergehen, ist die Schlüsselfrage nicht mehr, was sie tun können, sondern wie zuverlässig und sicher sie es tun können. Parlant bietet die Struktur und Kontrolle für diese Zuverlässigkeit, während Milvus die skalierbare Vektorinfrastruktur bereitstellt, die alles schnell und kontextabhängig macht.

Zusammen ermöglichen sie es Entwicklern, KI-Agenten zu entwickeln, die nicht nur fähig, sondern auch vertrauenswürdig, erklärbar und produktionsreif sind.

🚀 Schauen Sie sich Parlant auf GitHub an und integrieren Sie es mit Milvus, um Ihr eigenes intelligentes, regelbasiertes Agentensystem zu erstellen.

Haben Sie Fragen oder möchten Sie eine Funktion genauer kennenlernen? Treten Sie unserem Discord-Kanal bei oder stellen Sie Fragen auf GitHub. Sie können auch eine 20-minütige Einzelsitzung buchen, um Einblicke, Anleitung und Antworten auf Ihre Fragen über die Milvus Office Hours zu erhalten.