본문으로 건너뛰기

07_Langgraph 기반 방송 워크플로우 설계

· 약 6분
sbin
sbin
SceneMakerAI 팀

들어가며​

자연어 질의로 방송 클립 구간을 찾는 LangGraph 과정을 구체적으로 명시한 글이다.

clips[] JSON 으로 정형화하는 과정을 보여준다.

검색 과정을 가장 쉽게 요약하면 아래와 같다.

  1. 어느 회차인지 (scope)
  2. 뭘 찾을지 목록 짜기 (plan)
  3. 비슷한 조각 긁어오기 (retrieve)
  4. 앞뒤도 같이 보기 (expand)
  5. 어디~어디 자를지 확정 (select)
  6. 합치고 길이 맞추기 (assemble)

agent-search 한 곳에서 LangGraph 고정 그래프를 돌린다.

코드 구현은 아래와 같다.

# agent-search/lib/graph/graph.py

ainvoke(request)
 ├ scope      회차 선별 (v_ids 지정 or 투표+LLM)
 ├ plan       route + beats (회차별)
 ├ retrieve   multi-query 벡터 검색
 ├ expand     히트 ±12초 맥락
 ├ select     start~end 확정 (아니면 재질의 1회)
 └ assemble   겹침 병합 · budget · clips[] JSON

요청은 POST /api/v1/search 로 들어온다.

service 는 compilation / shortform / trailer / ad_slot 까지만 구축 진행 중이다.

1. scope — 어느 회차인지​
  • 시리즈물이라 먼저 “몇 화 얘기냐” 를 정한다.
  • v_ids 가 있으면 그대로 쓴다. 없으면 chapter 벡터 투표 → LLM 확정.
SCOPE_MAX_VIDEOS = 5      # 회차 선별 상한
SCOPE_VOTE_LIMIT = 40     # 투표용 챕터 검색 top-k

# lib/graph/scope.py

async def run(state: dict, deps) -> dict:
    req = state["request"]

    # ① 사용자가 회차를 지정 → 그대로
    if req.get("v_ids"):
        videos = await asyncio.to_thread(db.select_videos, req["v_ids"])
        scope = {"mode": "user", "reason": "요청에 v_ids 지정", "votes": []}

    # ② 자동 선별: chapter 투표 → LLM 확정
    else:
        qvec = (await asyncio.to_thread(embed.embed_query, [qtext]))[0]
        votes = await asyncio.to_thread(vdb.vote_videos, deps.vdb, qvec)
        # ... LLM 이 v_ids 확정 (최대 SCOPE_MAX_VIDEOS)
        scope = {"mode": "vote+llm", "reason": ..., "votes": votes}

    return {"videos": videos, "scope": scope}
  • 예시 입출력
입력:  query="한국시리즈 7차전 득점 장면", v_ids=없음

출력:  # mode=vote+llm
{
  "videos": [{"v_id": 1, "name": "코리안시리즈 KIA vs SK", ...}],
  "scope": {
    "mode": "vote+llm",
    "reason": "요청인 '한국시리즈 7차전'과 '득점 장면'이 v_id=1 ...에 부합함",
    "votes": [[1, 12], [4, 7], [3, 7], [5, 6], [6, 4], [2, 4]]
  }
}

합계 40표가 6개 회차에 퍼진다. 1위가 12표뿐이라 투표만으로 자르기 어렵고, LLM 이 1개로 확정한다.

2. plan — 뭘 찾을지 목록 짜기​
  • 회차마다 챕터 타임라인을 보고 검색 단위(beat) 를 짠다.
  • route 두 갈래: pinpoint (명장면 1~2개) / structural (서사에서 여러 클립).
# lib/graph/plan.py

# route:
#   pinpoint   — 명장면 클립, 명대사 카드
#   structural — 회차 요약, 예고편, 인물 하이라이트

# beat 하나:
#   title, queries[2~3], time_hint, want=["segment","dialogue"]
  • 예시 입출력
입력:  v_id=3 겨울연가, preset=감성_몽타주

출력:
{
  "route": "structural",
  "beats": [
    {
      "title": "호숫가 산책과 정류장의 애틋한 시선",
      "queries": [
        "남녀 주인공이 호숫가에서 조용히 산책하며 서로를 바라보는 장면",
        "버스 정류장에서 주변 친구들의 시선을 의식하며 어색하게 마주보는 두 사람",
        "겨울 연가 초반부 설렘이 느껴지는 눈빛 교환 클로즈업"
      ],
      "time_hint": [0, 600],
      "want": ["segment", "dialogue"],
      "v_id": 3
    }
    # ... beat 여러 개
  ]
}

queries 가 2~3개인 이유: 같은 장면을 화면 묘사 / 사건 / 대사 표현으로 넓게 던지기 위해서다. 벡터 검색은 싸니까.

3. retrieve — 비슷한 조각 긁어오기​
  • beat 의 queries 로 Milvus 벡터 검색. LLM 안 씀.
  • multi-query 병합 → 중복 제거 → 상위 SEARCH_TOPK (10).
  • cast 는 soft 필터 (빈 cast 행이 많아서 하드 필터면 recall 이 죽음).
SEARCH_TOPK = 10

# lib/graph/retrieve.py

def search_beat(client, beat: dict, cast_filter: list[str] | None) -> list[dict]:
    queries = beat.get("queries") or [...]
    qvecs = embed.embed_query(queries)
    # soft 필터 계단: [cast+event] → [event] → [cast] → [없음]
    # is_ad 는 후처리로 제외
    return sorted(...)[:SEARCH_TOPK]
  • 예시 입출력
입력:  queries=["남녀 주인공이 호숫가에서 ...", ...]

출력:  candidates (일부)
[
  {
    "ref_type": "segment", "ref_id": 26,
    "start_sec": 150, "end_sec": 156,
    "text": "남녀 두 사람이 호수 옆 산책로를 걷고 있다. 남녀 걷기 여자 멈추기 남자 멈추기",
    "cast": "",
    "distance": 0.6867   # 코사인 유사도 (클수록 유사)
  },
  ...
]
4. expand — 앞뒤도 같이 보기​
  • 6초 히트는 클립이 아니다. 앞뒤 ±12초(세그먼트 2칸)의 segment+dialogue 를 붙인다.
  • select 가 경계를 스냅할 재료.
EXPAND_PAD_SEC = 12   # 히트 앞뒤 확장 폭
_TOP_HITS = 4         # beat 당 확장할 상위 히트 수

# lib/graph/expand.py

def expand_beat(client, beat: dict) -> list[dict]:
    ctxs = []
    for h in beat["candidates"][:_TOP_HITS]:
        s = max(0, h["start_sec"] - EXPAND_PAD_SEC)
        e = h["end_sec"] + EXPAND_PAD_SEC
        ctxs.append({
            "anchor": f"{h['ref_type']}:{h['ref_id']}",
            "rows": vdb.window(client, beat["v_id"], s, e),
        })
    return ctxs
  • 예시 입출력
입력:  hit segment:26 (150~156초)

출력:  context (시간창 rows 일부)
[
  {"ref_type": "segment", "ref_id": 25, "start_sec": 144, "end_sec": 150, "text": "..."},
  {"ref_type": "segment", "ref_id": 26, "start_sec": 150, "end_sec": 156, "text": "남녀 두 사람이 호수 옆..."},
  {"ref_type": "segment", "ref_id": 27, "start_sec": 156, "end_sec": 162, "text": "호수 옆에서 두 명의..."},
]
5. select — 어디~어디 자를지 확정​
  • LLM 이 beat + 후보 + 주변 타임라인을 보고 클립 1개 [start_sec, end_sec] 를 확정한다.
  • 무관하면 usable=false + retry_query → 재검색·재확장 후 1회만 재판정 (REFINE_MAX=1 ).
REFINE_MAX = 1

# lib/graph/select.py

# 출력 JSON:
# {"usable": true|false, "start_sec", "end_sec",
#  "source_refs", "cast", "reason", "caption", "score",
#  "retry_query": "usable=false 일 때만"}
  • 예시 입출력 (성공)
입력:  beat="호숫가 산책과 정류장의 애틋한 시선" + candidates + context

출력:
{
  "usable": true,
  "start_sec": 150, "end_sec": 162,
  "caption": "호숫가 산책 중 멈춰 선 두 사람, 어색하지만 설레는 눈빛 교환",
  "score": 0.92,
  "reason": "호숫가 산책(26)과 직후 시선 교환(27)을 연결한 12초 구간"
}
  • 예시 입출력 (버림)
출력:
{
  "usable": false,
  "score": 0.0,
  "reason": "후보 타임라인에 실제 득점·타격 장면이 없고 해설만 있음",
  "retry_query": "KIA 드루와 안타 장면 / KIA 유케미 안타 장면 / ..."
}

못 찾으면 억지로 채우지 않는다. 그게 영상 RAG 에서 환각을 막는 방법이다.

6. assemble — 합치고 길이 맞추기​
  • usable 클립만 모은다. LLM 안 씀.
  • 같은 회차 겹침 병합 → duration budget(score 우선) → 정렬 → 최종 JSON.
BUDGET_TOLERANCE = 1.15   # target_duration_sec 초과 허용 배율

# lib/graph/assemble.py

# trailer 만 beat 기획순, 나머지는 방영순·시간순
# 출력의 v_id + start_sec/end_sec 가 그대로 다음 단계 ffmpeg 입력
  • 예시 입출력
{
  "service": "compilation",
  "preset": "감성_몽타주",
  "route": "structural",
  "total_duration_sec": 54,
  "clips": [
    {
      "v_id": 3,
      "v_name": "겨울연가",
      "order": 1,
      "beat": "호숫가 산책과 정류장의 애틋한 시선",
      "start_sec": 150,
      "end_sec": 162,
      "start_time": "00:02:30",
      "end_time": "00:02:42",
      "caption": "호숫가 산책 중 멈춰 선 두 사람, 어색하지만 설레는 눈빛 교환",
      "score": 0.92
    }
  ],
  "trace": {"llm_calls": 6, "retries": 0, "dropped_beats": []}
}

노드별 중간 결과는 output/<req_id>/<node>.json 으로 덤프된다.

마무리​

목표였던 검색 Flow 와 clips[] 정형화 과정까지 정리했다.

클립을 실제 영상으로 자르는 단계(ffmpeg Serving)와 서비스 UI 연동은 다음이다.

참고 링크)

본 글은 과학기술정보통신부·정보통신산업진흥원 「2026년 오픈소스 AI·SW 개발·활용 지원사업」의 지원으로 수행된 연구 결과입니다.