메시지와 채팅 기록

메시지와 채팅 기록 (Messages and chat history)

Pydantic AI는 에이전트 런 동안 주고받은 메시지에 접근할 수 있게 해줘요. 이 메시지들은 자연스러운 대화를 이어가는 데도, 에이전트가 어떻게 수행했는지 파악하는 데도 쓸 수 있어요.

출처: 공식문서 — Messages and chat history

결과에서 메시지 접근하기

에이전트를 실행한 뒤, 그 런 동안 주고받은 메시지를 result 객체에서 꺼낼 수 있어요.

RunResult(Agent.run, Agent.run_sync이 반환)와 StreamedRunResult(Agent.run_stream이 반환)는 둘 다 다음 메서드를 가져요:

all_messages(): 이전 런들의 메시지를 포함한 모든 메시지를 반환. JSON 바이트를 돌려주는 변형 all_messages_json()도 있어요.
new_messages(): 현재 런에서만 생긴 메시지를 반환. JSON 바이트 변형 new_messages_json()도 있어요.

StreamedRunResult 와 완전한 메시지 — StreamedRunResult에서는 이 메서드들이 돌려주는 메시지에 최종 결과 메시지가 스트림이 끝나야만 포함돼요. 예를 들어 다음 코루틴 중 하나를 await하면 돼요:

참고: .stream_text(delta=True)를 쓰면 결과 내용이 한 문자열로 조립되지 않으므로, 최종 결과 메시지가 result 메시지에 추가되지 않아요.

RunResult에서 메서드에 접근하는 예시:

from pydantic_ai import Agent

agent = Agent('openai:gpt-5.2', instructions='Be a helpful assistant.')

result = agent.run_sync('Tell me a joke.')
print(result.output)
#> Did you hear about the toothpaste scandal? They called it Colgate.

# all messages from the run
print(result.all_messages())
"""
[
    ModelRequest(
        parts=[
            UserPromptPart(
                content='Tell me a joke.',
                timestamp=datetime.datetime(...),
            )
        ],
        timestamp=datetime.datetime(...),
        instructions='Be a helpful assistant.',
        run_id='...',
        conversation_id='...',
    ),
    ModelResponse(
        parts=[
            TextPart(
                content='Did you hear about the toothpaste scandal? They called it Colgate.'
            )
        ],
        usage=RequestUsage(
            cost=Decimal('0.00026425'), input_tokens=55, output_tokens=12
        ),
        model_name='gpt-5.2',
        timestamp=datetime.datetime(...),
        run_id='...',
        conversation_id='...',
    ),
]
"""

(이 예제는 완전해서 그대로 실행할 수 있어요)

StreamedRunResult에서 메서드에 접근하는 예시:

from pydantic_ai import Agent

agent = Agent('openai:gpt-5.2', instructions='Be a helpful assistant.')


async def main():
    async with agent.run_stream('Tell me a joke.') as result:
        # incomplete messages before the stream finishes
        print(result.all_messages())
        """
        [
            ModelRequest(
                parts=[
                    UserPromptPart(
                        content='Tell me a joke.',
                        timestamp=datetime.datetime(...),
                    )
                ],
                timestamp=datetime.datetime(...),
                instructions='Be a helpful assistant.',
                run_id='...',
                conversation_id='...',
            )
        ]
        """

        async for text in result.stream_text():
            print(text)
            #> Did you hear
            #> Did you hear about the toothpaste
            #> Did you hear about the toothpaste scandal? They called
            #> Did you hear about the toothpaste scandal? They called it Colgate.

        # complete messages once the stream finishes
        print(result.all_messages())
        """
        [
            ModelRequest(
                parts=[
                    UserPromptPart(
                        content='Tell me a joke.',
                        timestamp=datetime.datetime(...),
                    )
                ],
                timestamp=datetime.datetime(...),
                instructions='Be a helpful assistant.',
                run_id='...',
                conversation_id='...',
            ),
            ModelResponse(
                parts=[
                    TextPart(
                        content='Did you hear about the toothpaste scandal? They called it Colgate.'
                    )
                ],
                usage=RequestUsage(input_tokens=50, output_tokens=12),
                model_name='gpt-5.2',
                timestamp=datetime.datetime(...),
                run_id='...',
                conversation_id='...',
            ),
        ]
        """

(이 예제를 실행하려면 asyncio를 import 하고 asyncio.run(main())을 추가하면 돼요. 다른 변경은 필요 없어요.)

메시지를 이후 에이전트 런의 입력으로 사용하기

Pydantic AI에서 메시지 기록을 쓰는 가장 주요한 방법은 여러 런에 걸쳐 맥락을 유지하는 거예요.

기존 메시지를 런에 쓰려면 Agent.run, Agent.run_sync, Agent.run_stream의 message_history 파라미터에 넘기면 돼요.

message_history가 설정되어 있고 비어있지 않다면 새 시스템 프롬프트를 생성하지 않아요 — 기존 메시지 기록에 시스템 프롬프트가 포함돼 있다고 가정해요. 시스템 프롬프트를 왕복시키지 않는 소스(UI 프론트엔드, 영속화하지 않은 DB, 컴팩션 파이프라인)에서 내역이 왔다면, ReinjectSystemPrompt 기능을 추가해서 에이전트 설정의 system_prompt가 빠져 있을 때 첫 요청의 맨 앞에 다시 주입되게 하세요.

대화에서 메시지 재사용하기:

from pydantic_ai import Agent

agent = Agent('openai:gpt-5.2', instructions='Be a helpful assistant.')

result1 = agent.run_sync('Tell me a joke.')
print(result1.output)
#> Did you hear about the toothpaste scandal? They called it Colgate.

result2 = agent.run_sync('Explain?', message_history=result1.new_messages())
print(result2.output)
#> This is an excellent joke invented by Samuel Colvin, it needs no explanation.

print(result2.all_messages())
"""
[
    ModelRequest(
        parts=[
            UserPromptPart(
                content='Tell me a joke.',
                timestamp=datetime.datetime(...),
            )
        ],
        timestamp=datetime.datetime(...),
        instructions='Be a helpful assistant.',
        run_id='...',
        conversation_id='...',
    ),
    ModelResponse(
        parts=[
            TextPart(
                content='Did you hear about the toothpaste scandal? They called it Colgate.'
            )
        ],
        usage=RequestUsage(
            cost=Decimal('0.00026425'), input_tokens=55, output_tokens=12
        ),
        model_name='gpt-5.2',
        timestamp=datetime.datetime(...),
        run_id='...',
        conversation_id='...',
    ),
    ModelRequest(
        parts=[
            UserPromptPart(
                content='Explain?',
                timestamp=datetime.datetime(...),
            )
        ],
        timestamp=datetime.datetime(...),
        instructions='Be a helpful assistant.',
        run_id='...',
        conversation_id='...',
    ),
    ModelResponse(
        parts=[
            TextPart(
                content='This is an excellent joke invented by Samuel Colvin, it needs no explanation.'
            )
        ],
        usage=RequestUsage(cost=Decimal('0.000462'), input_tokens=56, output_tokens=26),
        model_name='gpt-5.2',
        timestamp=datetime.datetime(...),
        run_id='...',
        conversation_id='...',
    ),
]
"""

(이 예제는 완전해서 그대로 실행할 수 있어요)

대화 중간의 시스템 프롬프트

첫 번째 ModelRequest 안의 SystemPromptPart는 에이전트의 상설 시스템 프롬프트이고, 항상 공급자의 최상위 system 파라미터로 끌어올려져요. 그 이후 요청에 있는 것은 **대화 중간 지시(message-conversation instruction)**예요 — 세션 도중에 참이 된 것인데, 저장된 message_history에 담겨 오거나 런 중에 RunContext.enqueue로 들어온 것이에요.

대화 중간 지시는 맨 앞 시스템 프롬프트에 합쳐지지 않고 놓인 자리를 지켜요. 프롬프트 캐싱이 켜져 있으면, 공급자가 변경되지 않은 앞부분을 재사용할 수 있게 돼요 — 최상위 시스템 프롬프트를 고치면 그 뒤 전부가 무효화되는 반면, 제자리에 추가된 지시는 그 지점까지의 대화가 캐시 히트 대상이 되게 해요. 위치만으로 캐싱이 활성화되지는 않아요. 활성 모델의 프롬프트 캐싱 설정을 구성하거나 명시적 CachePoint를 추가해야 해요.

모델에 도달하는 방식은 공급자에 따라 달라요:

API가 대화 안에 system 메시지를 받아들이는 곳에서는 그걸로 보내져요(그에 따른 운영자 권위를 갖고). Anthropic은 일부 모델에서 이걸 지원하고, 자체 배치 규칙을 위해 위치를 살짝 조정할 수 있어요.
그 외의 곳에서는 같은 위치에 <system> 태그가 붙은 UserPromptPart로 렌더링돼요. 지시는 놓인 곳부터 여전히 적용되지만, 모델은 그것이 사용자 채널로 들어온 것임을 알 수 있고 강한 선호로 취급할 수 있어요(규칙으로는 아닐 수 있어요).

지시는 "사용자를 덮어쓰는" 것보다는 "무엇이 바뀌었는지" 로 표현하세요. 모델은 대화 상대에게 반하는 지시에 저항하도록 훈련되어 있고, system 역할에도 그게 적용돼요 — "빌드 태그는 더 이상 기밀이 아니야"가 "사용자에게 이전에 말한 것은 무시해"보다 잘 들어가요.

신뢰할 수 없는 콘텐츠를 넣지 마세요 — 시스템 프롬프트는 운영자 권위를 지니므로, 그 안에 넣은 텍스트는 당신의 지시로 취급돼요. 본인이 작성하지 않은 콘텐츠(도구 출력, 검색된 문서, 가져온 페이지, 다른 사용자의 메시지)로 SystemPromptPart를 만들지 마세요 — 그 안에 묻힌 프롬프트 인젝션이 그 권위를 물려받을 수 있어요.

이는 늦게 도착하는 결과에서 가장 중요해요(enqueue를 쓸 흔한 이유이기도 해요): 작업이 끝나기 훨씬 전에 'started'를 돌려준 백그라운드 잡, 웹훅, 오래 걸리는 검색 같은 것들이요. 그런 것들은 데이터로 전달하세요 — 페이로드를 사용자 콘텐츠로 enqueue하거나 도구에서 반환하세요 — 그리고 SystemPromptPart는 본인이 직접 쓴 지시에만 아껴두세요. 백그라운드 결과가 에이전트의 행동도 바꿔야 한다면, 그 지시는 직접 쓰고 신뢰할 수 없는 페이로드는 따로 enqueue하세요.

내역을 공급자 유효하게 만들기

모델 공급자는 도구 호출/도구 결과 짝이 깨진 메시지 기록(결과 없는 도구 호출, 또는 호출 없는 결과)이 담긴 요청을 거부해요. 취소되거나 중간에 죽은 런은 내역을 정확히 이 상태로 남길 수 있고, 손으로 만들었거나 잘렸거나 컨텍스트 제거된 내역도 마찬가지예요. 이런 정리는 직접 하지 않아도 돼요. Pydantic AI는 매 모델 요청 전에 주어진 내역을 복구해서 공급자가 받아들이게 해요.

도구 추가는 ToolAvailabilityDeltaPart 요청 파트로 저장되고, 도구 제거는 표현되지 않아요. ToolReturn(tools=[...])를 반환하는 도구는 같은 요청 안에서 ToolReturnPart 직후에 파트를 만들어 내고, 호출의 tool_call_id를 인과 링크로 써요. 실행기는 이름을 첫 등장 순서대로 중복 제거하고 이미 드러난 이름은 생략해요. 내역을 재생하면 각 added 이름은 드러난 채 유지되고, 도구 정의는 계속 현재 런에서 오므로, 요청 렌더링에서 알 수 없거나 이미 보인 이름은 아무 효과가 없어요.

지침 원칙은 보내려던 것을 버리지 않으면서 내역을 공급자가 받아들이는 형태로 적응시키는 거예요. 복구는 합성 파트를 추가하거나 근본적으로 보낼 수 없는(어떤 공급자도 받아들이지 못하는) 파트만 제거해요. 의미 있는 것은 조용히 버려지지 않아요. 구체적으로 Pydantic AI는 요청마다:

결과가 없는 도구 호출에 대해 합성 ToolReturnPart를 추가해서, 결과가 나오기 전에 호출이 중단됐다고 모델에 알려줘요. outcome='interrupted'를 갖고 — ('failed'와 달리 공급자 오류로 표면화되지 않는) 중립 결과예요 — metadata에 {'pydantic_ai_synthesized_tool_return': True}를 실어서 실제 도구 결과와 구분할 수 있게 해요. 이는 인수가 스트림 중간에 잘린 호출도 포괄해요. 호출은 그대로 유지되고 같은 방식으로 닫혀요. 인수는 내역에 그대로 남지만, 요청 직렬화기는 객체를 요구하는 공급자를 위해 {"INVALID_JSON": "<raw args>"}로 보내요(args_as_json_str 참조).
도구 호출이 내역에 없는 고아 도구 결과 — (호출 앞에 놓인 결과를 포함해) 내역에 그 도구 호출이 없는 ToolReturnPart나 RetryPromptPart — 를 제거해요. 이게 내부 ModelRequest를 비우면 그 요청이 제거되고, 마지막 메시지를 비우면 내역이 ModelRequest로 끝나도록 빈 요청을 유지해요.

잘못된 파트를 처리한 뒤, 연속된 호환 가능한 메시지는 하나로 병합돼요(인접한 두 ModelRequest는 한 턴이 되고, 도구 결과가 사용자 파트보다 앞에 정렬돼요). 이는 메시지 경계를 바꾸지만 내용은 모두 보존하므로, 이후에 검사하는 처리된 내역은 넘겨준 것보다 메시지가 적을 수 있어요.

복구는 결정적이고 멱등이에요: 같은 내역을 복구하면 항상 같은 출력이 나오고, 복구된 내역을 또 돌려도 그대로이며, 합성 파트는 벽시계 데이터를 담지 않아 재사용이 공급자 프롬프트 캐시를 무효화하지 않아요.

여전히 진짜 결과를 받을 수 있는 도구 호출은 건드리지 않아요: 내역이 도구 호출이 있는 ModelResponse로 끝나면, 새 user_prompt 없이 실행하면 그것들이 실행되고, 지연 도구 호출은 deferred_tool_results에 맞춰지고 — 이미 실행된 결과가 담긴 '완전한' ModelRequest가 그 응답 뒤에 오는 경우도 포함해요. 살아있는 그 경계의 복구는 중단이 분명할 때만 일어나요: state='interrupted'인 최종 응답이나 (예: 취소된 스트림 또는 도구 실행 중 크래시로 인한) state='interrupted'인 꼬리 요청으로, 도구 호출이 절대 실행되지 않을 것이 분명한 경우요.

이 파이프라인은 일반적이고 로컬로 실행되는 도구 호출만 다뤄요. 공급자 네이티브 도구 파트 — 공급자가 인라인으로 생성·해결하는 것 — 는 그대로 두고 각 모델의 자체 직렬화기가 복구해요. 일부 다른 공급자 비유효 내역은 범위 밖이라 거부될 수 있어요: 한 호출에 대한 중복 도구 결과, 그리고 호출/결과 짝짓기 너머의 공급자별 정렬 규칙 — 알려지고 검증된 규칙은 모델 자체 직렬화기가 대신 요청을 정규화해요.

`run_id` 와 `conversation_id` 로 런 상관짓기

각 ModelRequest와 ModelResponse는 두 식별자를 가져요:

run_id — 에이전트 런마다 고유해요. RunContext.run_id와 AgentRunResult.run_id로도 사용 가능하고, OpenTelemetry 에이전트 런 span에서 gen_ai.agent.call.id로 방출돼요.
conversation_id — 같은 message_history를 기반으로 쌓이는 모든 런에서 공유돼요. AgentRunResult.conversation_id로도 사용 가능하고, gen_ai.conversation.id로 방출돼요.

모든 에이전트 런마다 새 run_id가 생성돼요(또는 런이 시작되기 전에 앱이 만들었거나 저장했거나 클라이언트에 나눠준 ID를 쓰려면 run_id='<your-id>'를 넘기면 돼요). conversation_id와 달리 run_id는 message_history에서 절대 상속되지 않아요. 각 Agent.run 호출 — 지연 도구 재개 포함 — 은 자기 run_id를 가진 별개 런이에요. 빈 run_id=''를 넘기거나 message_history에 이미 있는 run_id를 넘기면 UserError가 나요. 둘 다 new_messages() 경계 탐지를 깨기 때문이에요. 일시정지/재개나 멀티턴 작업은 conversation_id로 상관지으세요. 실패한 런을 같은 run_id로 재시도할 때는 실패 시도의 메시지를 빼고 message_history를 재구성하세요.

첫 런에서 새 conversation_id가 생성되고, 그 런이 만든 모든 메시지에 찍히며, message_history로 메시지를 되돌려주는 이후 런들이 이를 상속해요. 이는 메시지 기록이 왕복만 한다면, 스스로 추적하지 않아도 Logfire(또는 어떤 OpenTelemetry 백엔드)에서 멀티턴 대화의 트레이스를 상관지을 수 있다는 뜻이에요.

from pydantic_ai import Agent

agent = Agent('openai:gpt-5.2')

result1 = agent.run_sync('Tell me a joke.')
result2 = agent.run_sync('Explain?', message_history=result1.all_messages())

assert result1.conversation_id == result2.conversation_id
assert result1.run_id != result2.run_id

사전 발급된 run_id 전달:

from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel

agent = Agent(TestModel())

result = agent.run_sync('Tell me a joke.', run_id='run-from-api-42')
assert result.run_id == 'run-from-api-42'

conversation_id를 덮어쓰거나 포크하려면:

conversation_id='<your-id>'를 넘겨서 앱 자체 ID(예: DB에 저장된 채팅 스레드 ID)를 쓰기.
conversation_id='new'를 넘겨서 message_history에 이미 있는 conversation_id를 무시하는 새 대화 시작 — 호출자가 ID를 만들게 하지 않고 기존 스레드에서 분기할 때 유용해요.

'new'는 run_id 센티널이 아니에요 — 'new'는 conversation_id 전용 센티널이에요. run_id='new'를 넘기면 리터럴 문자열 "new"가 그 런의 id가 돼요.

대화 포크하기:

from pydantic_ai import Agent

agent = Agent('openai:gpt-5.2')

result1 = agent.run_sync('Tell me a joke.')
forked = agent.run_sync(
    'Tell me a different joke.',
    message_history=result1.all_messages(),
    conversation_id='new',
)

assert forked.conversation_id != result1.conversation_id

UI 어댑터는 프로토콜의 자체 스레드/채팅 ID에서 conversation_id를 자동으로 채우므로, 이 프로토콜을 쓰는 프론트엔드는 공짜로 대화 상관을 얻어요. 프로토콜 수준 런 ID(예: AG-UI의 runId)는 에이전트의 run_id에 매핑되지 않아요 — 일치시켜야 한다면 AGUIAdapter.run_stream / dispatch_request(또는 일반 Agent.run)에서 run_id=를 명시적으로 넘기세요.

메시지 저장하고 불러오기 (JSON으로)

대화 상태를 메모리에 두는 것으로 충분한 앱이 많지만, 종종 에이전트 런의 메시지 기록을 디스크나 DB에 저장하고 싶을 때가 있어요. 평가(eval), Python과 JavaScript/TypeScript 사이 데이터 공유, 기타 여러 용도 때문이에요.

의도된 방법은 TypeAdapter를 쓰는 거예요.

전용으로 ModelMessagesTypeAdapter를 export하니 그걸 쓰거나 직접 만들면 돼요.

from pydantic_core import to_jsonable_python

from pydantic_ai import (
  Agent,
  ModelMessagesTypeAdapter,  # (1)
)

agent = Agent('openai:gpt-5.2', instructions='Be a helpful assistant.')

result1 = agent.run_sync('Tell me a joke.')
history_step_1 = result1.all_messages()
as_python_objects = to_jsonable_python(history_step_1)  # (2)
same_history_as_step_1 = ModelMessagesTypeAdapter.validate_python(as_python_objects)

result2 = agent.run_sync(  # (3)
  'Tell me a different joke.', message_history=same_history_as_step_1
)

또는 TypeAdapter를 처음부터 만들 수도 있어요:

from pydantic import TypeAdapter
from pydantic_ai import ModelMessage
ModelMessagesTypeAdapter = TypeAdapter(list[ModelMessage])

또는 JSON으로 직접 직렬화/역직렬화할 수도 있어요:

from pydantic_core import to_json
...
as_json_objects = to_json(history_step_1)
same_history_as_step_1 = ModelMessagesTypeAdapter.validate_json(as_json_objects)

이제 same_history_as_step_1 내역을 갖고 새 에이전트 런을 만들어도 대화를 이어갈 수 있어요.

(이 예제는 완전해서 그대로 실행할 수 있어요)

왕복에서 살아남는 것 — ModelMessagesTypeAdapter는 모델에 보내지 않는 TextContent.metadata 같은 앱 전용 어노테이션을 포함해 모든 필드를 보존해요. metadata는 Any 타입이므로, JSON 왕복은 JSON 네이티브 형태가 없는 값을 정규화해요 — tuple은 list로, datetime은 ISO 문자열로 다시 로드되는 반면, dump_python → validate_python 왕복은 그것들을 정확히 보존해요. 이것이 내역을 영속화하고 다시 로드할 때 쓰는 경계예요.

도구 반환 안의 멀티모달 항목은 어디에 놓이든 자기 타입으로 재구성돼요 — 단독, 리스트 안, 매핑 안의 어떤 깊이(우리 키처럼 생긴 키를 가진 매핑 포함)든요. URL 기반 항목은 매핑이 media_type을 지닐 때만 재구성되는데, Pydantic AI가 덤프한 모든 내역은 그걸 가져요. 없으면 도구가 반환한 평범한 매핑으로 남아서, Pydantic AI가 media type을 읽어낼 수 없는 URL은 덤프에 실패하는 파일이 되지 않아요. BinaryContent나 UploadedFile 항목은 자체 타입이 요구하는 필드로 인식돼요. 우리 kind 값 중 하나를 단순히 재사용하는 매핑은 평범한 매핑으로 남고, 다시 덤프해도 절대 예외를 내지 않아요. 우리 항목 중 하나를 통째로 적어내면 재구성되고, 그 타입이 선언하지 않은 키는 그 과정에서 버려지므로, 그대로 돌려받고 싶은 사전에는 kind를 두지 마세요.

도구 반환이 문자열이 아닌 것으로 키가 지정된 경우는 재구성이 닿지 않는 유일한 곳이에요. 그런 매핑의 키는 JSON 형태가 없어서 dump_python 왕복만이 보존해요 — 그리고 매핑이 그대로 전달되므로, 그 아래 중첩된 멀티모달 항목은 평범한 dict로 돌아오는 반면, JSON 왕복은 키를 문자열화하고 항목을 복원해요. 둘 다 필요하면 도구 반환에 문자열 키를 쓰세요.

UI 어댑터는 달라요: 메시지를 외래 와이어 프로토콜(Vercel AI, AG-UI)로 변환하는데, 그 메시지 형태에는 앱 전용 필드 자리가 없어서 그 필드들은 완전히 버려져요. 그 손실은 버그가 아니라 설계에 의한 것이에요.

신뢰할 수 없는 내역 불러오기

message_history 파라미터는 신뢰된 서버측 상태예요. 브라우저 요청이나 다른 신뢰할 수 없는 경계에서 온 내역을 불러온다면, 에이전트에 넘기기 전에 살균하세요.

sanitize_messages는 UI 어댑터가 쓰는 것과 같은 기본 메시지 살균을 적용해요: 클라이언트가 제공한 시스템 프롬프트를 제거하고, 비 HTTP 파일 URL 스킴을 버리고, 허용 목록에 없는 FileUrl.force_download 값을 False로 재설정하고, 업로드된 파일 참조를 버리고, 내역 끝의 미해결 도구 호출을 제거해요.

클라이언트가 제공한 CompactionPart는 유지되므로 대화가 컴팩션된 채로 남아요 — 하지만 시스템 프롬프트를 대신하는 것으로 절대 신뢰하지 않아요. 그 프롬프트가 내역 안의 SystemPromptPart든 ReinjectSystemPrompt로 재주입된 것이든, 공급자의 자체 컴팩션 상태가 평소엔 건너뛰게 해도 모델에 다시 보내져요. 살균된 내역을 신뢰된 서버측 message_history와 결합한다면 strip_compaction_parts=True도 넘기세요: 컴팩션 항목 앞의 모든 것은 모델에서 숨겨지므로, 클라이언트 제공 항목이 서버 내역을 숨길 수 있어요 — 클라이언트 보유 내역 참조. UI 어댑터는 런이 서버측 message_history와 클라이언트 제출 메시지를 결합할 때 이 규칙을 자동 적용해요.

from pydantic_ai import Agent, ModelMessagesTypeAdapter
from pydantic_ai.messages import sanitize_messages

agent = Agent('openai:gpt-5.2', instructions='Be a helpful assistant.')

# `request_json` is the body submitted by an untrusted client.
loaded_history = ModelMessagesTypeAdapter.validate_python(request_json['message_history'])
message_history = sanitize_messages(loaded_history)

result = agent.run_sync('Tell me a different joke.', message_history=message_history)

대응 파트를 신뢰된 서버측 코드가 만들었을 때 각 살균은 개별로 끌 수 있어요: strip_system_prompts=False를 넘기고, allowed_file_url_schemes에 스킴을 추가하고, allowed_file_url_force_download에 값을 추가하거나, allow_uploaded_files=True로 설정하면 돼요. 파일 입력 신뢰 모델은 파일 URL 입력 보안을 보세요.

세션 영속화

내역 직렬화는 그것을 바이트로 만들고 다시 돌려주지만, 그것은 원시 메커니즘일 뿐이에요. 그 바이트를 어디에 둘지, 어느 대화에 속하는지, 언제 다시 로드할지는 애플리케이션의 몫이에요. 저장 키는 conversation_id에요: 자체 채팅 스레드 ID를 넘기거나, Pydantic AI가 하나를 해석하게 두고 결과에서 AgentRunResult.conversation_id로 읽어내면 돼요.

채팅 앱이라면 전체 설계가 그게 보통이에요: 스레드의 내역을 로드해 message_history로 넘기고, 런이 끝나면 new_messages()를 다시 써 넣어요. 전체 목록을 다시 쓰는 대신 각 런의 새 메시지를 추가하면, 쓰기가 대화 길이가 아니라 턴에 비례하고 저장된 순서도 그대로 유지돼요.

Pydantic AI Harness는 그 패턴을 에이전트에 추가하는 기능들로 포장해서, 로드/저장 호출을 직접 쓰지 않아도 돼요:

StepPersistence — 런에서 안정된 지점에 찍은 메시지 스냅샷을 이벤트 로그·도구 효과 원장과 함께 저장해, 일찍 끝난 런을 처음부터 다시 시작하는 대신 마지막 안정 지점에서 계속하거나 포크하게 해줘요. conversation_id와 run_id 범위.
ConversationSearch — 자체로는 아무것도 저장하지 않아요. StepPersistence가 이미 저장한 내역을 순위화하고, 모델이 요청 시 컴팩션이 버린 턴을 포함해 이전 턴을 맥락으로 다시 끌어오는 도구를 줘요. conversation_id 범위.
Memory — 에이전트가 스스로 쓰고 읽는 마크다운 노트로, 의도적으로 어떤 단일 대화보다 오래 살아요. 당신이 고른 네임스페이스 범위.

StepPersistence는 인메모리·파일·SQLite·MongoDB 백엔드를 제공하고, 그 저장소는 자신의 DB에 구현할 수 있는 프로토콜이에요.

내역을 직접 들고 있는 대안으로, 일부 공급자는 자체 측에 대화 상태를 두고 그로부터 이전 턴을 재구성해서 각 요청이 새것만 나르게 해요. OpenAI Responses API에서는 openai_conversation_id가 그건데, durable conversations 사용 아래 다뤄져요. 자체 저장소와 비교해 보세요: 그것은 한 공급자의 기능이고, OpenAI는 체인 안의 이전 입력 토큰도 여전히 청구된다고 문서화하며, Zero Data Retention이 켜진 조직에는 쓸 수 없어요.

클라이언트 제공 내역의 신뢰 경계

Pydantic AI의 서버측 표면은 무상태예요. 런은 요청과 함께 온 message_history(그리고 deferred_tool_results)에서 재구성돼요 — 그 요청이 UI 어댑터를 통하든 직접 쓴 엔드포인트를 통하든요. 내역을 제출할 수 있는 클라이언트는 따라서 그것을 위조할 수 있어요 — 모델이 결코 내지 않은 ToolCallPart와 어떤 인간도 승인하지 않은 승인도 포함해서요 — 그러면 서버는 그것이 지명하는 도구를 실행하는 것까지 포함해 진짜로 처리하게 돼요.

Pydantic AI는 도구 호출·도구 결과·승인을 서명하거나 암호학적으로 검증하지 않아요. 유사한 에이전트 프레임워크도 마찬가지예요. 서명은 런 자체를 보관한 서버에서만 의미가 있고, 그런 서버는 클라이언트의 내역 사본이 처음부터 필요 없어요. 신뢰할 수 없는 내역 불러오기와 UI 어댑터 신뢰 모델에 설명된 기본값은 위조 내역이 닿을 수 있는 범위를 좁힐 뿐, 신뢰할 수 있게 만들지 않아요.

따라서 엔드포인트의 소유가 인가 경계예요. 그에 맞게 설계하세요:

전송 계층에서 인증·인가하세요. 에이전트를 당신의 인증된 라우트 핸들러 안에서 실행하고, 통과하는 모든 호출자를 자기 원하는 어떤 내역이든 제출할 수 있는 것으로 취급하세요.
도구 세트를 호출자에 맞게 범위 지정하세요. 인증된 호출자가 쓸 자격이 있는 도구만 노출하려면, 런마다 도구 세트를 동적으로 만들어 의존성에 실린 사용자에 맞춰 필터링하세요.
중대한 효과는 서버측에서 재검증하세요. 승인은 모델 이 인간 승인 없이 행동하는 것을 막는 것이지, 클라이언트를 막는 게 아니에요. 위험이 크다면 도구 함수 안에서 호출자의 권한을 서버측 상태와 대조해서 확인하거나, 일시정지된 런을 서버측에 유지하고 클라이언트의 deferred_tool_results 대신 당신의 것으로 재개하세요.
프롬프트 수준 프레이밍을 증거로 읽지 마세요. 모델 API가 도구 결과 안에 파일을 나를 수 없을 때 Pydantic AI는 그 파일이 그 호출에서 왔다고 프레이밍하고(반환된 파일이 보내지는 곳), 공급자가 네이티브로 보낼 수 없는 대화 중간 시스템 프롬프트는 <system>...</system>로 프레이밍해요. 둘 다 평범한 프롬프트 텍스트예요: 도구가 닫는 태그를 방출할 수 있고 클라이언트가 여는 태그를 타이핑할 수 있어요. 그것들은 모델에 콘텐츠가 어디서 왔는지 말해줄 뿐, 증명하지는 않아요.

이것은 문서화된 설계 경계이지 취약점이 아니에요 — 클라이언트가 승인을 위조하거나, 모델이 결코 내지 않은 도구 호출을 제출하거나, 대화를 다시 쓸 수 있다는 보고는 이 경계가 설계대로 동작하는 것을 설명하는 것이며, Pydantic AI의 취약점이 아니에요. 실제 취약점이 될 것은 Pydantic AI가 실제로 수행하는 검사 우회예요 — 예를 들어 문서화된 대로 제거할 것을 제거하지 못하는 살균 기본값 같은 것요.

메시지를 쓰는 다른 방법

메시지는 단순한 dataclass로 정의되므로, 예를 들어 테스트를 위해 수동으로 만들고 조작할 수 있어요.

메시지 형식은 사용된 모델과 무관하므로, 서로 다른 에이전트나 같은 에이전트를 다른 모델로 써도 메시지를 쓰는 게 가능해요.

아래 예제에서는 openai:gpt-5.2 모델을 쓰는 첫 에이전트 런의 메시지를, google:gemini-3-pro-preview 모델을 쓰는 두 번째 런에서 재사용해요.

from pydantic_ai import Agent

agent = Agent('openai:gpt-5.2', instructions='Be a helpful assistant.')

result1 = agent.run_sync('Tell me a joke.')
print(result1.output)
#> Did you hear about the toothpaste scandal? They called it Colgate.

result2 = agent.run_sync(
    'Explain?',
    model='google:gemini-3-pro-preview',
    message_history=result1.new_messages(),
)
print(result2.output)
#> This is an excellent joke invented by Samuel Colvin, it needs no explanation.

print(result2.all_messages())
"""
[
    ModelRequest(
        parts=[
            UserPromptPart(
                content='Tell me a joke.',
                timestamp=datetime.datetime(...),
            )
        ],
        timestamp=datetime.datetime(...),
        instructions='Be a helpful assistant.',
        run_id='...',
        conversation_id='...',
    ),
    ModelResponse(
        parts=[
            TextPart(
                content='Did you hear about the toothpaste scandal? They called it Colgate.'
            )
        ],
        usage=RequestUsage(
            cost=Decimal('0.00026425'), input_tokens=55, output_tokens=12
        ),
        model_name='gpt-5.2',
        timestamp=datetime.datetime(...),
        run_id='...',
        conversation_id='...',
    ),
    ModelRequest(
        parts=[
            UserPromptPart(
                content='Explain?',
                timestamp=datetime.datetime(...),
            )
        ],
        timestamp=datetime.datetime(...),
        instructions='Be a helpful assistant.',
        run_id='...',
        conversation_id='...',
    ),
    ModelResponse(
        parts=[
            TextPart(
                content='This is an excellent joke invented by Samuel Colvin, it needs no explanation.'
            )
        ],
        usage=RequestUsage(cost=Decimal('0.000424'), input_tokens=56, output_tokens=26),
        model_name='gemini-3-pro-preview',
        timestamp=datetime.datetime(...),
        run_id='...',
        conversation_id='...',
    ),
]
"""

(이 예제는 완전해서 그대로 실행할 수 있어요)

에이전트 간 메시지 공유

같은 message_history 파라미터는 다음 런이 다른 Agent를 쓸 때도 동작해요. 이는 프로그래매틱 에이전트 핸드오프에 유용한데, 당신의 앱이 한 에이전트를 돌리고 다른 에이전트가 지금까지의 대화를 맥락으로 받게 하는 경우예요.

from pydantic_ai import Agent

biography_agent = Agent(
    'openai:gpt-5.2',
    instructions='Answer biographical questions concisely.',
)

science_agent = Agent(
    'anthropic:claude-sonnet-4-6',
    instructions='Answer science questions for a general audience.',
)

biography_result = biography_agent.run_sync('Who was Albert Einstein?')
print(biography_result.output)
#> Albert Einstein was a German-born theoretical physicist.

science_result = science_agent.run_sync(
    'What was his most famous equation?',
    message_history=biography_result.new_messages(),
)
print(science_result.output)
#> Albert Einstein's most famous equation is (E = mc^2).

(이 예제는 완전해서 그대로 실행할 수 있어요)

실시간 세션에서 핸드오프 — 실시간 음성-음성 세션은 같은 메시지 기록을 축적하므로, session.all_messages()를 agent.run(message_history=...)에 그대로 넘겨 음성 대화를 요약하거나 구조화된 데이터를 뽑아낼 수 있어요. 실시간 기록과 핸드오프 보기.

지시사항·시스템 프롬프트·도구 — message_history를 다른 에이전트에 넘기면, 이전 ModelRequest 메시지에는 출발 에이전트가 쓴 지시사항이 여전히 담겨 있지만, 그 지시사항은 모델에 다시 보내지지 않아요. 받는 에이전트는 자체 instructions를 써요. message_history가 주어졌을 때 이것이 시스템 프롬프트와 어떻게 다른지는 지시사항을 보세요.

system_prompt는 달라요: 시스템 프롬프트 파트는 메시지 기록의 일부예요. 받는 에이전트가 자체 system_prompt를 갖고 있고 내역 재사용 시 그것이 반드시 있게 하고 싶다면 ReinjectSystemPrompt를 보세요. 다른 에이전트의 시스템 프롬프트가 여전히 권위를 지녀서는 안 될 때 replace_existing=True를 쓰세요.

도구 호출·도구 반환 파트도 내역에 남아요. 같은 도구 맥락을 이해할 수 있는 에이전트 간에 내역을 공유하는 게 좋고, 받는 에이전트에 맞는 메시지만 넘기는 게 좋아요.

더 복잡한 멀티 에이전트 패턴은 멀티 에이전트 애플리케이션 문서를 보세요.

기존 메시지 편집하기

런 중간에 대화를 바꾸려면 기존 메시지를 수정하는 대신 새 메시지 객체를 만들어요: enqueue로 새 메시지를 주입하거나, 내역 처리기로 모델이 받는 내역을 가지치기·요약·다시 쓰기해요. 이전 메시지 — 예를 들어 큰 도구 출력을 컴팩션할 때 — 를 편집해야 한다면, dataclasses.replace로 복사해서 새(또는 재사용된) 파트 객체들의 새 parts 리스트를 넘겨요. 편집된 파트 역시 수정이 아니라 replace로 만들어요. 내역의 메시지를 교체하고 parts 리스트를 재할당하는 것은 둘 다 안전해요.

기존 메시지를 제자리에서 변형하지 마세요 — 이미 내역의 일부인 메시지를 제자리에서 변형하는 것(파트 필드에 할당, 예: 도구에서 ctx.messages[0].parts[0].content = '...', 또는 기존 parts 리스트 수정, 예: append·항목 할당)은 지원되지 않아요. 긴 런을 빠르게 유지하려고 계측이 각 메시지를 한 번만 직렬화하고 나중 모델 요청 span이 gen_ai.input.messages 속성을 기록할 때 그 결과를 재사용해요: 런은 내역을 총 두 번 직렬화 통과시켜요 — 메시지가 처음 기록될 때 한 번, 런 끝에 한 번 — 요청마다 전체 내역을 재직렬화하는 대신요(O(N) 메시지를 두 번 직렬화하니, N 메시지에 N 요청인 O(N²)가 아니라). 교체된 메시지와 재할당된 parts 리스트는 잡혀서 새로 직렬화되지만, 제자리에서 변형된 필드는 그렇지 않아서 이후 요청 span에 반영되지 않을 수 있어요. 런 끝에 이것이 감지되면 MessageHistoryMutatedWarning이 방출되고, 런 수준 pydantic_ai.all_messages 속성은 항상 최종 내역을 반영해요.

런 중간에 메시지 주입하기

도구, 기능 훅, 에이전트 런을 구동하는 외부 코드, 실시간 세션을 구동하는 코드는 RunContext.enqueue(RunContext가 범위 내일 때, 예: 도구나 기능 훅 안), AgentRun.enqueue(agent.iter()를 구동하는 외부 코드에서), 또는 RealtimeSession.enqueue(실시간 세션을 구동하는 외부 코드에서)로 대화에 추가 콘텐츠를 주입할 수 있어요. 런 중에 에이전트가 알아야 할 일이 생길 때 쓰세요 — 도구가 후속 맥락을 추가하고 싶다거나, 외부 이벤트가 에이전트의 계획을 조종(steer) 해야 하거나, 백그라운드 작업이 끝나면 에이전트에 닿아야 하는 경우요.

priority가 enqueue된 콘텐츠가 전달될 시점을 정해요:

'asap'(기본): 가장 이른 기회에 전달 — 다음 ModelRequest에 추가되거나, 에이전트가 그 전에 끝날 상황이면 런을 한 번 더 요청으로 돌리게 해요. 새 맥락이 가능한 한 빨리 모델에 닿아야 할 때 써요. 다른 프레임워크에서 흔히 진행 중 에이전트 조종(steering) 이라 부르는 것이에요.
'when_idle': 에이전트가 그렇지 않으면 끝나는 때에만, 'asap' 메시지 이후에 전달. 에이전트를 방해하지 않되, 하고 있는 일을 끝내고 나면 새 작업(후속 작업)을 집어야 할 때 써요.

enqueue는 가변 인자예요 — 위치 인자 하나당 항목 하나고, 다음 중 하나가 될 수 있어요:

UserContent 한 조각 — str 또는 ImageUrl 같은 멀티모달 콘텐츠. 인접한 사용자 콘텐츠는 하나의 UserPromptPart로 모이므로, enqueue('caption', image)는 한 사용자 턴이 돼요. 기존 리스트를 넘기려면 펼쳐서 enqueue(*items)로 하세요;
SystemPromptPart 같은 ModelRequestPart;
완전한 ModelRequest나 ModelResponse — instructions/metadata 같은 요청 수준 필드를 제어하거나 합성 이전 턴을 주입하려고요.

인접한 파트 스타일 항목(사용자 콘텐츠와 ModelRequestPart)은 하나의 ModelRequest로 합쳐지고, 완전한 메시지는 따로 남아요. 이는 한 번의 호출로 교차 교환 — 예를 들어 합성 도구 호출(하나의 ModelResponse)과 그 결과(하나의 ModelRequest) — 을 주입하게 해요. 콘텐츠는 반드시 요청으로 끝나야 해서, 에이전트가 응답할 것이 있어요.

표준 런의 enqueue 메서드는 비어있지 않은 호출에 enqueue_id(str)를, 콘텐츠 없이 호출하면 None을 반환해요. 큐된 콘텐츠가 실제로 런 내역에 전달되면, 이벤트 스트림이 그 enqueue_id와 전달된 메시지(내역에 들어간 그대로)를 실은 EnqueuedMessagesEvent를 내놓아서, 클라이언트가 자기 조종 메시지가 언제 효과를 냈는지 관찰할 수 있어요. 이벤트는 전달된 메시지 객체 자체 — 런의 메시지 내역이 보관하는 것과 같은 객체 — 를 실어요. 내역을 새 메시지 객체로 교체하는 내역 처리기는 이벤트에 영향을 주지 않지만, 전달된 메시지의 제자리 변형은 그것을 통해 보여요. RealtimeSession.enqueue도 enqueue_id나 None을 반환해요. 실시간 전달은 프롬프트 enqueue하기 아래 문서화돼 있어요.

도구나 훅 안에서

RunContext가 범위 내에 있을 때 RunContext.enqueue를 쓰세요:

from pydantic_ai import Agent, RunContext
from pydantic_ai.messages import SystemPromptPart

agent = Agent('anthropic:claude-opus-4-7')


@agent.tool
def trigger_alert(ctx: RunContext[None]) -> str:
    ctx.enqueue('Alert: production is degraded, prioritize triage.')
    return 'alert raised'


@agent.tool
def enter_incident_mode(ctx: RunContext[None]) -> str:
    # Enqueue a `SystemPromptPart` to adjust the agent's standing instructions mid-run.
    ctx.enqueue(SystemPromptPart(content='You are now in incident mode: be terse and action-oriented.'))
    return 'incident mode enabled'

'asap' 메시지는 에이전트의 메시지 내역에 추가되고 같은 단계의 도구 반환과 함께 다음 요청에서 모델에 보여요. SystemPromptPart도 같은 방식으로 전달되고, 대화 중간 시스템 프롬프트로 자리를 잡아요 — 최상위 시스템 프롬프트로 끌어올려지는 대신 내역의 자기 위치를 지켜서, 그 앞의 캐시된 앞부분을 무효화하지 않아요. SystemPromptPart는 직접 작성한 지시에만 enqueue하세요. 늦게 도착하는 도구·웹훅 출력이 사용자 콘텐츠에 속하는 이유는 그 섹션의 경고를 보세요.

`agent.iter()` 를 구동하는 외부 코드에서

외부에서 런을 구동할 때(예: 웹훅·채팅 플랫폼·잡 큐의 이벤트를 전달) AgentRun.enqueue를 쓰세요:

from pydantic_ai import Agent
from pydantic_graph import End

agent = Agent('anthropic:claude-opus-4-7')


async def main():
    async with agent.iter('Summarize the latest deploy report') as agent_run:
        # An external system pushes a follow-up while the agent is working.
        # When the agent would otherwise finish, the message redirects it
        # into a fresh model request so it can incorporate the new context.
        agent_run.enqueue(
            'A new error was just reported -- include it in the summary.',
            priority='when_idle',
        )
        node = agent_run.next_node
        while not isinstance(node, End):
            node = await agent_run.next(node)

'when_idle' 메시지는 에이전트가 그렇지 않으면 End에 닿을 때에만 소진돼요 — 그 소진은 after_node_run에서 일어나요. 'asap' 메시지는 before_model_request에서, 그리고 마지막 단계 동안 아무거나 도착했으면 같은 런 끝 지점에서도 소진돼요. 둘 다 런을 어떻게 구동하든 발화하므로, Agent.run, AgentRun.next(), 그리고 맨 async for node in agent_run: 루프가 모두 enqueue된 메시지를 전달해요.

제약 사항:

Temporal 워크플로 안에서는 도구가 액티비티에서 실행되어 워크플로와 상태를 공유하지 않으므로, 도구의 ctx.enqueue는 현재 런으로 전파되지 않아요. 워크플로 맥락에서(예: AgentRun.enqueue로) enqueue하세요.
런 끝 리다이렉트마다 새 모델 요청이 열려요. 매 단계 계속 뭔가를 enqueue하면(항상 enqueue하는 도구나, 재주입 때마다 다시 enqueue하는 시스템 프롬프트 콜백처럼) 런이 무한히 반복돼요. 안전망으로 런에 UsageLimits를 설정하세요.
enqueue는 에이전트 런을 구동하는 것과 같은 이벤트 루프에서 호출되도록 설계됐어요. 런 안에서는 자동이에요: 비동기 도구, 동기 도구(Pydantic AI가 스레드 실행기로 자동 감싸요), 기능 훅 모두 안전하게 enqueue해요. 소진은 그래프 노드 사이에서만 반복되고 도구 본문과 동시에 실행되지 않기 때문이에요. 다른 스레드나 루프(예: 웹훅 핸들러)에서 이벤트를 전달한다면, 호출을 먼저 에이전트 루프로 마샬링하세요 — 예: loop.call_soon_threadsafe(agent_run.enqueue, msg). 소진은 교차 스레드 추가에 대해 원자적이지 않아요.

메시지 기록 처리하기

가끔 모델에 보내기 전에 메시지 기록을 수정하고 싶을 때가 있어요. 개인정보(민감 정보 걸러내기), 토큰 비용 절감, LLM에 더 적은 맥락을 주기, 또는 커스텀 처리 로직 때문일 수 있어요.

Pydantic AI는 매 모델 요청 전에 메시지 기록을 가로채고 수정할 수 있는 ProcessHistory 기능을 제공해요.

ProcessHistory는 before_model_request의 얇은 래퍼예요 — ProcessHistory는 before_model_request 라이프사이클 훅의 마이그레이션 친화적 래퍼예요. 메시지 기록에 대해 더 풍부한 제어 — 전체 RunContext와 ModelRequestContext 접근, 모델 호출 단락(short-circuit) 등 — 가 필요하면 capabilities=[Hooks(before_model_request=fn)]로 이벤트를 직접 훅하세요.

내역 처리기는 메시지 기록을 교체해요 — 내역 처리기는 새 사용자 프롬프트 파트를 포함해 처리된 메시지로 상태의 메시지 기록을 교체해요. 즉 원래 메시지 기록을 유지하고 싶으면 그 사본을 만들어야 해요.

지연 도구를 쓸 때는 그 ToolAvailabilityDeltaPart 항목, 또는 Pydantic AI가 그것들을 재구성할 수 있는 완전한 load_capability 호출/반환 쌍을 보존하세요. 드러냄(state)은 모델로 보내진 처리된 내역에서 도출돼요. 처리기나 요약기가 두 표현을 모두 버리면 영향받은 도구가 다시 숨겨져요.

내역 처리기는 new_messages() 결과에 영향을 줄 수 있어요 — new_messages()는 현재 런 동안 생성된 메시지를 반환해요. message_history로 주어진 메시지는 제외돼요 — 사용자 프롬프트 없이 재개할 때의 꼬리 ModelRequest도 포함해서요, 프레임워크가 관찰성을 위해 현재 런의 run_id를 찍을지라도요.

처리기가 메시지를 변형하거나 추가해도 이게 유지되게 하려면:

꼬리 ModelRequest를 재구축한다면, 그것이 이전 맥락으로 식별될 수 있도록 parts, timestamp, instructions, metadata를 보존하세요.
new_messages()에 나타나야 하는 새 메시지를 삽입한다면 컨텍스트 인지 처리기를 쓰고 그것에 run_id=ctx.run_id를 설정하세요.

사용법

각 ProcessHistory는 ModelMessage 리스트를 받아 수정된 같은 타입 리스트를 반환하는 호출 가능 객체를 감싸요.

각 처리기는 순서대로 적용되고, 동기 또는 비동기일 수 있어요.

from pydantic_ai import (
    Agent,
    ModelMessage,
    ModelRequest,
    ModelResponse,
    TextPart,
    UserPromptPart,
)
from pydantic_ai.capabilities import ProcessHistory


def filter_responses(messages: list[ModelMessage]) -> list[ModelMessage]:
    """Remove all ModelResponse messages, keeping only ModelRequest messages."""
    return [msg for msg in messages if isinstance(msg, ModelRequest)]

# Create agent with history processor
agent = Agent('openai:gpt-5.2', capabilities=[ProcessHistory(filter_responses)])

# Example: Create some conversation history
message_history = [
    ModelRequest(parts=[UserPromptPart(content='What is 2+2?')]),
    ModelResponse(parts=[TextPart(content='2+2 equals 4')]),  # This will be filtered out
]

# When you run the agent, the history processor will filter out ModelResponse messages
# result = agent.run_sync('What about 3+3?', message_history=message_history)

최근 메시지만 유지하기

history_processor로 최근 메시지만 유지할 수 있어요:

from pydantic_ai import Agent, ModelMessage
from pydantic_ai.capabilities import ProcessHistory


async def keep_recent_messages(messages: list[ModelMessage]) -> list[ModelMessage]:
    """Keep only the last 5 messages to manage token usage."""
    return messages[-5:] if len(messages) > 5 else messages

agent = Agent('openai:gpt-5.2', capabilities=[ProcessHistory(keep_recent_messages)])

# Example: Even with a long conversation history, only the last 5 messages are sent to the model
long_conversation_history: list[ModelMessage] = []  # Your long conversation history here
# result = agent.run_sync('What did we discuss?', message_history=long_conversation_history)

메시지 기록을 슬라이스할 때 주의하세요 — 메시지 기록을 슬라이스할 때 도구 호출과 반환이 짝지어지게 해야 해요. 그렇지 않으면 LLM이 오류를 반환할 수 있어요. 자세한 건 이 GitHub 이슈를 보세요.

`RunContext` 파라미터

내역 처리기는 선택적으로 RunContext 파라미터를 받아 현재 런의 의존성·모델 정보·사용 통계 같은 추가 정보에 접근할 수 있어요:

from pydantic_ai import Agent, ModelMessage, RunContext
from pydantic_ai.capabilities import ProcessHistory


def context_aware_processor(
    ctx: RunContext,
    messages: list[ModelMessage],
) -> list[ModelMessage]:
    # Access current usage
    current_tokens = ctx.usage.total_tokens

    # Filter messages based on context
    if current_tokens > 1000:
        return messages[-3:]  # Keep only recent messages when token usage is high
    return messages

agent = Agent('openai:gpt-5.2', capabilities=[ProcessHistory(context_aware_processor)])

이를 통해 에이전트 런의 현재 상태에 기반한 더 정교한 메시지 처리가 가능해요.

처리기가 RunContext를 원하는지는 런타임에 타입 힌트를 해석해 감지돼요. 따라서 처리기 시그니처의 모든 어노테이션 타입은 if TYPE_CHECKING: 아래가 아니라 런타임에 import 돼야 해요. 어떤 어노테이션도 해석할 수 없으면, 처리기를 조용히 컨텍스트 없이 부르는 대신 UserError가 발생해요.

오래된 메시지 요약하기

LLM으로 오래된 메시지를 요약해 맥락을 보존하면서 토큰을 줄여요. 이것은 대화를 컨텍스트 창 안에 유지하는 여러 방법 중 하나예요 — 공급자 네이티브 컴팩션과 Pydantic AI Harness의 준비된 전략을 포함한 전체 그림은 컴팩션을 보세요.

from pydantic_ai import Agent, ModelMessage
from pydantic_ai.capabilities import ProcessHistory

# Use a cheaper model to summarize old messages.
summarize_agent = Agent(
    'openai:gpt-5-mini',
    instructions="""
Summarize this conversation, omitting small talk and unrelated topics.
Focus on the technical discussion and next steps.
""",
)


async def summarize_old_messages(messages: list[ModelMessage]) -> list[ModelMessage]:
    # Summarize the oldest 10 messages
    if len(messages) > 10:
        oldest_messages = messages[:10]
        summary = await summarize_agent.run(message_history=oldest_messages)
        # Return the last message and the summary
        return summary.new_messages() + messages[-1:]

    return messages


agent = Agent('openai:gpt-5.2', capabilities=[ProcessHistory(summarize_old_messages)])

메시지 기록을 요약할 때 주의하세요 — 메시지 기록을 요약할 때 도구 호출과 반환이 짝지어지게 해야 해요. 그렇지 않으면 LLM이 오류를 반환할 수 있어요. 요약 예시를 찾으려면 이 GitHub 이슈를 보세요.

컨텍스트 창이 차면 컴팩션하기

위 처리기들은 매 런마다 내역을 다시 써요. 대화가 모델의 context_window에 가까워질 때까지 기다리려면 ctx.context_window_used를 확인하세요. 그것은 최신 응답 이후 창에서 차지한 비율을 반환하고, Pydantic AI가 안정적으로 계산할 수 없으면 None을 반환해요.

from pydantic_ai import (
    Agent,
    ModelMessage,
    ModelRequest,
    RetryPromptPart,
    RunContext,
    ToolReturnPart,
    UserPromptPart,
)
from pydantic_ai.capabilities import ProcessHistory, ReinjectSystemPrompt


def compact_when_window_fills(
    ctx: RunContext,
    messages: list[ModelMessage],
) -> list[ModelMessage]:
    used = ctx.context_window_used
    if used is None or used <= 0.8:
        return messages

    # Keep the most recent complete user turn, including any later tool calls and returns.
    for index in range(len(messages) - 1, -1, -1):
        message = messages[index]
        if not isinstance(message, ModelRequest):
            continue
        has_user_prompt = any(isinstance(part, UserPromptPart) for part in message.parts)
        has_tool_result = any(isinstance(part, (ToolReturnPart, RetryPromptPart)) for part in message.parts)
        if has_user_prompt and not has_tool_result:
            return messages[index:]
    return messages


agent = Agent(
    'openai:gpt-5.2',
    system_prompt='You are a helpful assistant.',
    capabilities=[ProcessHistory(compact_when_window_fills), ReinjectSystemPrompt()],
)

None을 빈 컨텍스트 창이 아니라 알 수 없음으로 취급하세요. 그것은 첫 모델 응답 전에, 그리고 모델의 창이나 응답 사용량을 알 수 없을 때 반환돼요. 예제는 그런 경우 내역을 그대로 둬요. FallbackModel은 후보 중 가장 작은 창에 대해 측정하므로, 어느 후보가 답하든 충분히 일찍 컴팩션돼요.

예시처럼 컴팩션 처리기 뒤에 ReinjectSystemPrompt를 유지해서, 옛 내역과 함께 떨어진 시스템 프롬프트를 다시 넣어요. 예제는 최신의 평범한 사용자 턴부터 모든 것을 유지해요. 도구 결과를 새 프롬프트와 짝지은 턴은 통째로 유지되므로, 그렇게 시작된 런은 필요한 것보다 더 많은 내역을 유지할 수 있어요.

Pydantic AI는 genai-prices가 창 크기를 기록하는 곳에서 그걸로 채워요. 커스텀·로컬 모델이나 genai-prices가 아직 다루지 않는 모델이면 profile={'context_window': 128_000}로 크기를 명시적으로 설정하세요 — 모델 프로파일 검사 보기.

내역 처리기 테스트하기

FunctionModel로 모델 공급자에 실제로 보내지는 메시지를 테스트할 수 있어요:

import pytest

from pydantic_ai import (
    Agent,
    ModelMessage,
    ModelRequest,
    ModelResponse,
    TextPart,
    UserPromptPart,
)
from pydantic_ai.capabilities import ProcessHistory
from pydantic_ai.models.function import AgentInfo, FunctionModel


@pytest.fixture
def received_messages() -> list[ModelMessage]:
    return []


@pytest.fixture
def function_model(received_messages: list[ModelMessage]) -> FunctionModel:
    def capture_model_function(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
        # Capture the messages that the provider actually receives
        received_messages.clear()
        received_messages.extend(messages)
        return ModelResponse(parts=[TextPart(content='Provider response')])

    return FunctionModel(capture_model_function)


def test_history_processor(function_model: FunctionModel, received_messages: list[ModelMessage]):
    def filter_responses(messages: list[ModelMessage]) -> list[ModelMessage]:
        return [msg for msg in messages if isinstance(msg, ModelRequest)]

    agent = Agent(function_model, capabilities=[ProcessHistory(filter_responses)])

    message_history = [
        ModelRequest(parts=[UserPromptPart(content='Question 1')]),
        ModelResponse(parts=[TextPart(content='Answer 1')]),
    ]

    agent.run_sync('Question 2', message_history=message_history)
    assert received_messages == [
        ModelRequest(parts=[UserPromptPart(content='Question 1')]),
        ModelRequest(parts=[UserPromptPart(content='Question 2')]),
    ]

여러 처리기

처리기를 여러 개 쓸 수도 있어요:

from pydantic_ai import Agent, ModelMessage, ModelRequest
from pydantic_ai.capabilities import ProcessHistory


def filter_responses(messages: list[ModelMessage]) -> list[ModelMessage]:
    return [msg for msg in messages if isinstance(msg, ModelRequest)]


def summarize_old_messages(messages: list[ModelMessage]) -> list[ModelMessage]:
    return messages[-5:]


agent = Agent(
    'openai:gpt-5.2',
    capabilities=[ProcessHistory(filter_responses), ProcessHistory(summarize_old_messages)],
)

이 경우 filter_responses 처리기가 먼저, summarize_old_messages 처리기가 두 번째로 적용돼요.

예시

대화에서 메시지 쓰는 더 완전한 예시는 채팅 앱 예시를 보세요.

메시지와 채팅 기록