Flink Agents 0.4.0으로 보는 스트리밍 에이전트

10월 30일에 Apache Flink Agents 0.4.0이 배포되었다. 프로덕션용 1.0을 앞두고 기능과 API를 정리하는 프리뷰 릴리스다. 이번 0.4.0 릴리스에는 나도 컨트리뷰터로 참여했다. 배포를 계기로 Flink Agents가 어떤 프로젝트인지, 에이전트를 어떻게 실행하고 복구하는지 정리했다.

#Flink 위에 에이전트를 올린 이유

결제 실패가 발생하거나 센서가 이상값을 보내면, 시스템은 그 이벤트를 받아 원인을 판단하고 필요한 조치를 이어 가야 한다. 사람이 채팅창에 질문할 때까지 기다릴 수 없는 일이다. Flink Agents는 이런 작업을 위한 프로젝트로 시작됐다. 계속 들어오는 이벤트를 에이전트가 받아 판단하고, 도구를 호출해 그에 맞는 작업을 수행하도록 만드는 것이 목적이다.

이벤트가 많아지면 여러 서버에 처리를 나눠야 하고, 고객이나 장비별로 앞서 처리한 내용을 기억해야 한다. 계속 실행하는 시스템이므로 서버에 장애가 나거나 실행이 중단돼도 작업을 이어 갈 수 있어야 한다. 모델을 호출하는 코드만으로는 이 실행 환경을 만들 수 없다.

Flink는 데이터 스트림을 분산 처리하고, 키별로 상태를 관리하며, 장애가 나면 체크포인트에서 복구하는 기능을 갖추고 있다. Flink Agents는 이 런타임 위에 모델과 도구, 메모리를 사용하는 에이전트의 실행 구조를 더한 Apache Flink의 하위 프로젝트다. 이벤트를 받는 것부터 에이전트의 판단과 결과 처리까지 하나의 Flink 잡 안에서 연결한다.

#0.1.0에서 0.4.0까지

첫 프리뷰인 0.1.0은 2025년 10월에 배포됐다. 작업을 액션이라는 단위로 나누고 이벤트로 연결해, Flink 안에서 모델과 도구를 호출하는 기본 구조를 마련했다. 실행을 따라갈 수 있는 이벤트 로그와 완료된 액션의 기록을 복구에 재사용하는 기능도 이때 포함됐다. Flink 1.20.3을 기반으로 Python과 Java API를 제공했지만, 초기 기능은 Python 위주로 구현되었다.

2026년 2월의 0.2.0은 Java API의 기능을 보강하고 Flink 지원 범위를 1.20부터 2.2까지 넓혔다. 다른 언어로 구현된 리소스도 사용할 수 있게 됐다. 메모리는 한 번의 실행에만 필요한 것과 다음 실행에도 남길 것을 구분하도록 확장됐다. 복구할 때는 완료된 액션의 실행 기록뿐 아니라, 아직 끝나지 않은 액션 안에서 저장해 둔 개별 호출의 결과도 다시 쓸 수 있게 됐다.

0.3.0은 2026년 6월에 배포됐다. 모델과 도구 설정을 YAML로 선언하고, 다른 언어로 작성한 액션도 함께 실행할 수 있게 됐다. 장기 메모리는 Mem0을 기반으로 정보를 추출하고 정리하도록 바뀌었고, 작업 지침과 참고 자료를 제공하는 Agent Skills도 추가됐다. 실행 기록을 저장하는 곳으로 Apache Fluss를 선택할 수 있게 됐으며, 웹 UI에서 이벤트 로그를 확인하는 기능도 추가됐다. 복구할 때 결과를 알 수 없는 외부 호출을 확인하는 조정 함수도 0.3.0에 추가됐다. 이전에는 저장된 호출 결과를 다시 쓰는 방식이었다면, 이제 호출 도중 중단돼 결과가 남지 않은 경우에도 외부 시스템의 상태를 확인할 방법이 생겼다.

이번에 배포된 0.4.0은 그 위에 하위 에이전트와 모델 라우팅 같은 기능을 더했다. 모델 호출 뒤의 로직을 한 액션 안에 쓰는 ctx.chat도 추가하고, 이벤트뿐 아니라 모델과 도구 호출까지 추적하도록 관측 기능을 넓혔다. 1.0부터 API 호환을 약속하기에 앞서, 기존 API와 기본 동작도 함께 정리했다.

#에이전트의 실행 구조

에이전트는 Flink 잡 안의 연산자로 실행된다. 앞 단계에서 데이터를 받아 작업을 수행하고, 결과를 다음 연산자로 넘긴다. 에이전트 안의 작업도 이벤트로 이어진다. 기본 구성은 이벤트와 액션, 리소스다. 이벤트가 들어오면 조건에 맞는 액션이 실행되고, 액션은 필요한 작업을 한 뒤 새 이벤트를 보낸다. 하나의 이벤트에 여러 액션이 반응할 수도 있어서 분기나 반복을 만들 수 있다. 모델과 도구, 프롬프트처럼 여러 단계에서 함께 쓰는 것은 리소스로 등록하고 이름으로 참조한다. MCP 서버의 도구와 프롬프트도 연결할 수 있고, 0.3.0부터는 리소스를 YAML로 선언할 수 있다.

상품 리뷰를 요약하는 에이전트라면 입력 레코드가 InputEvent로 들어오고, 이를 받은 액션이 ChatRequestEvent로 요약을 요청한다. 내장 채팅 액션이 모델을 호출해 ChatResponseEvent를 보내면, 응답을 처리하는 액션이 요약문을 OutputEvent에 담는다. 그 값이 Flink의 다음 연산자로 전달된다. 이처럼 단계 사이를 이벤트로 연결하므로, 이벤트 기록을 따라가면 어떤 입력에서 어떤 판단과 출력이 이어졌는지 볼 수 있다.

작업 순서를 코드로 정할 수도 있고 모델에 다음 행동을 맡길 수도 있다. 내장 ReActAgent에는 모델을 지정하고, 필요에 따라 도구와 프롬프트를 더한다. 모델은 목표를 달성하는 데 필요한 도구를 고르고, 그 결과를 보고 다음 행동을 결정한다. 전체 작업은 코드로 정해 두고 일부 조사만 이런 에이전트에 맡기는 식으로 두 방식을 섞을 수도 있다.

에이전트를 정의했다면 Flink 잡에 연결한다. 0.4.0 문서의 Python 예시에서 그 부분만 가져오면 다음과 같다.

env = StreamExecutionEnvironment.get_execution_environment()
agents_env = AgentsExecutionEnvironment.get_execution_environment(env)

output_stream = (
    agents_env.from_datastream(
        input=input_stream, key_selector=lambda x: x.id
    )
    .apply(your_agent)
    .to_datastream()
)

입력은 키별로 나뉘고, 앞 단계에서 들어온 레코드마다 새 실행이 시작된다. 문서에서는 이를 agent run이라고 부른다. 위 코드의 x.id가 고객 ID라면 같은 고객의 입력끼리 메모리를 공유하도록 구성할 수 있다. 키 선택 함수를 넘기면 그 함수를 쓰고, 생략하면 DataStream에 이미 지정된 키를 쓴다. 둘 다 없으면 연결할 수 없다. Table도 연결할 수 있지만, 입력 테이블은 행을 추가하기만 해야 한다. 기존 행을 갱신하거나 삭제하는 집계 결과는 그대로 연결할 수 없다.

메모리는 얼마나 오래 남기고 어떻게 읽을지에 따라 구분한다. 감각 메모리와 단기 메모리는 모두 Flink 상태에 저장된다. 감각 메모리는 한 번의 실행이 끝나면 지워지고, 단기 메모리는 같은 키로 들어온 다음 실행에서도 읽는다. 고객별 에이전트라면 이번 요청을 처리하며 잠깐 필요한 값은 감각 메모리에, 다음 요청에도 쓸 값은 단기 메모리에 두는 식이다.

장기 메모리는 여러 실행에서 얻은 정보를 정리해 두고 의미 검색으로 찾는다. 0.3.0부터는 Mem0이 정보 추출과 정리를 맡고, 실제 데이터는 외부 벡터 저장소에 보관한다. 이를 위해 채팅 모델과 임베딩 모델, 벡터 저장소를 함께 설정해야 한다. 장기 메모리의 검색 범위도 기본적으로 같은 잡과 입력 키 안이다.

모델에 작업 지침을 제공하는 Agent Skills도 있다. SKILL.md에 적은 지침과 관련 스크립트, 참고 자료를 묶어 두면 에이전트가 필요할 때 내용을 읽어 쓴다.

#장애에서 복구하는 방식

에이전트는 작업 하나를 마칠 때까지 모델과 외부 도구를 여러 번 호출할 수 있다. 그사이에 타임아웃이나 호출 제한을 만날 수 있고, 에이전트를 실행하던 프로세스가 중단될 수도 있다. 호출 오류에 대응하는 재시도 정책과 별개로, 실행이 중단된 뒤에는 이미 끝난 작업을 반복하지 않고 남은 작업을 이어 갈 방법도 필요하다. 처음부터 다시 실행한다고 이전과 같은 결과를 얻는 것도 아니다. 모델은 같은 입력에도 다른 답을 낼 수 있고, 외부 시스템에서 조회한 값도 그사이에 바뀔 수 있다. 이미 보낸 메일이나 결제 요청까지 반복하면 외부 작업이 중복될 수 있다. 복구하려면 무엇이 끝났고 무엇을 다시 실행해야 하는지 구분해야 한다.

Flink는 장애가 나면 마지막 체크포인트에서 상태와 입력 위치를 복원하고, 그 뒤의 입력을 다시 처리한다. 외부 저장소에 결과를 기록할 때는 트랜잭션이나 멱등성을 지원하는 싱크를 사용해, 같은 결과가 중복으로 반영되지 않게 할 수 있다. 하지만 에이전트가 입력을 다시 처리하는 동안에는 모델과 도구도 다시 호출할 수 있고, 이미 수행한 외부 작업을 체크포인트로 되돌릴 수는 없다.

완료된 액션의 실행 기록을 복구에 쓰려면 Flink 체크포인트에 더해 액션 상태 저장소(Action State Store)를 설정한다. Kafka나 Apache Fluss에 액션의 실행 기록을 남기는 방식이며, 완료된 액션의 기록을 재사용하는 기능은 0.1.0부터 제공됐다. 0.4.0의 일관성 문서에 따르면, 완료 기록이 있는 액션은 본문을 다시 실행하지 않고 저장해 둔 감각 메모리와 단기 메모리의 변경, 출력 이벤트를 그대로 재생한다. 다만 비동기로 실행되던 액션들이 공유 메모리를 읽고 쓴 순서까지 재현하지는 않는다. 그 순서에 따라 결과가 달라진다면 필요한 순서를 이벤트 의존성으로 명시해야 한다.

0.2.0에서는 복구할 때 다시 실행하지 않을 범위를 액션 안의 호출 단위까지 좁혔다. 액션이 끝나지 않았더라도 그 안에서 내구 실행(durable execution)으로 감싼 호출이 끝나서 결과가 저장됐다면, 복구 때 그 호출은 다시 부르지 않는다. 액션 문서는 프로필 조회를 예로 든다.

class ProfileAgent(Agent):
    @action(EventType.InputEvent)
    @staticmethod
    def process_input(event: Event, ctx: RunnerContext) -> None:
        user_id = InputEvent.from_event(event).input
        profile = ctx.durable_execute(
            profile_service.get_profile,
            user_id,
            durable_id=f"profile:{user_id}",
        )
        ctx.send_event(OutputEvent(output=profile))

프로필 조회가 끝나고 결과까지 저장된 뒤에 액션이 실패하면, 복구된 액션은 서비스를 다시 부르지 않고 저장된 프로필을 꺼내 쓴다. 문서에도 적혀 있듯 내구 호출 바깥의 코드는 다시 실행될 수 있다. 완료 기록과 호출 결과는 외부 저장소에 동기 방식으로 기록하며, 복원할 체크포인트와 그 복구에 필요한 실행 기록도 함께 보존해야 한다. 복구한 실행이 이전 기록을 찾아 쓰려면 키별 입력 순서와 키 타입, 직렬화 설정, 액션 이름이 그대로여야 한다. 내구 호출도 같은 입력과 순서, 안정적인 식별자를 유지해야 하고, 저장된 값은 복원 시점에도 읽을 수 있어야 한다.

호출을 시작한 뒤 결과를 저장하기 전에 장애가 나면, 호출이 실제로 성공했는지 기록만으로 알 수 없다. 외부 서비스가 같은 요청을 반복해도 결과를 중복으로 만들지 않는다면 다시 시도할 수 있다. 그렇지 않다면 그 요청이 처리됐는지부터 확인해야 한다. 0.3 로드맵 해설에서도 이 문제를 짚고, 사용자가 외부 상태를 확인할 수 있는 함수를 제안했다.

0.3.0에는 이 확인 절차를 구현할 수 있는 조정 함수(reconciler)가 추가됐다. 복구할 때 진행 중인 호출의 기록은 있지만 결과가 없다면 이 함수가 실행된다. 조정 함수에는 외부 시스템의 처리 상태를 확인하고, 원래 결과를 돌려주거나 실패를 알리는 코드를 작성한다. 그렇게 얻은 결과도 기록에 남는다. 조정 함수가 없으면 원래 호출을 다시 실행한다.

액션 상태 저장소는 외부 동작과 그 결과 저장을 하나의 트랜잭션으로 묶지 못한다. 그래서 외부 상태를 바꾸는 호출은 반복될 수 있다고 보고 설계해야 한다. 외부 서비스의 멱등성 지원을 쓰거나, 내구 호출에 원래 결과를 찾아낼 수 있는 조정 함수를 두는 방법이 있다.

#새롭게 추가된 기능

이번 버전에서는 작업을 다른 에이전트에 맡기거나 요청에 맞는 모델을 선택하는 기능이 추가됐다. 모델 호출과 응답 처리를 한 액션에 작성할 수 있게 됐고, 운영 중에는 모델과 도구의 호출 기록도 확인할 수 있게 됐다.

0.3.0이 배포된 뒤인 2026년 9월 15일, TypeSafe는 Jev를 공개했다. 입력된 맥락을 정해 둔 기준에 따라 판단하고, 그 결과를 확률이나 선택지, 점수로 반환하는 모델이다. Flink Agents는 이번 버전에 의사결정 모델 API를 추가해 Jev와 OpenAI, 알리바바 클라우드 Model Studio를 같은 인터페이스로 연결했다. 예를 들어 고객 문의를 청구 문제와 기술 지원 중 어디로 보낼지 판단하거나, 경보의 위험도를 정해 둔 척도로 평가할 수 있다. 모델에 전달할 사실과 기준, 결과를 받아 실행할 후속 작업은 애플리케이션에서 정한다.

개발 도구와 모델 연동도 보강됐다. Java 비동기 실행을 JDK 21뿐 아니라 JDK 11과 17에서도 지원한다. 채팅 모델은 Java에서 Gemini를, Java와 Python에서 vLLM과 IBM watsonx.ai를 새로 쓸 수 있다. Java에는 OpenAI 임베딩 모델도 추가됐다. 코딩 에이전트가 Flink Agents 애플리케이션을 만들고 검증할 때 참고하는 flink-agents-dev 스킬도 추가됐다.

이 기능들로 심각한 경보만 골라 처리하는 에이전트를 만든다고 해 보자. 조건 트리거로 심각도가 높은 경보에만 액션을 실행한다. 그 액션 안에서 ctx.chat으로 모델을 부르고 응답을 바로 처리하며, 모델이 등록된 도구를 쓰겠다고 하면 프레임워크가 대신 실행한다.

이벤트 기반 채팅 API에서는 요청과 응답을 각각 다른 액션에서 처리한다. 0.4.0에서도 이 방식을 그대로 지원한다. ctx.chat은 모델 호출 뒤의 로직을 같은 액션에 이어 쓰고 싶을 때 쓰는 API다. 내구 실행을 켜 두면 ctx.chat의 응답도 복구할 때 다시 쓸 수 있다.

추가 조사는 하위 에이전트에 맡길 수 있고, 이 호출도 내구 실행을 지원한다. 정해진 기준에 따라 위험도를 매겨야 하면 의사결정 모델을 쓰고, 요청마다 다른 모델을 쓰고 싶으면 모델 라우팅을 쓴다. 간단한 요청은 가벼운 모델로 보내고, 복잡한 작업에만 더 큰 모델을 사용하는 식이다.

운영할 때는 추적 로그로 입력 하나가 어떤 단계와 호출을 거쳐 처리됐는지 확인한다. 내장 지표에서 실패나 지연을 발견하면 이 로그로 어느 단계에서 문제가 생겼는지 좁혀 갈 수 있다. 기본 기록 범위는 이벤트뿐이어서 단계와 호출까지 보려면 trace-log.targets를 설정해야 한다. 이벤트 사이의 인과관계를 나무 모양으로 다시 그려 주는 도구도 함께 추가됐다. 이 도구는 이벤트 기록만 읽고, 실행 단계와 호출 기록은 건너뛴다. 추적 로그를 쓰는 데 실패해도 이벤트 처리와 잡 실행은 계속된다. 그래서 추적 로그만 보고 모든 실행이 빠짐없이 기록됐다고 단정할 수는 없다.

#0.4.0으로 업그레이드하는 방법

0.x는 아직 프리뷰 버전이라 API와 설정이 호환되지 않게 바뀔 수 있다. 이번에는 초기 프로덕션 사용자의 피드백과 API 검토에서 확인한 문제를 정리하면서, 이전 어느 릴리스보다 호환되지 않는 변경이 많아졌다. 0.4.0을 안정시킨 뒤 작은 조정을 거쳐 1.0을 배포하고, 그때부터 API 호환을 약속한다는 계획이다. 구체적인 호환 정책은 별도로 발표할 예정이다. 0.3.x에서 0.4.0으로 업그레이드할 때는 다음 변경 사항을 먼저 확인해야 한다.

• 액션 선언: Java @Action(listenEventTypes = ...)는 @Action(...)로, Python add_action(events=...)와 YAML listen_to는 trigger_conditions로 바뀐다. 흔히 쓰는 Python @action("event_type") 선언은 그대로 쓸 수 있다.
• Python 실행 경로: from_list()와 to_list()로 Flink 없이 실행하던 경로가 사라졌다. Flink의 StreamExecutionEnvironment에 DataStream이나 Table로 연결해야 하고, 로컬에서 개발할 때는 로컬 Flink 런타임을 쓴다.
• 재시도 기본값: 프레임워크의 max-retries 기본값이 3에서 0으로 바뀌었고 error-handling-strategy는 없어졌다. 실패한 채팅 응답은 애플리케이션에서 직접 처리해야 한다. 모델 제공자 SDK의 재시도는 따로 설정한다.
• 동시 실행 기본값: 기본 비동기 설정에서는 서로 독립된 도구 호출이 동시에 실행될 수 있다. 순서가 중요한 도구라면 단계를 나누거나 병렬도를 제한해야 한다. JDK 11부터 20에서 Java 액션만 있는 경우에는 실험 기능인 병렬 액션 실행이 기본으로 켜지며, parallel-execution.enabled로 끌 수 있다.
• 도구 인자 검증: 모르는 필드나 틀린 타입, 잘못된 값이 들어오면 실행 전에 실패한다. 숫자 문자열을 숫자로 바꿔 주지도 않는다.
• 로그 설정: 이벤트 로그가 추적 로그로 바뀌면서 event-log.* 같은 예전 설정 키는 거부된다. trace-log.* 키로 옮겨야 한다.
• 실행 기록: 0.3.x가 남긴 액션 상태 저장소 데이터는 0.4.0과 호환되지 않으니, 이전 실행 기록을 다시 쓰지 않는 업그레이드 계획이 필요하다. Flink 체크포인트와 세이브포인트의 복원 가능 여부는 이 저장소의 데이터 호환성과 별도로 확인해야 한다.
• 처리 중인 액션: 처리 중인 액션이 들어 있는 연산자 스냅샷은 0.4.0으로 업그레이드할 때 복원에 실패할 수 있다. 이전 버전에도 있던 제약이니, 실제와 비슷한 상태로 먼저 복원을 시험해 보는 편이 좋다.

메시지 형식도 바뀌었다. ChatMessage는 내용을 blocks에, 추가 속성을 metadata에 담는다. 모델 호출은 응답 메시지와 모델 정보, 사용량을 나눠 담은 ChatResult를 돌려주고, 도구 결과는 ToolResponse로 통일됐다. 모델 연동을 직접 만들었거나 메시지를 JSON으로 저장해 두었다면 이 부분부터 손봐야 한다.

#컨트리뷰터가 본 0.4.0

공지의 컨트리뷰터 명단에는 38명이 올라 있고, 나도 이번 릴리스부터 기여할 수 있었다. 이번에 들어간 내 PR 23개 가운데 다섯 개가 앞에서 설명한 내구 실행의 재생 경로를 고친 것이었다. 기록된 결과가 null이면 재생할 때 예외가 났고 #1171, 피클로 저장은 되지만 다시 만들 수 없는 Python 예외는 엉뚱한 TypeError로 재생됐다 #1182. 조정 함수 없이 완료 결과만 기록하는 Python 내구 호출은 Java 쪽에서 결과 저장에 실패해도 호출한 쪽에 알리지 않았다 #1211.

일반 호출과 내구 호출을 비동기로 함께 기다리다가 일반 호출이 취소되면, 입력 순서상 앞에 있던 내구 호출은 이미 끝났는데도 결과가 저장되지 않는 경우가 있었다 #1239. 내장 Python 채팅 호출의 식별자가 프로세스마다 달라질 수 있던 문제도 고쳤다 #1240. 그 밖에는 모델 라우팅 검증과 메모리 참조, 런타임 큐, 지표, 문서를 고쳤다.

릴리스 직전에는 첫 후보판을 검증하고 투표에도 참여했다. 그 과정은 「오픈소스에도 투표가 있다」에 따로 적었다. 검증 중에는 버전을 업그레이드한 뒤 기존 스냅샷을 복원할 때 발생하는 문제를 확인했고, 이 내용은 공식 공지의 업그레이드 주의 사항에 반영됐다. 이후 버전 간 상태 복원을 검증하는 테스트 기반을 만드는 작업도 맡았다. 1.0에서는 API가 안정되고 업그레이드 검증도 보강돼, 운영 중인 에이전트를 더 수월하게 유지할 수 있기를 기대한다.

#시작하려면

실행 환경 설정에서 사용할 Flink와 Java, Python 버전을 확인한 뒤 첫 에이전트 만들기의 예제를 실행하면 된다. 작업 순서부터 작성하기보다 모델에 도구를 맡겨 보고 싶다면 ReActAgent 빠른 시작이 맞다. 코딩 에이전트로 개발한다면 flink-agents-dev 스킬을 추가할 수 있다. 코딩 에이전트가 API와 예제를 참고하도록 돕는 개발 자료이며, Flink나 Flink Agents 실행 환경은 위 안내에 따라 따로 설치한다.

npx skills add 'apache/flink-agents#release-0.4@flink-agents-dev'

버그와 기능 요청은 GitHub 이슈로 받고, 사용 사례나 아이디어는 GitHub 토론과 Apache Flink 슬랙의 #flink-agents-user 채널에서 나눈다.

#참고 자료

• Apache Flink Agents 0.4.0 Release Announcement, Wenjin Xie, 2026. 0.4.0에서 추가된 기능과 호환되지 않는 변경, 그리고 0.4.0이 1.0 전 마지막 릴리스라는 계획을 적었다.
• Apache Flink Agents 0.3.0 Release Announcement, 2026. Flink 위에서 에이전트를 실행하는 이유와 0.4.0과 1.0으로 이어지는 API 호환 계획을 설명한다.
• Apache Flink Agents 0.2.0 Release Announcement, 2026. 세 가지 메모리와 호출 단위 내구 실행, 여러 Flink 버전 지원을 소개한다.
• Apache Flink Agents 0.1.0 Release Announcement, 2025. 프로젝트가 왜 필요한지와 첫 프리뷰 버전의 핵심 기능을 소개한다.
• Flink Agents 0.3 Roadmap解读, Flink 중문 커뮤니티, 2026. 0.3.0 로드맵을 컨트리뷰터 시점에서 해설하고, 장애 복구에서 남은 문제와 해법 후보를 짚는다. 해당 커뮤니티에는 아직 0.4.0 내용이 반영되지 않았다.
• Flink Agents 0.4.0 문서. 시작하기와 개발 가이드, 일관성과 복구 흐름을 설명하는 운영 장이 있다.