Milvus
Zilliz
  • Home
  • Blog

  • 超越情境超載：Parlant × Milvus 如何為 LLM 代理行為帶來控制與清晰度

超越情境超載：Parlant × Milvus 如何為 LLM 代理行為帶來控制與清晰度

  • Tutorials
November 05, 2025
Min Yin

想像一下，您被要求完成一項涉及 200 個業務規則、50 個工具和 30 個示範的任務，而您只有一小時的時間來完成。這簡直是不可能的事。然而，當我們將大型語言模型變成「代理」，並為它們加載過多的指令時，我們往往期望它們能夠完全做到這一點。

實際上，這種方法很快就會瓦解。傳統的代理框架，例如 LangChain 或 LlamaIndex，會一次將所有規則和工具注入模型的上下文，這會導致規則衝突、上下文過載，以及在生產中出現不可預測的行為。

為了解決這個問題，一個名為 Parlant的開放原始碼代理框架最近在 GitHub 上逐漸受到矚目。它引進了稱為 Alignment Modeling 的新方法，以及監督機制和條件轉換，讓代理程式的行為更容易控制和解釋。

若搭配開放原始碼向量資料庫Milvus，Parlant 的功能將更加強大。Milvus 增加了語意智慧，讓代理能即時動態擷取最相關的規則與情境，保持準確、有效率，並為生產做好準備。

在這篇文章中，我們將探討 Parlant 如何在遮罩下運作，以及如何將其與 Milvus 整合以達到生產級的效果。

為何傳統的代理程式框架會分崩離析

傳統的代理程式框架喜歡大而全：數百個規則、數十個工具、數個示範，全都塞進一個過度膨脹的單一提示中。這在示範或小型沙箱測試中可能看起來很棒，但一旦將它推到生產中，裂縫就會迅速顯現。

  • 衝突的規則帶來混亂：當兩個或更多的規則同時適用時，這些框架沒有內建的方法來決定哪一個獲勝。有時它會選擇其中一個。有時它會混合兩者。有時它會做一些完全無法預測的事情。

  • 邊緣案例暴露缺口：您不可能預測使用者可能說的每一句話。當您的模型遇到訓練資料以外的情況時，它就會預設為一般、不具承諾性的答案。

  • 調試既痛苦又昂貴：當代理程式發生不當行為時，幾乎不可能找出是哪個規則造成了問題。由於所有東西都在一個巨大的系統提示中，唯一的修復方法就是重寫提示，然後從頭重新測試所有東西。

Parlant 是什麼？

Parlant 是 LLM 代理程式的開放原始碼 Alignment Engine。透過以結構化、基於規則的方式建模代理程式的決策過程，您可以精確控制代理程式在不同情境下的行為。

為了解決傳統代理程式框架中發現的問題，Parlant 引進了一種新的強大方法：對齊建模 (Alignment Modeling)。其核心思想是將規則定義與規則執行分開，確保在任何特定時間，只有最相關的規則才會注入 LLM 的上下文。

詳細指引：對齊建模的核心

Parlant 對齊模型的核心是「細粒指引」（Granular Guidelines）的概念。您可以定義小的、模組化的指導方針，而不是撰寫一個充滿規則的巨型系統提示，每個指導方針都會說明代理程式應該如何處理特定類型的情況。

每個指引都由三個部分組成：

  • 條件- 自然語言描述，說明規則在何時適用。Parlant 將此條件轉換為語意向量，並將其與使用者的輸入進行比對，以判斷是否相關。

  • 動作 (Action) - 定義代理程式在符合條件時應如何回應的明確指令。這個動作只有在觸發時才會注入 LLM 的上下文。

  • 工具- 與特定規則相關的任何外部功能或 API。只有當指導方針啟動時，代理程式才會接觸到這些工具，以保持工具使用的可控性和情境感知。

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]
)

每次使用者與代理程式互動時，Parlant 都會執行一個輕量級的匹配步驟，找出三到五個最相關的準則。只有這些規則會被注入模型的上下文中，以保持提示的簡潔與重點，同時確保代理程式持續遵循正確的規則。

準確性與一致性的監督機制

為了進一步保持準確性和一致性，Parlant 引入了監督機制，作為第二層品質控制。該流程分三個步驟進行：

1.生成候選回應- 代理程式會根據匹配的指引和當前的對話內容創建初始回覆。

2.檢查是否符合規定- 回覆會與作用中的指引進行比較，以驗證是否正確遵循了每項指示。

3.修改或確認- 如果發現任何問題，系統會修正輸出；如果一切都沒問題，回覆就會被核准並傳送給使用者。

這種監督機制可確保代理不僅瞭解規則，而且在回覆之前實際遵守規則，從而提高可靠性和控制力。

控制與安全的條件轉換

在傳統的代理程式架構中，每個可用的工具都會隨時暴露給 LLM。這種「一切盡在掌握」的方式經常會導致超載提示和非預期的工具呼叫。Parlant 透過條件轉換來解決這個問題。類似於狀態機運作的方式，動作或工具只有在符合特定條件時才會被觸發。每個工具都與相對應的準則緊密結合，只有當該準則的條件被啟動時，它才會變得可用。

# 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]
)

此機制將工具調用轉換為條件轉換-工具只有在滿足其觸發條件時，才會從 「非作用中 」轉換為 「作用中」。透過這樣的執行結構，Parlant 可確保每個動作都是經過深思熟慮且符合情境的，在提高效率與系統安全性的同時，也能防止誤用。

Milvus 如何為 Parlant 提供動力

當我們探究 Parlant 的指引匹配流程時，一個核心技術挑戰變得很明顯：系統如何能在幾毫秒的時間內，從數百個甚至上千個選項中找出三到五個最相關的規則？這正是向量資料庫的用武之地。語意檢索讓這一切成為可能。

Milvus如何支持Parlant的指南匹配流程

指南匹配是通過語義相似性進行的。每個指南的 Condition 欄位都會轉換成向量嵌入，以捕捉其意義，而不僅僅是其文字。當使用者傳送訊息時，Parlant 會將該訊息的語意與所有儲存的指引嵌入進行比較，以找出最相關的指引。

以下是整個過程的步驟：

1.編碼查詢- 將使用者的訊息和最近的對話記錄轉換成查詢向量。

2.搜尋相似性- 系統在指引向量儲存庫中執行相似性搜尋，找出最接近的匹配項目。

3.擷取 Top-K 結果- 會傳回語意最相關的前三到五個指引。

4.插入上下文- 這些匹配的指引會動態插入 LLM 的上下文，以便模型可以根據正確的規則行事。

為了讓這個工作流程成為可能，向量資料庫必須提供三項關鍵功能：高效能的近似近鄰 (ANN) 搜尋、彈性的元資料篩選，以及即時向量更新。Milvus 是開放原始碼、雲端原生向量資料庫，可在這三個領域提供生產級的效能。

要瞭解 Milvus 在實際情境中的運作方式，讓我們以金融服務代理商為例。

假設系統定義了 800 個業務指引，涵蓋帳戶查詢、資金轉帳和財富管理產品諮詢等任務。在此設定中，Milvus 充當所有指引資料的儲存和檢索層。

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)

現在，當用戶說 「我想向我母親的帳戶轉帳 100,000 元人民幣 」時，運行流程為

1.Rectorize the query- 將使用者輸入轉換成 768 維向量。

2.混合檢索- 在 Milvus 中執行向量相似性搜尋，並進行元資料篩選 (例如：business_domain="transfer")。

3.結果排序- 依據相似性分數結合優先順序值，對候選指南進行排序。

4.情境注入- 將前三個符合的指南action_text 注入 Parlant 代理的情境中。

在此設定中，即使指引庫擴充至 100,000 個項目，Milvus 的 P99 延遲仍低於 15 毫秒。相較之下，使用傳統關聯式資料庫與關鍵字比對，通常會導致超過 200 毫秒的延遲，以及明顯較低的比對準確度。

Milvus 如何實現長期記憶與個人化

Milvus 的功能不僅限於指引匹配。在代理需要長期記憶和個人化回應的情況下，Milvus 可作為記憶層，將使用者過去的互動以向量嵌入的方式儲存和檢索，幫助代理記住之前討論的內容。

# 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))

當同一位使用者回來時，代理程式可以從 Milvus 擷取最相關的歷史互動，並利用這些互動來產生更連結的、類似人類的體驗。舉例來說，如果使用者上週詢問投資基金的相關資訊，代理程式就能回想當時的情境並主動回應："歡迎回來！您對我們上次討論的基金還有問題嗎？

如何優化由 Milvus 驅動的代理系統的性能

在生產環境中部署由 Milvus 驅動的代理系統時，性能調整變得非常重要。要實現低延遲和高吞吐量，需要注意幾個關鍵參數：

1.選擇正確的索引類型

選擇適當的索引結構非常重要。舉例來說，HNSW (Hierarchical Navigable Small World) 適合金融或醫療保健等高召回率的情境，因為在這些情境中，精確度是至關重要的。IVF_FLAT 更適合電子商務推薦等大型應用程式，在這些應用程式中，可以接受稍低的召回率，以換取更快的效能和更少的記憶體使用。

2.分片策略

當儲存的指引數量超過一百萬個項目時，建議使用Partition來按業務領域或用例分割資料。分區可減少每次查詢的搜尋空間，提高擷取速度，即使資料集成長，也能保持延遲穩定。

3.快取設定

對於頻繁存取的指引，例如標準客戶查詢或高流量工作流程，您可以使用 Milvus 查詢結果快取。這可讓系統重用先前的結果，將重複搜尋的延遲時間縮短至 5 毫秒以下。

實際操作示範：如何使用 Parlant 和 Milvus Lite 建立智慧型問答系統

Milvus Lite是 Milvus 的輕量級版本 - 可輕鬆嵌入應用程式的 Python 函式庫。它非常適合在 Jupyter Notebooks 等環境中快速建立原型，或在運算資源有限的邊緣與智慧型裝置上執行。儘管 Milvus Lite 的佔用空間較小，但它支援與其他 Milvus 部署相同的 API。這表示您為 Milvus Lite 寫的用戶端程式碼，日後可無縫連接至完整的 Milvus 或 Zilliz Cloud 實例 - 無須重構。

在這個示範中，我們將結合 Parlant 使用 Milvus Lite 來展示如何建立一個智慧型問答系統，以最少的設定提供快速、上下文感知的答案。

先決條件： 1.Parlant GitHub

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

2.Parlant 文件： https://parlant.io/docs

3.python3.10以上

4.OpenAI_key

5.MlivusLite

步驟 1: 安裝相依性

# 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

步驟 2：配置環境變數

# 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

步驟 3: 執行核心程式碼

  • 建立自訂的 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
  • 初始化知識庫

1.建立一個名為 kb_articles 的 Milvus 套件。

2.插入範例資料 (例如退款政策、換貨政策、出貨時間)。

3.建立 HNSW 索引以加速檢索。

  • 建立向量搜尋工具
@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
  • 設定 Parlant Agent

準則 1：對於事實或政策相關的問題，代理必須先執行向量搜尋。

準則 2：當找到證據時，代理必須使用結構化的範本 (摘要 + 重點 + 來源) 來回覆。

# 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=[],



)
  • 撰寫完整程式碼
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())

步驟 4：執行程式碼

# Run the main program
python main.py

  • 造訪 Playground：
<http://localhost:8800>

現在您已經成功地使用 Parlant 和 Milvus 建立了一個智慧型問答系統。

Parlant vs. LangChain/LlamaIndex：它們的差異與合作方式

與現有的代理框架如LangChain或LlamaIndex 相比，Parlant 有何不同？

LangChain 和 LlamaIndex 是通用框架。它們提供廣泛的元件與整合，非常適合快速原型設計與研究實驗。然而，當需要在生產中部署時，開發人員往往需要自行建立額外的層級，例如規則管理、合規性檢查和可靠性機制，以保持代理的一致性和可信度。

Parlant 提供內建的準則管理 (Guideline Management)、自我批評機制以及可解釋工具，協助開發人員管理代理程式的行為、回應與理由。這使得 Parlant 特別適用於高風險、面對客戶的使用個案，在這些使用個案中，準確性和責任非常重要，例如金融、醫療保健和法律服務。

事實上，這些框架可以一起運作：

  • 使用 LangChain 建立複雜的資料處理管道或檢索工作流程。

  • 使用 Parlant 管理最後的互動層，確保輸出遵循業務規則，並保持可解釋。

  • 使用 Milvus 作為向量資料庫的基礎，在整個系統中提供即時語意搜尋、記憶和知識檢索。

結論

當 LLM 代理從實驗走向生產時，關鍵問題不再是它們能做什麼，而是它們如何可靠、安全地做到這一點。Parlant 提供了結構與控制的可靠性，而 Milvus 則提供了可擴充的向量基礎架構，以保持一切快速且情境感知。

兩者相輔相成，讓開發人員可以建立不僅有能力，而且值得信賴、可解釋且可生產的 AI 代理。

在 GitHub 上查看 Parlant，並將其與 Milvus整合，建立您自己的智慧型規則驅動代理系統。

對任何功能有問題或想要深入瞭解？加入我們的 Discord 頻道或在 GitHub 上提出問題。您也可以透過 Milvus Office Hours 預約 20 分鐘的一對一會議，以獲得深入的瞭解、指導和問題解答。

    Try Managed Milvus for Free

    Zilliz Cloud is hassle-free, powered by Milvus and 10x faster.

    Get Started

    Like the article? Spread the word

    繼續閱讀