AbstractAgent

• AbstractAgent는 AG-UI 프로토콜에서 에이전트 역할을 수행하는 모든 참여자가 구현해야 하는 추상 베이스 클래스
• run(input) → Observable<BaseEvent> 단 하나의 추상 메서드만 구현하면 AG-UI 참여자가 되는 최소 계약(contract)
• 구현 위치에 따라 역할이 달라지는 양면 추상화: 백엔드에서 구현하면 AI 실행 서버, 프론트엔드에서 구현하면 이벤트 수신 클라이언트

해당 개념이 필요한 이유

• 서버·클라이언트가 동일한 인터페이스를 사용해 transport 방식(HTTP, WebSocket 등)과 무관하게 연결 가능
• run() 하나만 구현하면 상태 관리·이벤트 검증·미들웨어·구독자 알림이 자동으로 처리됨
• 백엔드 에이전트를 교체해도 프론트엔드 코드 변경이 불필요한 벤더 독립성 확보

AS-IS: 직접 구현 시

sequenceDiagram
    autonumber
    participant F as Frontend
    participant B as Backend Agent

    F->>B: HTTP POST (raw fetch)
    B-->>F: SSE raw bytes
    Note over F: bytes 파싱<br/>상태 관리<br/>이벤트 검증<br/>모두 직접 구현

TO-BE: AbstractAgent 사용 시

sequenceDiagram
    autonumber
    participant F as Frontend
    participant HA as "HttpAgent (AbstractAgent)"
    participant B as "Backend (AbstractAgent)"

    F->>HA: runAgent(input)
    HA->>B: HTTP POST + RunAgentInput
    B-->>HA: SSE raw bytes
    HA-->>HA: bytes → Observable[BaseEvent] 변환
    Note over HA: transformChunks<br/>verifyEvents<br/>apply (state/messages 자동 갱신)<br/>subscribers 알림
    HA-->>F: subscribers 콜백

구현 위치에 따른 역할 차이

같은 추상 클래스지만 어디서 구현하느냐에 따라 역할이 완전히 달라진다.

// 백엔드 — LLM 호출 후 이벤트 직접 생성
class BackendAgent extends AbstractAgent {
  run(input: RunAgentInput): Observable<BaseEvent> {
    return new Observable(subscriber => {
      const stream = await llm.stream(input.messages)
      subscriber.next({ type: EventType.RUN_STARTED, ... })
      for await (const chunk of stream) {
        subscriber.next({ type: EventType.TEXT_MESSAGE_CONTENT, delta: chunk.text })
      }
      subscriber.next({ type: EventType.RUN_FINISHED, ... })
      subscriber.complete()
    })
  }
}
 
// 프론트엔드 — 서버 SSE를 수신해 Observable로 변환 (핵심 수도코드)
class FrontendAgent extends AbstractAgent {
  run(input: RunAgentInput): Observable<BaseEvent> {
    return new Observable(subscriber => {
      fetch(url, { method: "POST", body: JSON.stringify(input) })
        .then(res => res.body.getReader())
        .then(async reader => {
          while (true) {
            const { done, value } = await reader.read()  // Uint8Array 청크
            if (done) break
            const event = parseSSE(value)                // JSON.parse(...)
            subscriber.next(event)                       // re-emit
          }
          subscriber.complete()
        })
    })
  }
}
// → HttpAgent가 FrontendAgent를 이미 구현해 제공
const agent = new HttpAgent({ url: "https://my-backend/agent" })

구분	백엔드 구현	프론트엔드 구현
run()의 역할	LLM 호출 → 이벤트 생성	HTTP bytes → Observable 변환
이벤트 방향	emit (창조)	re-emit (변환·전달)
기본 제공 구현체	없음 (직접 구현)	HttpAgent

실제 AI와 통신하려면 서버는 반드시 실행 중이어야 한다

run() 안에서 무엇을 호출하든, 실제 AI 처리가 일어나는 서버가 떠 있어야 한다.

Ollama 로컬 서버 연동:  run() → fetch("http://localhost:11434") → Ollama 서버 (필수 실행)
외부 LLM API 연동:      run() → fetch("https://api.openai.com") → OpenAI 서버 (외부, 상시 실행)

단, 서버 없이 동작하는 경우도 있다 — 테스트용 MockAgent:

// 테스트 목적으로만 사용. 실제 AI와 통신하지 않음
class MockAgent extends AbstractAgent {
  run(input: RunAgentInput): Observable<BaseEvent> {
    return of([고정된 이벤트들])
  }
}
// CopilotKit의 selfManagedAgents prop이 이 패턴을 지원
<CopilotKitProvider selfManagedAgents={{ mock: new MockAgent() }}>

run()이 반환하는 Observable의 의미 — 프론트엔드에서 run()을 구현하는 이유

왜 이 질문이 생기는가

서버가 이미 이벤트를 만들어 전달하니 클라이언트는 그냥 옵저빙만 하면 될 것 같다는 직관이 있다. 그렇다면 프론트엔드에서 run()을 구현할 필요가 있는가?

답은 “무엇을 옵저빙하느냐” 에 있다. 서버의 Observable<BaseEvent>는 네트워크를 통과하는 순간 소멸된다. HTTP는 바이트만 전달할 수 있기 때문이다. 프론트엔드 run()은 이벤트를 창조하는 것이 아니라 소멸된 Observable을 raw bytes로부터 복원하는 역할이다.

sequenceDiagram
    autonumber
    participant B as "BackendAgent.run()"
    participant NET as "Network (HTTP)"
    participant C as "FrontendAgent.run()"
    participant PP as "runAgent() Pipeline"

    Note over B: Observable[BaseEvent] 생성
    B->>B: subscriber.next(TEXT_MESSAGE_CONTENT, delta:안녕)

    Note over B,NET: Observable은 네트워크 전송 불가<br/>직렬화 필수
    B->>NET: SSE data: type=TEXT_MESSAGE_CONTENT, delta=안녕

    Note over NET: raw bytes (Uint8Array)<br/>Observable 소멸

    NET->>C: Uint8Array 청크 수신
    Note over C: bytes → Observable 복원 필요<br/>프론트엔드 run()의 역할
    C->>C: TextDecoder → SSE 텍스트
    C->>C: SSE 파싱 → JSON.parse()
    C->>C: subscriber.next(event) re-emit
    C->>PP: Observable[BaseEvent] 전달
    Note over PP: transformChunks → verifyEvents<br/>→ apply → subscribers

runAgent() 호출 흐름 — 실제 메서드 실행 순서

runAgent()의 호출 주체는 **Application(프론트엔드 UI)**이다. 호출 이후 내부 파이프라인이 자동으로 실행된다.

sequenceDiagram
    autonumber
    box Frontend
        participant App as "Application (UI)"
        participant AA as "AbstractAgent.runAgent()"
        participant HA as "HttpAgent.run()"
        participant TR as "transformHttpEventStream()"
        participant PP as Pipeline
        participant SUB as Subscribers
    end
    box Backend
        participant B as "BackendAgent.run()"
    end

    App->>AA: agent.runAgent(parameters)
    AA->>AA: prepareRunAgentInput()
    AA->>HA: this.run(input)
    HA->>B: HTTP POST + RunAgentInput (JSON body)
    B->>B: LLM 호출 → BaseEvent emit
    B-->>HA: SSE raw bytes
    HA->>TR: transformHttpEventStream()
    TR->>TR: SSE 파싱 → JSON.parse()
    TR-->>AA: Observable[BaseEvent]
    AA->>PP: transformChunks → verifyEvents → apply
    PP-->>SUB: onMessagesChanged / onStateChanged
    SUB-->>App: UI 업데이트

RunAgentInput — 누가 어떤 값을 넣는가

Frontend 관점 — AbstractAgent가 자동으로 조립

prepareRunAgentInput()이 runAgent() 호출 시 자동으로 구성한다. 개발자가 직접 채울 필요가 없다.

// agent.ts - prepareRunAgentInput()
{
  threadId:      this.threadId,               // AbstractAgent가 생성·관리 (uuid)
  runId:         uuidv4(),                    // 매 실행마다 새로 생성
  messages:      this.messages,               // 누적된 대화 이력 (자동 관리)
  state:         this.state,                  // 현재 공유 상태 (자동 관리)
  tools:         parameters?.tools,           // runAgent() 호출 시 개발자가 전달
  context:       parameters?.context,         // runAgent() 호출 시 개발자가 전달
  forwardedProps: parameters?.forwardedProps, // 추가 props 전달용
}

개발자가 직접 넣는 값은 tools, context, forwardedProps뿐이다:

agent.runAgent({
  tools: [{ name: "search", description: "...", parameters: schema }],
  context: [{ description: "User timezone", value: "Asia/Seoul" }],
})
// threadId, runId, messages, state 는 AbstractAgent가 자동 처리

Backend 관점 — HTTP POST body로 수신, run()의 파라미터로 전달

Backend는 RunAgentInput을 만들지 않고 받는다. Frontend가 조립한 값을 그대로 활용한다.

class BackendAgent extends AbstractAgent {
  run(input: RunAgentInput): Observable<BaseEvent> {
    // input.messages → LLM context (이전 대화 이력)
    // input.tools    → LLM에 넘길 tool 목록
    // input.state    → 현재 공유 상태 참조
    // input.threadId → 세션 추적용
    const response = await llm.chat({
      messages: input.messages,
      tools: input.tools,
    })
    ...
  }
}

필드	생성 주체	Frontend 역할	Backend 역할
threadId	AbstractAgent (자동)	세션 식별	로그·추적용
runId	AbstractAgent (자동)	실행 식별	로그·추적용
messages	AbstractAgent (자동 누적)	대화 이력 전송	LLM context로 사용
state	AbstractAgent (자동 관리)	현재 상태 전송	상태 참조·분기
tools	개발자 → runAgent()	tool 목록 전달	LLM에 tool 등록
context	개발자 → runAgent()	부가 정보 전달	프롬프트 보강

Subscribers — 상태 변화 구독

raw 이벤트가 아닌 처리 후 결과를 받는 콜백 시스템이다.

agent.subscribe({
  onMessagesChanged:  ({ messages }) => { /* 새 메시지 수신 시 */ },
  onStateChanged:     ({ state })    => { /* 상태 변경 시 */ },
  onRunFinishedEvent: ({ result })   => { /* 실행 완료 시 */ },
  onRunFailed:        ({ error })    => { /* 에러 발생 시 */ },
})

raw BaseEvent를 직접 보려면 run(input).subscribe(event => ...) 사용.

참고 문서

• AG-UI Architecture
• AG-UI Middleware
• agent.ts 소스