nova tech
Other

LLM to Agentic 1) - LLM 단일 모델에서 ReAct까지

박재연·2026. 7. 17.·조회 11

배경

OpenManus 오픈소스를 기반으로 프로젝트를 개선하고 프론트엔드 작업에 집중한 지 2~3일 만에, 에이전트 개발은 기존의 REST API, CRUD 중심 웹 개발과는 본질적으로 다른 영역이라는 것을 체감했습니다.

Manus나 AI Scientist 같은 에이전트는 단순히 LLM과 메시지를 주고받는 서비스가 아닙니다. 샌드박스 환경에서 CodeAct 등의 기법을 활용해 코드를 생성·실행하고, 브라우저와 도구를 사용하며, 목표를 달성할 때까지 여러 단계의 작업을 반복합니다. 모델의 컨텍스트와 출력 토큰 제한에 맞춘 분할 처리도 필요하기 때문에 작업 시간이 길어질 수밖에 없습니다.

이에 따라 프론트엔드의 역할도 달라집니다. 단순한 채팅 UI를 넘어 첨부파일, 도구 호출, 코드 실행 결과, 현재 진행 단계와 대기 상태 등 에이전트의 중간 과정을 사용자에게 이해하기 쉽게 보여줘야 합니다. 이는 부가적인 최적화가 아니라, 에이전트 제품의 신뢰도와 사용성을 결정하는 핵심 UI/UX 요소입니다.

결국 단순 LLM에서 샌드박스 기반 에이전트로 발전하는 과정에는 모델뿐 아니라 프론트엔드, 백엔드 인프라, 통신 프로토콜, 요청·응답 스키마의 변화가 함께 있었습니다. 이러한 중간 과정을 건너뛴 채 현재의 에이전트 구조를 이해하려는 것은 덧셈과 뺄셈만 익힌 상태에서 미적분 문제를 마주하는 것과 비슷하다고 느꼈습니다.

그래서 2022년경의 단순 텍스트 요청·응답 방식부터 멀티턴 대화, 스트리밍, 도구 호출, 샌드박스 실행, 자율 에이전트로 이어지는 흐름을 작은 예제 프로젝트로 직접 구현해 보며 기록하려고 합니다. 이 과정이 정답인지는 아직 모르지만, 현재의 에이전트 시스템을 더 정확히 이해하기 위한 출발점이 될 것이라 생각합니다.

1) LLM 단발

codex, claude code, manus 등 우리가 사용하는 모든 AI들은 하네스가 적용된거라고 보면된다.<br/>모든 모델의 기초는 { role:str, content:str }로 이루어진 메세지 배열이다.

from openai import OpenAI
from fastapi import FastAPI
from pydantic import BaseModel

client = OpenAI()
app = FastAPI(title="Manus Clone Backend - 1단계 (LLM)")

class ChatRequest(BaseModel):  
    message: str  
  
class ChatResponse(BaseModel):  
    role: str  
    content: str
    
@app.post('/chat', response_model=ChatResponse)
async def chat(req:ChatRequest):
	messages = [
		{ role: 'system', content:'당신은 친절한 한국어 비서입니다.'},
		{ role: 'user', content:req.message }
	]
	
	result = client.chat.completions.create(  
	    model="gpt-4o-mini",  
	    messages=messages,  
	)
	
	reply = completion.choices[0].message.content
	
	return ChatResponse(role="assistant", content=reply or "")

단순히 API를 사용만 하면 일회성으로 요청을 보내고 대화가 이어지지 않는다.

Front에서의 상태/액션/API

type status = 'idle' | 'sending' | 'error'
type action = 'change_input' | 'send' | 'reject' | 'resolve'

const messages = []

async function sendChat(message:string){
	const res = await axios.post('/chat', { message })
	return res.data as { role: string, content: string }
}
  • default status =idle
  • change_input → send (idle → sending) / sending 일땐 채팅입력 불가.
    • reject → status = error
    • resolve → status = idle

1.1) 멀티턴

AI와 대화 문맥을 이어나갈려면 어떻게 해야하냐?

매번 대화 전체를 LLM에 보내야한다. 간단한 질문을 해도 context window에 쌓여있는 만큼 토큰값을 청구하는 이유다.

...

class Message(BaseModel):
	role: str
	content: str

class ChatRequest(BaseModel):
	messages: list[Messages]

SYSTEM_PROMPT = { role: 'system', content:'당신은 친절한 한국어 비서입니다.'}

@app.post('chat', response_model=ChatResponse)
async def chat(req: ChatRequest):
	llm_messages = [SYSTEM_PROMPT] + [m.model_dump() for m in req.messages]
	
	result = client.chat.completions.create(  
	    model="gpt-4o-mini",  
	    messages=messages,  
	)
	
	reply = completion.choices[0].message.content # 안녕하세요
	
	return ChatResponse(role="assistant", content=reply or "")

2) SSE + StreamingResponse

JSONResponse:
함수 실행 완료
→ 객체를 JSON으로 직렬화
→ Content-Length 결정
→ 응답 전체 전송

StreamingResponse:
yield 결과 생성
→ 즉시 HTTP chunk 전송
→ 다음 yield 대기
→ 반복 후 연결 종료

JSONResponse를 사용하면 작업 중간에 이벤트나 로그를 남겨도 모든 작업이 완료되기전에 프론트에 실시간으로 보여줄 수 없다. AI가 작업하는 과정을 보여주는 것은 에이전트 제품의 신뢰도와 사용성을 결정하는 핵심 UI/UX 요소의 영역으로 넘어 왔기에 전환을 해야한다.

StreamingResponse을 사용해서 chuck가 만들어 질 때마다 이벤트를 발생시켜 내려 보낼 것이며 사용자가 일방적으로 요청을 보내고 답변을 기다리는 구조이기에 SSE가 가장 적합하다.

from fastapi.responses import StreamingResponse

SYSTEM_PROMPT = { role: 'system', content:'당신은 친절한 한국어 비서입니다.'}

@app.post('/chat/stream')
async def chat(req: ChatRequest) -> StreamingResponse :
	llm_messages = [SYSTEM_PROMPT] + [m.model_dump() for m in req.messages]
	
	def event_generator():
		stream = client.chat.completions.create(
			model="gpt-4o-mini",
			messages=llm_messages,
			stream=True
		)
		ai_answer = ''
		for chuck in stream:
			delta = chuck.choices[0].delta.content # 안,녕하,세요
			if delta:
				ai_answer += delta
				yield f"data: {json.dumps({'delta': delta}, ensure_ascii=False)}\n\n"
		def 메세지추가(ai_answer)
		yield f"data: {json.dumps({'done': True })}\n\n"
	
	return StreamingResponse( 
		event_generator(), 
		media_type='text/event-stream', 
		header: {"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}
)

프론트엔드에서는 어디가 끝인지 알 수 없기 때문에 {'done': True }으로 끝 신호를 줘야한다. 그리고 StreamingResponse을 사용함으로써 디버깅이랑 프론트도 복잡도가 오른다.

서버에서 JSON을 완성시켜 yield 를 발생시켜 전송해줘도 서버의 yield 경계와 브라우저의 reader.read() 경계는 일치하지 않는다.

data: {"delta":"안"}\n\n
data: {"delta":"녕하세요\n"}\n\n
data: {"delta":"오늘"}\n\n
data: {"delta":"은 목요일"}\n\n
data: {"delta":"입니다."}\n\n
data: {"done":true}\n\n

브라우저까지 전달되는데 여러 계층들이 있고 그로 인해 나눠서 온다.

read 1: data: {"delta":"안"}\n\ndata: {"delta":"녕
read 2: 하세요.\n"}\n\ndata: {"delta":"오늘
read 3: 은 목"}\n\ndata: {"delta":"요일
read 4: 입니다."}\n\ndata: {"done":tr
read 5: ue}\n\n

type status = 'idle' | 'sending' | 'streaming' | 'error'
type action =  'input_change' | 'send' | 'delta' | 'done' | 'reject'
type Message = { role: 'user' | 'assistant', content: string}

async function sendChat(messages: Message[], onDelta: (delta:string)=> void){
	const res = await axios.post(
		'/chat/stream', 
		{ messages }, 
		{ adapter: "fetch", responseType: "stream"}
	)
	
	const reader = res.data.getReader()
	const decoder = new TextDecoder()
	let buffer = ""
	
	while(True){
		const {value, done} = await reader.read()
		if(done) break;
		
		buffer += decoder.decode(value, {stream: true})
		const parts = buffer.split("\n\n");
		buffer = parts.pop() ?? "";
		
		for (const part of parts) {  
		    const line = part.replace(/^data: /, "").trim();  
		    if (!line) continue;  
		    const payload = JSON.parse(line);  
		    if (payload.error) throw new Error(payload.error);  
		    if (payload.done) return;  
		    if (payload.delta) onDelta(payload.delta);  
		}
	}
}

parts의 마지막 배열의 경우의 수는 2가지이다.

  1. 마지막이 \n\n로 떨어졌을 때
  2. 완성되지않은 데이터

고로 buffer.split("\n\n")를 진행하면 마지막 배열값은 완성되지않은 데이터 이거나 빈 문자열이 남게 된다. 그 후 buffer = parts.pop() ?? "";을 진행하면 buffer에는 완성되지않은 값이나 빈 문자열, parts에는 완성된 데이터 리스트들이 들어가게 되고 상태값에 따라 처리한다.

2.5) DB 영속화

현재 채팅 메세지는 클라이언트에서 상태 관리 중이라 새로고침하면 없어진다. 이걸 서버로 옮기기 위해 대화방, 대화방과 매칭되는 메세지를 DB에 넣어서 보관한다.

기존 방의 경우 대화방ID값을 같이 올려보내면 메세지 전체를 DB에서 가져오고 새로운 메세지를 추가하고 LLM에 요청한다. 그 후에 유저메세지와 답변 완료된 LLM 메세지를 갱신하고 답변을 내려준다. 새로운 방의 경우 대화방을 생성하고 ID값 가져와서 동일하게 진행.

그리고 yield로 맨 먼저 conversation_id 값을 응답한다.

if (payload.conversation_id) onCid(payload.conversation_id);

프론트에서는 조건문을 추가하고 url을 /app/new → /app/:cid 로 URL 이동

3) 파일 업로드/다운로드

단순 텍스트를 주고 받는 단계에서 첨부 파일를 서버에 올리고 그걸 LLM이 인식하는 단계다.

from fastapi.responses import StreamingResponse

client = FastAPI()

class ChatRequest:
	message: str
	conversation_id: str | None = None
	file_ids: list[str] = []
	
SYSTEM_PROMPT = { role: 'system', content:'당신은 친절한 한국어 비서입니다.'}	

def event_generator(cid, is_new, llm_messages) : 
	yield f"data: {json.dumps({'conversation_id': cid, 'is_new': is_new})}\n\n"
	...

@app.post('/chat/stream')
async def chat(req:ChatRequest) -> StreamingResponse :
	cid = req.conversation_id
	if cid is None :
		cid = def 방생성 -> cid

	context = req.messages
	
	if file_ids:
		names = []
		for fid in file_ids:
			meta = get_file(fid)
			names.append(meta['name'])
		context = f"['첨부파일: {', '.join(names)}']\n{content}"
	
	add_message(cid,context)
	chat_history = get_messages(cid)
	llm_messages = [SYSTEM_PROMPT] + chat_history
		
	return StreamingResponse( 
		event_generator(cid, is_new, llm_messages), 
		media_type='text/event-stream', 
		header: {"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}

LLM에 넣는 context 내에 첨부파일 이렇게 있다고 넣어주는 식이다.

지금 주는 파일 이름 뭐야?

이렇게 물으면 파일이름은 말해줄 수 있지만 읽지는 못하는 상태다. 이걸 읽을려면 도구를 붙여야한다.

그리고 시작하자마자 is_new, conversation_id 값을 내려줘서 프론트에서 URL 이동처리 진행한다.

파일업로드

프론트에서 파일첨부 선택
백엔드로 바로 업로드 (파일 크기제한) -> id, fid

-> 파일업로드를 하면 해당파일과 매칭되는 ID값이 생긴다.

파일 다운로드

1 [브라우저]
fid를 프론트에서 서버로 요청

2 [서버]
get_file(fid)
만료된 토큰 제거
token 생성 -> {"url": f"/files/download?token={token}"}

download_tokens = {
    "ba171c15-a46e-4215-aa1b-c8444cb27fe9"   ← 키: 랜덤 uuid 토큰(str)
        : DownloadTicket(                     ← 값: 네임드튜플 한 개
            file_meta = {                     ← ① get_file이 준 dict 통째로
                'id': '9f3a...',
                'name': '보고서.pdf',
                'mime': 'application/pdf',
                'size': 20480,
                'conversation_id': 'c1...'
            },
            expires_at = 1784091240.836341    ← ② 만료 '절대 시각'(unix time, 지금+60)
        )
}

3 [브라우저]
응답 받은 url로 요청

4 [서버]
download_tokens.pop(token)
유효시간 체크
return file

Redis를 써서 ttl 핸들링해도 되지만 일단은 간단하게 처리 + S3로 전환할 때 공수가 최대한 없게하기 위해 S3에서 file을 다운받을때 패턴을 그대로 가져와 썼다.

마지막 4가 없어지고 S3로 변환될 예정이다.

S3 presigned URL은 다음 두 요구를 동시에 해결하려고 생긴 패턴입니다.

  1. S3 파일은 비공개로 유지해야 함
  2. 파일 데이터는 애플리케이션 서버를 거치지 않고 S3와 브라우저가 직접 주고받아야 함

버킷과 객체가 비공개면 URL을 가지고 있어도 접근이 안된다. 반대로 Public으로 설정하면 URL을 아는 사람은 누구나 받을 수 있다. presigned URL는 Public으로 둔 채 임시 출입증을 만드는 방식이다. 그래서 유효기간을 짧게 설정해놔야한다.

2번 문제도 해결된다.

[일반적인 구조]
브라우저
   ↓ 다운로드 요청
FastAPI
   ↓ S3에서 파일 읽기
S3
   ↓ 파일 데이터
FastAPI
   ↓ 파일 데이터
브라우저


[S3 presigned URL]
브라우저
   ↓ 인증 요청
FastAPI
   ↓ 사용자와 파일 권한 확인
짧게 만료되는 S3 서명 URL 발급
   ↓
브라우저 ───── 직접 다운로드 ───── S3

3.5) ReAct

이제 전달한 파일을 LLM이 인식한 상태에서 해당 파일을 단순히 읽어보는 도구를 추가해볼거다. 단순 도구 추가가 아니라 ReAct(think → act → observe)를 적용할거다.

messages = history + [currnent_prompt]

1. [think]
messages와 Tools을 LLM에게 전달
LLM이 해당 프롬프트가 Tools을 필요로 하는지 판단하고 답변

도구가 필요없다면 바로 종료

-> [첨부파일: document.md]\n이 파일 보고 Novaair가 뭐하는 회산지 요약해줘.
-> tool : search_files

2. [act]
도구가 만약 필요하다면 해당 도구(미리 작성해 놓은 python 코드) 실행
시작할때 완료후 결과를 사용자에게 미리 전달
사용한 도구와 결과를 observe에 전달

-> search_files.py run() 실행 -> document.md 위치 파악

3. [observe]
messages에 {role:'assistant'}가 이런 도구를 썻고 결과가 어땟는지 추가

-> messages에 내용 추가

4. [think]
수정된 메세지로 다시 루프 시작 tool: read_files.py run()로 파일 읽고 다시 루프

-> 추가된 내용보고 요약된 결과 출력

주의점은 무한루프를 도는 걸 막기위해 Limit를 걸어줘야한다

프론트에선 아래처럼 대기하다가 도구 시작과 결과가 오면 랜더링해준다.

if (payload.tool_call) onTool(도구 호출)
if (payload.tool_result) onTool(도구 결과)

마치며

단순 단발로 대화를 주고 받는 구조에서 ReAct를 적용해서 루프돌면서 LLM이 추론 → 액션 → 다음단계 결정 까지 진행해봤고 챕터1 마무리 하겠습니다.

댓글 0