Home
Blog
컨텍스트 과부하를 넘어서: Parlant × Milvus가 LLM 에이전트 행동에 제어와 명확성을 제공하는 방법

컨텍스트 과부하를 넘어서: Parlant × Milvus가 LLM 에이전트 행동에 제어와 명확성을 제공하는 방법

Tutorials

November 05, 2025

Min Yin

200개의 비즈니스 규칙, 50개의 도구, 30개의 데모가 포함된 작업을 완료하라는 지시를 받았는데 시간이 한 시간밖에 없다고 상상해 보세요. 불가능에 가까운 일입니다. 하지만 우리는 종종 대규모 언어 모델을 '에이전트'로 전환하고 지침으로 과부하를 주면 정확히 그렇게 할 수 있을 것으로 기대합니다.

실제로 이러한 접근 방식은 금방 무너집니다. LangChain이나 LlamaIndex와 같은 기존의 에이전트 프레임워크는 모든 규칙과 도구를 모델의 컨텍스트에 한 번에 주입하기 때문에 규칙 충돌, 컨텍스트 과부하, 프로덕션에서 예측할 수 없는 동작으로 이어집니다.

이 문제를 해결하기 위해 최근 Parlant라는 오픈 소스 에이전트 프레임워크가 GitHub에서 주목을 받고 있습니다. 이 프레임워크는 에이전트 행동을 훨씬 더 제어 가능하고 설명할 수 있도록 하는 감독 메커니즘 및 조건부 전환과 함께 정렬 모델링이라는 새로운 접근 방식을 도입합니다.

오픈 소스 벡터 데이터베이스인 Milvus와 함께 사용하면 Parlant의 기능이 더욱 향상됩니다. Milvus는 시맨틱 인텔리전스를 추가하여 상담원이 가장 관련성이 높은 규칙과 컨텍스트를 실시간으로 동적으로 검색하여 정확하고 효율적이며 프로덕션에 바로 사용할 수 있도록 지원합니다.

이 게시물에서는 Parlant가 내부에서 어떻게 작동하는지, 그리고 이를 Milvus와 통합하여 프로덕션급을 구현하는 방법에 대해 살펴봅니다.

기존 에이전트 프레임워크가 실패하는 이유

기존 에이전트 프레임워크는 수백 개의 규칙, 수십 개의 툴, 몇 개의 데모 등 모든 것을 하나의 과대 포장된 프롬프트에 집어넣어 큰 것을 좋아합니다. 데모나 소규모 샌드박스 테스트에서는 훌륭해 보일 수 있지만 일단 프로덕션에 적용하면 금방 결함이 드러나기 시작합니다.

상충되는 규칙은 혼란을 가져옵니다: 두 개 이상의 규칙이 동시에 적용되는 경우, 이러한 프레임워크에는 어떤 규칙이 승리할지 결정하는 기본 제공 방법이 없습니다. 때로는 한 가지를 선택하기도 합니다. 때로는 두 가지를 혼합하기도 합니다. 때로는 전혀 예측할 수 없는 일을 하기도 합니다.
엣지 케이스는 격차를 드러냅니다: 사용자가 할 수 있는 모든 말을 예측할 수는 없습니다. 그리고 모델이 학습 데이터를 벗어난 상황에 부딪히면 기본적으로 일반적이고 확정적이지 않은 답변으로 돌아갑니다.
디버깅은 힘들고 비용이 많이 듭니다: 에이전트가 오작동을 하면 어떤 규칙이 문제를 일으켰는지 정확히 파악하는 것은 거의 불가능합니다. 모든 것이 하나의 거대한 시스템 프롬프트 안에 있기 때문에 문제를 해결할 수 있는 유일한 방법은 프롬프트를 다시 작성하고 모든 것을 처음부터 다시 테스트하는 것입니다.

Parlant란 무엇이며 어떻게 작동하나요?

Parlant는 LLM 에이전트를 위한 오픈 소스 정렬 엔진입니다. 구조화된 규칙 기반 방식으로 의사 결정 프로세스를 모델링하여 다양한 시나리오에서 에이전트의 작동 방식을 정확하게 제어할 수 있습니다.

기존 에이전트 프레임워크에서 발견되는 문제를 해결하기 위해 Parlant는 새롭고 강력한 접근 방식을 도입했습니다: 바로 정렬 모델링입니다. 이 접근법의 핵심은 규칙 정의와 규칙 실행을 분리하여 주어진 시간에 가장 관련성이 높은 규칙만 LLM의 컨텍스트에 주입되도록 하는 것입니다.

세분화된 가이드라인: 정렬 모델링의 핵심

Parlant의 정렬 모델의 핵심은 세분화된 지침이라는 개념입니다. 규칙으로 가득 찬 하나의 거대한 시스템 프롬프트를 작성하는 대신 상담원이 특정 유형의 상황을 어떻게 처리해야 하는지를 설명하는 작은 모듈식 가이드라인을 정의합니다.

각 가이드라인은 세 부분으로 구성됩니다:

조건 - 규칙이 언제 적용되어야 하는지에 대한 자연어 설명입니다. Parlant는 이 조건을 시맨틱 벡터로 변환하고 사용자의 입력과 일치시켜 관련성이 있는지 파악합니다.
액션 - 조건이 충족되면 상담원이 어떻게 응답해야 하는지를 정의하는 명확한 지침입니다. 이 액션은 트리거될 때만 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는 가벼운 매칭 단계를 실행하여 가장 관련성이 높은 3~5개의 가이드라인을 찾습니다. 이러한 규칙만 모델의 컨텍스트에 주입되어 프롬프트가 간결하고 집중력을 유지하면서 에이전트가 일관되게 올바른 규칙을 따르도록 합니다.

정확성과 일관성을 위한 감독 메커니즘

정확성과 일관성을 더욱 유지하기 위해 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의 가이드라인 매칭 프로세스의 내부를 살펴보면 한 가지 핵심적인 기술적 과제가 명확해집니다. 시스템이 수백, 수천 개의 옵션 중에서 어떻게 단 몇 밀리초 만에 가장 관련성이 높은 3~5개의 규칙을 찾을 수 있을까요? 바로 여기에 벡터 데이터베이스가 등장합니다. 시맨틱 검색이 이를 가능하게 합니다.

Milvus가 Parlant의 가이드 라인 매칭 프로세스를 지원하는 방법

가이드라인 매칭은 시맨틱 유사성을 통해 작동합니다. 각 가이드라인의 조건 필드는 벡터 임베딩으로 변환되어 문자 그대로의 텍스트가 아닌 그 의미를 포착합니다. 사용자가 메시지를 보내면 Parlant는 해당 메시지의 의미를 저장된 모든 가이드라인 임베딩과 비교하여 가장 관련성이 높은 임베딩을 찾습니다.

이 프로세스의 단계별 작동 방식은 다음과 같습니다:

1. 쿼리 인코딩 - 사용자의 메시지와 최근 대화 내역이 쿼리 벡터로 변환됩니다.

2. 유사성 검색 - 시스템이 가이드라인 벡터 저장소 내에서 유사성 검색을 수행하여 가장 가까운 일치 항목을 찾습니다.

3. Top-K 결과 검색 - 의미적으로 가장 연관성이 높은 상위 3~5개의 가이드라인이 반환됩니다.

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. 쿼리 정규화 - 사용자 입력을 768차원 벡터로 변환합니다.

2. 하이브리드 검색 - 메타데이터 필터링(예: business_domain="transfer")을 사용하여 Milvus에서 벡터 유사성 검색을 실행합니다.

3. 결과 순위 - 우선순위 값과 결합된 유사도 점수를 기준으로 후보 가이드라인의 순위를 매깁니다.

4. 컨텍스트 삽입 - 일치하는 상위 3개 가이드라인의 action_text 을 Parlant 에이전트의 컨텍스트에 삽입합니다.

이 구성에서 Milvus는 가이드라인 라이브러리가 100,000개의 항목으로 확장되는 경우에도 15ms 미만의 P99 지연 시간을 제공합니다. 이에 비해 키워드 매칭에 기존의 관계형 데이터베이스를 사용하면 일반적으로 지연 시간이 200ms를 초과하고 매칭 정확도가 현저히 낮아집니다.

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(계층적 탐색이 가능한 작은 세계)가 이상적입니다. 전자 상거래 추천과 같은 대규모 애플리케이션의 경우, 빠른 성능과 메모리 사용량을 줄이는 대신 약간 낮은 리콜을 허용하는 IVF_FLAT이 더 효과적입니다.

2. 샤딩 전략

저장된 가이드라인의 항목 수가 100만 개를 초과하는 경우 파티션을 사용하여 데이터를 비즈니스 도메인 또는 사용 사례별로 분할하는 것이 좋습니다. 파티셔닝은 쿼리당 검색 공간을 줄여 검색 속도를 개선하고 데이터 세트가 증가하더라도 지연 시간을 안정적으로 유지합니다.

3. 캐시 구성

표준 고객 쿼리 또는 트래픽이 많은 워크플로우와 같이 자주 액세스하는 가이드라인의 경우 Milvus 쿼리 결과 캐싱을 사용할 수 있습니다. 이렇게 하면 시스템이 이전 결과를 재사용하여 반복 검색 시 대기 시간을 5밀리초 미만으로 단축할 수 있습니다.

실습 데모: Parlant와 Milvus Lite로 스마트 Q&A 시스템을 구축하는 방법

MilvusLite는 애플리케이션에 쉽게 임베드할 수 있는 Python 라이브러리인 Milvus의 경량 버전입니다. Jupyter 노트북과 같은 환경에서 빠르게 프로토타이핑하거나 컴퓨팅 리소스가 제한된 엣지 및 스마트 기기에서 실행하는 데 이상적입니다. 설치 공간은 작지만 Milvus Lite는 다른 Milvus 배포와 동일한 API를 지원합니다. 즉, Milvus Lite용으로 작성한 클라이언트 측 코드는 나중에 리팩토링 없이도 전체 Milvus 또는 Zilliz Cloud 인스턴스에 원활하게 연결할 수 있습니다.

이 데모에서는 최소한의 설정으로 상황에 맞는 빠른 답변을 제공하는 지능형 Q&A 시스템을 구축하는 방법을 보여드리기 위해 Milvus Lite를 Parlant와 함께 사용합니다.

전제 조건

1.Parlant 깃허브: 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 임베더 생성

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 에이전트 구성

가이드라인 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

플레이그라운드를 방문합니다:

<http://localhost:8800>

이제 Parlant와 Milvus를 사용하여 지능형 Q&A 시스템을 성공적으로 구축했습니다.

Parlant 대 LangChain/LlamaIndex: 차이점과 함께 작동하는 방법

LangChain이나 LlamaIndex와 같은 기존 에이전트 프레임워크와 비교했을 때 Parlant는 어떻게 다른가요?

LangChain과 LlamaIndex는 범용 프레임워크입니다. 이들은 광범위한 구성 요소와 통합을 제공하므로 신속한 프로토타이핑과 연구 실험에 이상적입니다. 그러나 프로덕션에 배포할 때는 에이전트의 일관성과 신뢰성을 유지하기 위해 개발자가 직접 규칙 관리, 규정 준수 검사, 안정성 메커니즘과 같은 추가 계층을 구축해야 하는 경우가 많습니다.

Parlant는 기본으로 제공되는 가이드라인 관리, 자체 비판 메커니즘, 설명 가능성 도구를 통해 개발자가 에이전트의 행동, 응답 및 이유를 관리할 수 있도록 도와줍니다. 따라서 금융, 의료, 법률 서비스 등 정확성과 책임감이 중요한 고객 대면 사용 사례에 특히 적합합니다.

실제로 이러한 프레임워크는 함께 작동할 수 있습니다:

복잡한 데이터 처리 파이프라인이나 검색 워크플로우를 구축하려면 LangChain을 사용하세요.
Parlant를 사용해 최종 상호 작용 계층을 관리하고, 출력이 비즈니스 규칙을 따르고 해석 가능한 상태를 유지할 수 있도록 하세요.
Milvus를 벡터 데이터베이스 기반으로 사용해 시스템 전체에 실시간 시맨틱 검색, 메모리 및 지식 검색을 제공하세요.

결론

LLM 에이전트가 실험에서 생산 단계로 넘어가면서 더 이상 무엇을 할 수 있는지가 아니라 얼마나 안정적이고 안전하게 수행할 수 있는지가 핵심 질문이 되었습니다. Parlant는 이러한 안정성을 위한 구조와 제어 기능을 제공하며, Milvus는 모든 것을 빠르고 컨텍스트 인식적으로 유지하는 확장 가능한 벡터 인프라를 제공합니다.

이 두 가지를 함께 사용하면 개발자는 성능뿐 아니라 신뢰할 수 있고 설명이 가능하며 프로덕션에 바로 사용할 수 있는 AI 에이전트를 구축할 수 있습니다.

GitHub에서 Parlant를 확인하고 Milvus와 통합하여 나만의 지능형 규칙 기반 에이전트 시스템을 구축하세요.

궁금한 점이 있거나 기능에 대해 자세히 알아보고 싶으신가요? Discord 채널에 참여하거나 GitHub에 이슈를 제출하세요. 또한 Milvus 오피스 아워를 통해 20분간 일대일 세션을 예약하여 인사이트, 안내 및 질문에 대한 답변을 얻을 수도 있습니다.