이전 기사에서 저는 하나의 에이전트가 모든 것을 수행하기를 기대하는 대신, 문제를 독립적인 책임 단위로 나누었습니다.

만약 모든 에이전트가 서로 독립적이라면, 그들은 어떻게 서로 통신할까요? 다음에 무엇을 실행할지는 누가 결정할까요?

그 지점에서 오케스트레이터 에이전트 (Orchestrator Agent)가 교통 관제사처럼 작동하여, 에이전트들이 서로에게 컨텍스트 (context)를 전달하는 방식을 관리하고 실행 순서를 유지합니다.

linkedin-agent/
├── agents/
│   ├── orchestrator.py     
...

각 에이전트는 자신만의 파일을 가집니다. 코어 (core) 폴더에는 다음이 포함되어 있습니다:

• llm.py: 단일 call_llm() 함수를 통해 Ollama와의 모든 상호작용을 래핑 (wrap) 합니다.
• models.py: 에이전트 전반에 걸쳐 사용되는 공유 Pydantic 스키마 (schemas)를 저장합니다.
• utils.py: 사용자가 입력한 토픽 (topic)에서 불필요한 공백을 제거하는 clean_topic() 함수를 포함합니다.
• main.py: 에이전트 파이프라인 (pipeline)을 실행하며, api.py는 FastAPI 서버를 가집니다.

오케스트레이터 에이전트 (Orchestrator Agent)의 임무는 단 하나입니다. 토픽을 이해하고, 이를 분류하며, 최적의 콘텐츠 관점 (angle)을 결정한 뒤, 리서치 에이전트 (Research Agent)를 위한 구조화된 출력 (structured output)을 반환하는 것입니다.

우리가 수행하는 모든 LLM 호출에 대해, 에이전트가 어떻게 작동해야 하는지를 정의하기 위해 작성하는 프롬프트인 SYSTEM_PROMPT와, 이전 에이전트들로부터 모든 컨텍스트를 가져온 user_message가 필요합니다.

그럼, 오케스트레이터 에이전트 (Orchestrator Agent)를 위한 SYSTEM_PROMPT를 작성하는 것부터 시작해 봅시다.

SYSTEM_PROMPT = """
You are the orchestrator agent for WanderMesh.

...

가독성을 위해 프롬프트를 축약했습니다. 프로덕션 (production) 버전에는 분류 일관성을 높이기 위해 도메인 특화 카테고리, 출력 제약 조건, 그리고 퓨샷 (few-shot) 예시도 포함되어 있습니다.

모델에게 개방형 질문을 던지는 대신, 출력을 고정된 구조로 제한했습니다. 이를 통해 다운스트림 (downstream) 파이프라인을 훨씬 더 예측 가능하게 만들었습니다.

초기에 겪었던 문제 중 하나는 모델이 항상 유효한 JSON을 반환하지는 않는다는 점이었습니다. 때로는 응답을 마크다운 코드 펜스 (markdown code fences)로 감싸기도 하고, 때로는 끝에 쉼표 (trailing commas)를 생성하거나, 가끔은 표준 JSON 따옴표 대신 스마트 따옴표 (smart quotes)를 반환하기도 했습니다.

모든 다운스트림 에이전트 (downstream agent)가 유효한 JSON을 기대했기 때문에, 응답을 파싱 (parsing)하기 전에 전처리 단계가 필요했습니다.

gemma2:2b와 같은 작은 로컬 모델 (local models)로 테스트하는 동안, 유효한 JSON을 명시적으로 요청했음에도 불구하고 가끔 포맷팅 지침을 무시하고 잘못된 형식의 JSON을 반환한다는 것을 발견했습니다.

JSON 형식을 수정하는 데 도움을 주기 위해 strip_json_fences() 함수를 작성했습니다.

만약 아직 Python에 익숙해지는 과정에 있다면, CS50 Python을 강력히 추천합니다. 저는 먼저 노트를 읽고 나서 모든 것을 직접 구현해 봄으로써 많은 것을 배웠습니다.

def strip_json_fences(raw:str) -> str:
    raw = raw.strip()
    if raw.startswith("```json"):
        raw = raw.removeprefix("```json").strip()
    elif raw.startswith("```"):
        raw = raw.removeprefix("```").strip()
    if raw.endswith("```"):
        raw = raw.removesuffix("```").strip()
    raw = re.sub(r',\s*]', ']', raw)
    raw = re.sub(r',\s*}', '}', raw)
    raw = raw.replace('\u201c', '\\"').replace('\u201d', '\\"')
...

유효한 JSON만으로는 충분하지 않습니다. 모델이 필드를 누락하거나, 이름을 바꾸거나, 예상치 못한 구조를 반환할 수도 있습니다. 출력을 다음 에이전트에게 전달하기 전에, 파이프라인 (pipeline)이 기대하는 것과 정확히 일치하는지 검증 (validate)하고 싶었습니다.

추가 읽을거리: 만약 Pydantic이 처음이라면, Models와 Fields에 대한 공식 문서는 여기서 사용된 검증 기능을 이해하기에 매우 좋은 자료입니다.

이제 오케스트레이터 에이전트 (Orchestrator Agent)를 위한 Pydantic 객체 (Object)를 만들어 보겠습니다. 지금은 가능한 값들을 제한하지 않고 구조만 검증합니다. 이렇게 하면 파이프라인의 나머지 부분이 여전히 진화하는 동안 오케스트레이터의 유연성을 유지할 수 있습니다.

class OrchestratorOutput(BaseModel):
    domain: str
    angle: str
...

이제 주제를 분류하기 위한 메인 함수를 작성해 보겠습니다. 코드를 작성하기 전에, 이 함수가 기본적으로 무엇을 수행해야 하는지 분석해 보겠습니다.

Topic
   │
   ▼
...

이 함수는 주제(topic)를 입력으로 받아, Research Agent (리서치 에이전트)가 사용할 수 있도록 Pydantic Object (Pydantic 객체)를 출력으로 반환해야 합니다.

먼저, clean_topic()을 사용하여 주제를 정제해야 합니다. 이 과정에서 불필요한 공백을 제거하고 사용자가 빈 문자열을 입력하지 않았는지 확인합니다.

그 다음, 정제된 주제와 SYSTEM_PROMPT를 call_llm() 함수에 전달한 뒤, strip_json_fences() 함수를 사용하여 JSON 펜스(fences)를 제거함으로써 유효한 JSON을 반환하도록 합니다.

마지막으로, JSON을 파싱(parsing)하고 검증(validation)합니다. 만약 오류가 발생하면, 함수는 잘못된 데이터를 파이프라인 하단으로 전달하는 대신 None을 반환합니다.

표준적인 관례에 따라, 무언가 잘못되었을 때 오류를 처리할 수 있도록 try 및 except 블록 내에서 JSON을 파싱하고 검증합니다.

def classify_topic(topic:str) -> OrchestratorOutput|None:
    cleaned_topic = clean_topic(topic)
    raw_response = call_llm(SYSTEM_PROMPT, cleaned_topic)
...

이 시점에서 오케스트레이터(orchestrator)는 일관되게 주제를 분류할 수 있으며, 모든 다운스트림 에이전트(downstream agent)가 신뢰할 수 있는 구조화된 출력(structured output)을 반환할 수 있습니다.

다음 글에서는 콘텐츠가 생성되기 전 사실 정보를 수집하는 역할을 담당하는 컴포넌트인 Research Agent (리서치 에이전트)를 구축해 보겠습니다.

Github Repo: https://github.com/Manav-N4/linkedin-agent#linkedin-agent