대화 검색

ConversationSearch는 모델에게 search_conversation_history 도구를 주고, StepPersistence capability가 이미 영속한 히스토리를 BM25로 랭킹해요. 컴팩션이 라이브 컨텍스트에서 버린 이전 턴들과, 기본적으로 같은 대화의 과거 실행까지 포괄하죠.

Source

Pydantic AI Harness가 0.x 릴리스인 동안 API는 마이너 릴리스 사이에 바뀔 수 있어요. 바뀔 때는 폐기 경고와 릴리스 노트 마이그레이션 안내가 정확히 어떻게 업그레이드할지 알려줘요. 버전 정책 참고.

출처: 문서

본문

문제 (The problem)

컴팩션 capability(SlidingWindowCompaction, SummarizingCompaction, ...)는 라이브 히스토리를 좁혀 컨텍스트 윈도우에 맞게 해요. SummarizingCompaction은 그 편집을 영속해요. 프리픽스가 요약으로 교체되면 원본은 다음 턴의 message_history에서 사라져요. 모델은 더 이상 이전에 말한 정확한 파일 경로·결정·값을 기억하지 못해요. 요약의 의역만 기억하죠. 그리고 지난 실행의 어떤 것도 접근 불가능해요. 아무리 잘 영속돼 있어도요.

해결책 (The solution)

ConversationSearch는 스스로 아무것도 영속하지 않아요. HistorySource를 통해 영속 capability가 이미 저장하는 무엇이든 읽고, search_conversation_history 도구 하나를 노출해 그 히스토리를 BM25로 랭킹해서 모델이 정확한 세부사항을 요청 시 컨텍스트로 다시 끌어오게 해요.

동봉된 소스 SnapshotHistorySource는 StepPersistence가 쓰는 스냅샷을 읽어요. 두 capability를 공유 스토어 인스턴스에 짝지으면, 추가 쓰기 경로·순서 제약·훅 조정 없이 리콜이 동작해요. 검색은 기본적으로 대화로 범위가 한정되므로, 말뭉치를 공유해야 하는 모든 실행에 같은 conversation_id를 넘기세요:

from pydantic_ai import Agent
from pydantic_ai_harness import ConversationSearch, SlidingWindowCompaction, StepPersistence
from pydantic_ai_harness.conversation_search import SnapshotHistorySource
from pydantic_ai_harness.step_persistence import SqliteStepStore

store = SqliteStepStore(database='sessions.db')
agent = Agent(
    'openai:gpt-5',
    capabilities=[
        StepPersistence(store=store),
        ConversationSearch(SnapshotHistorySource(store), scope='conversation'),
        SlidingWindowCompaction(max_messages=40),
    ],
)


async def ask(question: str, conversation_id: str) -> str:
    result = await agent.run(question, conversation_id=conversation_id)
    return result.output
  • 랭킹은 BM25(Lucene/Elasticsearch 뒤에 있는 알고리즘)로, 순수 파이썬으로 구현되어 새 의존성이 없어요. 희귀 용어와 정확 일치가 더 높게 점수화되고, 다중 단어 쿼리는 각 단어를 독립적으로 점수화해요.
  • 과거 실행에 도달하려면 두 실행이 conversation_id를 공유해야 해요. pydantic-ai가 실행당 하나를 해석해요. 명시적 conversation_id=가 이기고, 아니면 message_history의 가장 최근 conversation_id를 상속하고, 아니면 새 것을 생성해요. 그러므로 후속 실행을 통해 message_history를 실어 나르면 그것들이 한 대화에 유지돼요. 명시적 id도 히스토리 체인도 공유하지 않는 실행은 분리됩니다.
  • 결과는 근원(provenance)(run: ... | conversation: ...)을 담고, 도구의 선택적 run_id 인자는 검색을 한 실행으로 범위 한정해요. 그래서 다른 곳에서 참조된 실행(예: 컴팩션 영수증의 트랜스크립트 핸들이 가리키는 실행)은 직접 해석 가능해요.
  • 검색은 호출 시점에 스토어를 지연(lazily) 읽어서, 지금까지 영속된 모든 것, 현재 실행의 이전 스텝을 포함해 항상 봐요.

복구가 동작하는 방식 (How recovery works)

StepPersistence는 모든 스텝 경계에서 전체 히스토리 스냅샷을 저장해요. 그 편집을 영속하는 컴팩션 전략(SummarizingCompaction 같은)은 그 편집을 이후 스냅샷에 담아요. 하지만 같은 실행의 이전 스냅샷은 원본이 아직 라이브일 때 찍혔어요. SnapshotHistorySource는 각 실행의 스냅샷을 쓰기 순서로 합치고, 파생된 요약 아티팩트를 건너뛰며, 누적된 히스토리 접미사와 각 스냅샷 프리픽스 사이의 겹침을 제거해요. 이렇게 해서 컴팩션이 건드리지 않은 모든 것과 함께 원본을 복구해요. 반복된 메시지는 뚜렷한 순서 위치에 보존하면서요 — 스토어가 그 전-컴팩션 스냅샷을 여전히 보유하는 한. complete 스냅샷만 기여해요. interrupted 캡처(해결되지 않은 도구 작업, 합성 도구 반환)는 스토어의 기본 읽기 게이트에서 제외됩니다.

겹침 매칭은 객체 정체성이 아니라 각 직렬화된 메시지의 콘텐츠 해시를 기준으로 해요. 연속 스냅샷이 같은 커지는 히스토리를 다시 직렬화하고, 영속 실행기(Temporal, DBOS)가 스텝 사이에 메시지를 재인스턴스화하기 때문이에요.

HistorySource는 의도적으로 기판 중립적이에요("실행 나열, 각 실행의 영속 메시지 레코드 산출"). 추가 전용 엔트리 로그를 유지하는 영속 기판은 재생으로 직접 구현할 수 있고, 스냅샷-합치 어댑터를 검색 계층을 건드리지 않고 교체해요.

범위 (Scope)

scope='conversation'은 말뭉치를 호출 실행의 conversation_id와 일치하는 실행으로 제한해요. 에이전트를 실행할 때 인증된 테넌트 범위 값을 conversation_id로 넘기세요:

agent = Agent(
    'openai:gpt-5',
    capabilities=[
        StepPersistence(store=store),
        ConversationSearch(SnapshotHistorySource(store), scope='conversation'),
    ],
)


async def ask(question: str, user_id: str) -> str:
    result = await agent.run(question, conversation_id=user_id)
    return result.output

그러면 말뭉치는 호출 실행의 conversation_id와 일치하는 실행으로 제한되고, 도구 자신의 설명이 모델에게 그 제한이 적용된다고 알리며, 도구의 run_id 인자는 그것을 넘을 수 없어요. 범위 밖의 실행은 존재하지 않는 실행과 같은 "영속된 히스토리 없음" 답변을 보고해요.

pydantic-ai는 호출 실행의 conversation_id를 고정 순서로 해석해요. Agent.run(...)에 명시적 conversation_id= 인자, 그다음 message_history에 실린 가장 최근 conversation_id, 그다음 새 UUID7. conversation_id='new'를 넘기면 새 것을 강제해서, 제공된 히스토리에서 대화를 포크해요.

그 순서는 이 범위에 두 가지 결과를 낳아요. message_history를 실어 나르는 후속 실행은 이전 실행의 id를 상속하므로, 인자를 넘기지 않고도 뒤의 실행들을 검색할 수 있어요. 명시적 id도 히스토리 체인도 넘기지 않는 실행은 자기만의 id를 얻고 자기 자신에게만 닿아요. 인자 누락은 여기서 오류가 아니라 더 좁은 말뭉치일 뿐이에요. 어떤 동작도 멀티 테넌트 분리를 위해 믿을 격리 경계가 아니에요. 인증된 테넌트 범위 conversation_id=를 명시적으로 넘기세요(StepPersistence가 실행에 기록하는 값이에요).

conversation_id가 설정되지 않은 RunContext는 이 범위에서 아무것도 검색하지 않고 도구가 왜 그런지 말해요. "conversation id가 설정 안 됨"을 매칭하면 스토어의 모든 라벨 없는 실행을 한 말뭉치로 풀게 되는데, 이는 그 범위가 방지하려는 노출이에요. 그래서 닫힌 실패(fails closed) 대신.

범위는 HistorySource가 반환하는 RunRecord에 적용되므로, 커스텀 소스는 기본 범위가 어떤 것이든 매칭하려면 그것들에 conversation_id를 채워야 해요. 스토어가 이미 한 주체로 격리된 경우에만 scope='all'을 설정하세요. 이 선택 모드는 소스가 열거하는 모든 실행을 검색하고, 그 중 어떤 것에서든 원문 그대로 발췌를 반환할 수 있어요.

scope='all' 기본값에서 마이그레이션

이 기본값이 바뀌었어요. 이전 릴리스는 scope='all'이 기본이라 한 번의 검색이 스토어의 모든 실행을 랭킹했지만, 지금은 conversation이 기본이에요. 업그레이드는 아무것도 발생시키지 않아요. 옛 기본값을 믿던 호출자는 계속 동작하고 단지 다른 대화를 보지 못할 뿐이죠. 그래서 scope를 설정하지 않으면 capability 인스턴스당 한 번 HarnessDeprecationWarning을 발생해요.

옵션을 명시적으로 설정해 해결하세요. 두 값 모두 지원되며 어느 것도 폐기되지 않아요:

  • scope='all'은 이전의 스토어 전역 동작을 복원해요. 스토어가 한 주체의 히스토리를 담을 때 올바름.
  • scope='conversation'은 새 동작을 유지하고 경고를 조용히 해요.

나중에 마이그레이션하고 싶다면 모든 하네스 폐기 경고를 한 번에 조용히 하세요:

import warnings

from pydantic_ai_harness import HarnessDeprecationWarning

warnings.filterwarnings('ignore', category=HarnessDeprecationWarning)

핵심 옵션 (Key options)

옵션 기본 용도
source (필수) 말뭉치가 오는 곳. StepPersistence가 쓰는 스토어 위에 SnapshotHistorySource(store) 사용
scope 'conversation' (설정 안 하면 경고) 검색을 호출 실행의 conversation_id로 제한. 한 주체로 격리된 스토어에서만 'all' 사용. 설정 안 두면 한 번 경고; Scope 참고
max_matches 10 검색 도구가 반환하는 최대 매칭 발췌 수
context_lines 5 각 매칭 주변에 보이는 줄 수(그 매칭의 실행 안에서)
bm25_k1 1.5 BM25 용어 빈도 포화. 이 capability의 기본값; Lucene의 BM25Similarity는 1.2 사용
bm25_b 0.75 BM25 길이 정규화 (Lucene 기본값)
add_instructions True 리콜 도구가 존재한다고 모델에게 알리는 짧은 노트 방출
tool_id conversation-search 검색 도구의 toolset id

영속된 지시문 교체와 철회는 시스템 텍스트로 검색 가능해요. 표시된 발췌는 200자로 제한되고, 검색 인덱스는 전체 텍스트를 유지해요.

한계 (Limitations)

  • 검색은 영속된 것에만 닿아요. StepPersistence로 실행된 적 없는 실행에서 상속된 히스토리(예: 영속되지 않은 세션에서 넘어온 긴 message_history)는, 컴팩션이 첫 스냅샷 전에 버리면 복구할 수 없어요.
  • 컴팩션이 버린 원본의 복구는 전-컴팩션 스냅샷이 여전히 유지되는 것에 달려요. 경계 있는 스냅샷 보존을 가진 스토어(예: 실행당 스냅샷 상한)는 그 원본을 담은 초기 스냅샷을 잘라낼 수 있어요. 그러면 검색은 살아남은 스냅샷이 여전히 담는 것만 반환하며, 오류 대신 부분 결과로 열화돼요. 완전한 복구가 중요하면 실행에 대한 전체 스냅샷 히스토리를 유지하세요.
  • 말뭉치는 각 도구 호출에서 범위 내의 모든 실행 스냅샷을 읽어 재구성돼요. 스냅샷 저장은 누적적이라(각 스냅샷이 커지는 히스토리를 다시 직렬화), 큰 범위 내 히스토리는 각 검색을 비례적으로 비싸게 만들어요. 영속 인덱스(SQLite FTS5, #124에서 추적)가 확장 경로예요.
  • 스냅샷을 읽으면 외부화된 미디어(큰 바이너리 페이로드)가 복원되는데, 텍스트 인덱스는 그것을 절대 쓰지 않아요. 원격 미디어 백엔드가 있는 스토어는 검색마다 그 fetch 비용을 지불해요.

더 읽기 (Further reading)

API 참고 (API reference)

ConversationSearch

Bases: AbstractCapability[AgentDepsT]

의존성 없는 BM25 도구로 영속된 대화 히스토리를 검색해요.

이 capability는 스스로 아무것도 영속하지 않아요. HistorySource를 통해 영속 capability가 이미 저장하는 히스토리를 읽죠. StepPersistence와 같은 스토어를 공유하도록 짝지으면, 모델이 컴팩션이 라이브 컨텍스트에서 버린 것과 같은 대화의 과거 실행의 어느 것이든 리콜할 수 있어요:

from pydantic_ai import Agent
from pydantic_ai_harness.compaction import SlidingWindowCompaction
from pydantic_ai_harness.conversation_search import ConversationSearch, SnapshotHistorySource
from pydantic_ai_harness.step_persistence import SqliteStepStore, StepPersistence

store = SqliteStepStore(database='sessions.db')
agent = Agent(
    'openai:gpt-5',
    capabilities=[
        StepPersistence(store=store),
        ConversationSearch(SnapshotHistorySource(store), scope='conversation'),
        SlidingWindowCompaction(max_messages=40),
    ],
)


async def ask(question: str, conversation_id: str) -> str:
    result = await agent.run(question, conversation_id=conversation_id)
    return result.output

검색은 기본적으로 대화로 범위가 한정되므로, 과거 실행에 도달하려면 두 실행이 conversation_id를 공유해야 해요. pydantic-ai가 실행당 하나를 해석해요. 명시적 conversation_id=가 이기고, 아니면 message_history의 가장 최근 conversation_id를 상속하고, 아니면 새 것을 생성해요. message_history를 후속 실행을 통해 실어 나르면 한 대화에 유지되고, 명시적 id도 히스토리 체인도 공유하지 않는 실행은 분리돼요.

scope는 이전 릴리스에서 all이 기본이었고 지금은 conversation이 기본이에요. 업그레이드는 아무것도 발생시키지 않아요. 스토어 전역 호출자가 더 좁은 말뭉치로 계속 동작하죠. 그래서 scope를 설정하지 않으면 인스턴스당 한 번 HarnessDeprecationWarning을 발생해요. scope='all'을 넘기면 예전 동작을, scope='conversation'을 넘기면 새 동작을 유지해요. 둘 다 지원되고 어느 것도 폐기되지 않아요.

일부 컴팩션 전략은 그 편집을 실행의 영속 메시지 히스토리에 영속해요(SummarizingCompaction은 요약된 프리픽스를 완전히 교체하고, SlidingWindowCompaction 트림은 각 요청이 보내는 것만 좁혀요). 어느 쪽이든 StepPersistence는 다음 컴팩션이 실행되기 전에 각 스텝 경계를 스냅샷하므로, 실행 스냅샷들의 합집합이 여전히 원본을 담아요. SnapshotHistorySource가 그것을 복구합니다. capability들 사이에 순서나 훅 조정은 필요 없어요. 검색 도구는 호출 시점에 스토어를 지연 읽어요.

속성 (Attributes)
source

검색 말뭉치가 오는 곳. StepPersistence capability가 쓰는 스토어 위에 SnapshotHistorySource 사용.

타입: HistorySource

scope

한 번의 검색이 닿을 수 있는 스토어의 양.

conversation은 말뭉치를 호출 실행의 conversation_id와 일치하는 실행으로 제한해요. conversation_id가 없는 실행은 아무것도 검색하지 않고 도구가 그렇게 말해요. 다른 라벨 없는 실행으로 폴백하지 않아요. all은 소스가 열거하는 모든 실행을 검색하며, 스토어가 이미 한 주체로 격리된 경우에만 써야 해요.

None은 호출자가 선택하지 않았다는 뜻이고, conversation으로 해석되며 인스턴스당 한 번 HarnessDeprecationWarning을 발생해요. 이 기본값은 이전 릴리스에서 all이었고, 스토어 전역 호출자가 계속 동작하고 단지 다른 대화를 보지 못할 뿐이라 변경은 달리 조용합니다. 옵션을 명시적으로 설정하면 경고를 벗어나요. 두 값 모두 지원되고 어느 것도 폐기되지 않아요.

타입: SearchScope | None 기본: None

tool_id

search_conversation_history 도구의 toolset id.

타입: str 기본: 'conversation-search'

max_matches

검색 도구가 반환하는 최대 매칭 발췌 수. 음수여서는 안 됨.

타입: int 기본: 10

context_lines

각 검색 매칭 주변에 보이는 주변 줄 수. 음수여서는 안 됨.

타입: int 기본: 5

bm25_k1

BM25 용어 빈도 포화, 음수 아님. 이 capability의 기본값; Lucene의 BM25Similarity는 1.2 사용.

타입: float 기본: 1.5

bm25_b

BM25 길이 정규화, 0.0~1.0 사이 (Lucene/Elasticsearch 기본값).

타입: float 기본: 0.75

add_instructions

리콜 도구가 존재한다고 알리는 짧은 지시문 방출.

타입: bool 기본: True

effective_scope

실제 적용되는 범위: 호출자의 선택, 또는 설정 안 됐을 때 conversation.

타입: SearchScope

메서드 (Methods)
post_init
def __post_init__() -> None

호출자가 scope를 바뀐 기본값에 맡겼을 때, 생성 시 한 번 경고.

생성이 호출자의 선택(또는 비선택)이 표현되는 곳이므로, 여기서 경고하면 search_conversation_history 호출마다가 아니라 인스턴스당 한 번 발생해요.

반환

None

get_toolset
def get_toolset() -> AgentToolset[AgentDepsT] | None

소스 위에 search_conversation_history 도구를 제공해요.

반환

AgentToolset[AgentDepsT] | None

get_instructions
def get_instructions() -> AgentInstructions[AgentDepsT] | None

add_instructions가 false가 아니면 모델에게 리콜 도구가 존재한다고 알려요.

문구는 scope를 따라 구성된 것보다 더 좁은 말뭉치를 절대 설명하지 않아요.

반환

AgentInstructions[AgentDepsT] | None

더 알아보기 (Learn more)