기술리서치 종합정리

################################################################################

INIT

################################################################################

CONDA_INIT

layout: post
title: 260602 기술 리서치 종합 정리
comments: true
tags:

• 기술리서치
• 종합정리

• 리서치

0. Executive Summary

이번 문서는 여러 기술 키워드를 “실무 설계 판단” 관점에서 묶어 정리한다. 핵심 축은 다음과 같다.

flowchart LR
  A["멀티테넌트 SaaS"] --> B["비동기 Python 서버"]
  B --> C["Agent Framework"]
  C --> D["Vector DB / RAG"]
  B --> E["STT-LLM-TTS 스트리밍"]
  E --> F["GStreamer / 멀티플렉싱"]
  A --> G["Docker on AWS"]
  H["RCPS 보통주 전환"] --> I["재무/지배구조 판단"]

1. 멀티테넌트

1.1 소개

멀티테넌트는 하나의 소프트웨어/인프라가 여러 고객, 조직, 팀, 사용자 그룹을 동시에 서비스하는 구조다. 여기서 tenant는 단순 사용자 하나가 아니라 “격리와 과금, 권한, 데이터 경계가 필요한 단위”로 이해하는 것이 좋다.

Azure Architecture Center는 테넌트 격리를 완전 공유에서 완전 분리까지의 스펙트럼으로 설명한다. AWS SaaS 문서와 보안 블로그에서는 흔히 Pool, Bridge, Silo 모델을 사용한다.

1.2 핵심 모델

flowchart LR
  P["Pool<br/>공유 앱 + 공유 DB"] --> B["Bridge<br/>공유 앱 + 테넌트별 DB"]
  B --> S["Silo<br/>테넌트별 앱 + DB + 네트워크"]
  P -. "저비용/고밀도" .-> B
  S -. "강격리/고비용" .-> B

1.3 설계 특징

• 모든 요청에는 신뢰 가능한 tenant_id가 있어야 한다.
• 인증의 subject와 tenant membership을 분리해서 검증해야 한다.
• DB Row Level Security, query filter, schema-per-tenant, database-per-tenant 중 하나를 선택한다.
• 캐시 키, 큐 메시지, 로그, 파일 경로에도 tenant 경계를 넣어야 한다.
• 운영 지표는 tenant별로 쪼개야 noisy neighbor를 찾을 수 있다.

1.4 장단점

1.5 실용 패턴

from dataclasses import dataclass

@dataclass(frozen=True)
class TenantContext:
    tenant_id: str
    user_id: str
    roles: set[str]

def build_query(ctx: TenantContext, symbol: str) -> dict:
    return {
        "tenant_id": ctx.tenant_id,
        "symbol": symbol,
    }

실무에서는 이 정도 코드만으로는 부족하다. DB 레벨 제약, API gateway, middleware, 로그 마스킹, 권한 테스트가 함께 들어가야 한다.

1.6 근거 URL

• Azure 멀티테넌트 격리 모델: https://learn.microsoft.com/en-us/azure/architecture/guide/multitenant/considerations/tenancy-models
• AWS SaaS Lens: https://docs.aws.amazon.com/wellarchitected/latest/saas-lens/saas-lens.html
• AWS 멀티테넌트 보안 모델: https://aws.amazon.com/blogs/security/security-practices-in-aws-multi-tenant-saas-environments/
• Kubernetes 멀티테넌시: https://kubernetes.io/docs/concepts/security/multi-tenancy/
• GKE 엔터프라이즈 멀티테넌시: https://docs.cloud.google.com/kubernetes-engine/docs/best-practices/enterprise-multitenancy

2. Python 비동기 프로그래밍

2.1 소개

Python 비동기 프로그래밍은 async/await 문법과 이벤트 루프를 사용해 I/O 대기 시간을 겹쳐 처리하는 방식이다. asyncio는 네트워크 서버, DB 클라이언트, 분산 큐, 스트리밍 처리의 기반으로 쓰인다.

sequenceDiagram
  participant App
  participant Loop as Event Loop
  participant API1
  participant API2
  App->>Loop: create_task(API1)
  App->>Loop: create_task(API2)
  Loop->>API1: request
  Loop->>API2: request
  API2-->>Loop: response
  API1-->>Loop: response
  Loop-->>App: gather results

2.2 특징

2.3 장단점

2.4 간단 예제

import asyncio

async def fetch_price(symbol: str) -> str:
    await asyncio.sleep(0.2)
    return f"{symbol}=100"

async def main() -> None:
    results = await asyncio.gather(
        fetch_price("AAPL"),
        fetch_price("MSFT"),
        fetch_price("NVDA"),
    )
    print(results)

asyncio.run(main())

2.5 실용 예제: 동시 요청 + timeout + 제한

import asyncio
from collections.abc import Awaitable, Callable

async def bounded_map(
    items: list[str],
    worker: Callable[[str], Awaitable[str]],
    concurrency: int = 5,
    timeout: float = 2.0,
) -> list[str]:
    semaphore = asyncio.Semaphore(concurrency)

    async def run_one(item: str) -> str:
        async with semaphore:
            return await asyncio.wait_for(worker(item), timeout=timeout)

    return await asyncio.gather(*(run_one(item) for item in items))

이 패턴은 외부 API 호출, 종목별 데이터 수집, 멀티테넌트 배치 작업에 자주 쓴다.

2.6 근거 URL

• Python asyncio 공식 문서: https://docs.python.org/3.14/library/asyncio.html
• Python asyncio 개념 문서: https://docs.python.org/3/howto/a-conceptual-overview-of-asyncio.html
• Python event loop 문서: https://docs.python.org/3.15/library/asyncio-eventloop.html

3. LangChain, LangGraph

3.1 소개

LangChain은 LLM 애플리케이션과 agent를 빠르게 만들기 위한 고수준 프레임워크다. LangGraph는 장기 실행, 상태 저장, human-in-the-loop, streaming, durable execution에 초점을 맞춘 저수준 orchestration framework다.

LangChain v1 계열의 agent는 내부적으로 LangGraph 기반 runtime을 사용한다. 즉, 간단한 agent는 LangChain으로 시작하고, 상태 전이/복구/분기 제어가 커지면 LangGraph를 직접 쓰는 식의 흐름이 자연스럽다.

3.2 특징

flowchart TD
  U["User Message"] --> A["LangChain create_agent"]
  A --> M["Model Node"]
  M -->|tool call| T["Tool Node"]
  T --> M
  M -->|final| R["Response"]
  A -. "built on" .-> G["LangGraph Runtime"]

3.3 장단점

3.4 LangChain 간단 예제

from langchain.agents import create_agent

def get_weather(city: str) -> str:
    """Get weather for a given city."""
    return f"{city}: sunny"

agent = create_agent(
    model="openai:gpt-5-mini",
    tools=[get_weather],
    system_prompt="You are a concise assistant.",
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "weather in Seoul?"}]
})

3.5 LangGraph 실용 예제: 간단 상태 그래프

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    text: str
    route: str

def classify(state: State) -> State:
    route = "finance" if "stock" in state["text"].lower() else "general"
    return {"text": state["text"], "route": route}

def answer(state: State) -> State:
    return {"text": f"[{state['route']}] {state['text']}", "route": state["route"]}

graph = StateGraph(State)
graph.add_node("classify", classify)
graph.add_node("answer", answer)
graph.add_edge(START, "classify")
graph.add_edge("classify", "answer")
graph.add_edge("answer", END)
app = graph.compile()

print(app.invoke({"text": "stock news", "route": ""}))

3.6 근거 URL

• LangChain overview: https://docs.langchain.com/oss/python/langchain/overview
• LangChain agents: https://docs.langchain.com/oss/python/langchain/agents
• LangGraph overview: https://docs.langchain.com/oss/python/langgraph
• LangGraph persistence: https://langchain-5e9cc07a.mintlify.app/oss/javascript/langgraph/persistence

4. PydanticAI

4.1 소개

PydanticAI는 Pydantic 팀이 만든 Python agent framework다. FastAPI가 타입 힌트와 Pydantic validation으로 웹 API 개발 경험을 단순화한 것처럼, PydanticAI는 GenAI/agent 개발에서 타입, 의존성, structured output, tool validation을 중심에 둔다.

4.2 특징

flowchart LR
  A["Agent"] --> B["Instructions"]
  A --> C["Tools"]
  A --> D["Deps via RunContext"]
  A --> E["Structured Output"]
  E --> F["Pydantic Validation"]

4.3 장단점

4.4 간단 예제

from pydantic_ai import Agent

agent = Agent(
    "openai:gpt-5-mini",
    instructions="Answer in Korean. Be concise.",
)

result = agent.run_sync("PydanticAI가 뭐야?")
print(result.output)

4.5 실용 예제: 멀티테넌트 의존성 + structured output

from dataclasses import dataclass
from pydantic import BaseModel
from pydantic_ai import Agent, RunContext

@dataclass
class Deps:
    tenant_id: str
    plan: str

class StockAnswer(BaseModel):
    summary: str
    risk_level: str
    tickers: list[str]

agent = Agent[Deps, StockAnswer](
    "openai:gpt-5-mini",
    deps_type=Deps,
    output_type=StockAnswer,
)

@agent.instructions
def add_tenant_policy(ctx: RunContext[Deps]) -> str:
    return f"Tenant={ctx.deps.tenant_id}, plan={ctx.deps.plan}. Respect tenant policy."

result = agent.run_sync(
    "NVDA와 MSFT를 짧게 비교해줘.",
    deps=Deps(tenant_id="tenant-a", plan="pro"),
)
print(result.output)

4.6 근거 URL

• PydanticAI overview: https://pydantic.dev/docs/ai/overview/
• PydanticAI agent: https://pydantic.dev/docs/ai/core-concepts/agent/
• PydanticAI output: https://pydantic.dev/docs/ai/core-concepts/output/
• PydanticAI dependencies: https://pydantic.dev/docs/ai/core-concepts/dependencies/
• PydanticAI graph: https://pydantic.dev/docs/ai/graph/graph/

5. LangChain/LangGraph vs PydanticAI 비교분석

5.1 핵심 비교

quadrantChart
  title Agent Framework Positioning
  x-axis "간단한 앱" --> "복잡한 orchestration"
  y-axis "느슨한 타입" --> "강한 타입"
  quadrant-1 "Typed Orchestration"
  quadrant-2 "Typed App"
  quadrant-3 "Fast PoC"
  quadrant-4 "Runtime Control"
  "LangChain": [0.35, 0.45]
  "LangGraph": [0.85, 0.55]
  "PydanticAI": [0.45, 0.85]

5.2 선택 가이드

5.3 실무 조합

PydanticAI와 LangGraph는 경쟁만 하는 관계가 아니다. 예를 들어 LangGraph node 안에서 PydanticAI agent를 호출하면 “상태 orchestration은 LangGraph, typed agent는 PydanticAI”로 역할을 나눌 수 있다.

def pydantic_agent_node(state: dict) -> dict:
    result = agent.run_sync(state["question"], deps=state["deps"])
    return {"answer": result.output.model_dump()}

6. VectorDB in MongoDB

6.1 소개

MongoDB Atlas Vector Search는 MongoDB 문서 컬렉션에 embedding vector를 저장하고, $vectorSearch aggregation stage로 semantic search를 수행하는 기능이다. 운영 데이터와 vector embedding을 같은 document model 안에서 다룰 수 있다는 점이 핵심이다.

6.2 특징

flowchart LR
  D["Documents"] --> E["Embedding Model"]
  E --> M["MongoDB Collection<br/>text + metadata + embedding"]
  Q["User Query"] --> QE["Query Embedding"]
  QE --> V["$vectorSearch"]
  M --> V
  V --> R["Top-k Documents"]
  R --> L["LLM Answer"]

6.3 장단점

6.4 간단 예제

db.articles.aggregate([
  {
    $vectorSearch: {
      index: "article_embedding_index",
      path: "embedding",
      queryVector: [0.12, 0.34, 0.56],
      numCandidates: 100,
      limit: 5
    }
  },
  {
    $project: {
      title: 1,
      score: { $meta: "vectorSearchScore" }
    }
  }
])

6.5 실용 예제: 멀티테넌트 RAG 검색

db.knowledge.aggregate([
  {
    $vectorSearch: {
      index: "knowledge_vector_index",
      path: "embedding",
      queryVector: queryEmbedding,
      numCandidates: 200,
      limit: 8,
      filter: {
        tenant_id: "tenant-a",
        visibility: "internal"
      }
    }
  },
  {
    $project: {
      _id: 0,
      title: 1,
      content: 1,
      tenant_id: 1,
      score: { $meta: "vectorSearchScore" }
    }
  }
])

6.6 근거 URL

• MongoDB Vector Search overview: https://www.mongodb.com/docs/vector-search/
• MongoDB Vector Search index: https://www.mongodb.com/docs/vector-search/index/vector-search-type/
• MongoDB Vector Search getting started: https://www.mongodb.com/products/platform/atlas-vector-search/getting-started

7. VectorDB in other SQL DB

7.1 소개

SQL DB 안의 vector search는 “정형 데이터와 embedding을 같은 SQL 질의 체계에서 다루는 방식”이다. PostgreSQL pgvector, SQL Server VECTOR, Oracle AI Vector Search, MariaDB Vector, MySQL HeatWave Vector Store, DuckDB VSS 등이 대표적이다.

7.2 주요 선택지

flowchart TD
  A["SQL Vector Search"] --> P["Postgres pgvector"]
  A --> S["SQL Server VECTOR"]
  A --> O["Oracle AI Vector Search"]
  A --> M["MariaDB Vector"]
  A --> H["MySQL HeatWave"]
  A --> D["DuckDB VSS"]

7.3 장단점

7.4 PostgreSQL + pgvector 간단 예제

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE documents (
  id bigserial PRIMARY KEY,
  tenant_id text NOT NULL,
  content text NOT NULL,
  embedding vector(3) NOT NULL
);

CREATE INDEX ON documents
USING hnsw (embedding vector_cosine_ops);

SELECT id, content
FROM documents
WHERE tenant_id = 'tenant-a'
ORDER BY embedding <=> '[0.1,0.2,0.3]'
LIMIT 5;

7.5 실용 예제: hybrid SQL + vector

SELECT id, title, published_at
FROM articles
WHERE tenant_id = $1
  AND published_at >= now() - interval '90 days'
ORDER BY embedding <=> $2
LIMIT 10;

실무에서는 tenant_id, published_at, category 같은 정형 필터를 먼저 좁히고 vector search를 수행하는 구조가 비용과 품질 면에서 유리하다.

7.6 근거 URL

• pgvector GitHub: https://github.com/pgvector/pgvector
• SQL Server vector type: https://learn.microsoft.com/en-us/sql/t-sql/data-types/vector-data-type
• Oracle AI Vector Search: https://blogs.oracle.com/database/oracle-announces-general-availability-of-ai-vector-search-in-oracle-database-23ai
• MariaDB Vector: https://mariadb.com/docs/server/reference/sql-structure/vectors/vector-overview
• MySQL HeatWave Vector Store: https://dev.mysql.com/doc/heatwave/en/mys-hw-genai-vector-store-overview.html
• DuckDB VSS: https://duckdb.org/docs/current/core_extensions/vss.html

8. L2 Distance vs Cosine Distance

8.1 소개

L2 distance와 cosine distance는 vector search, embedding retrieval, clustering, recommendation에서 가장 자주 비교되는 거리 지표다. 둘 다 “가까운 벡터”를 찾지만, 무엇을 가깝다고 보는지가 다르다.

flowchart LR
  A["Embedding Vectors"] --> N{"Normalize?"}
  N -->|Yes| C["Cosine / Inner Product<br/>방향 중심"]
  N -->|No| L["L2 Distance<br/>크기 + 방향"]
  C --> R["Semantic Similarity"]
  L --> M["Magnitude-sensitive Matching"]

8.2 수식

두 벡터 $x, y$에 대해 L2 distance는 다음과 같다.

Cosine similarity와 cosine distance는 다음과 같다.

즉, 정규화된 벡터에서는 L2 distance와 cosine distance가 같은 순위를 만들 수 있다. 반대로 정규화하지 않은 벡터에서는 L2가 벡터 크기의 영향을 강하게 받는다.

8.3 장단점

8.4 언제 L2를 쓰는가?

8.5 언제 cosine을 쓰는가?

8.6 L2, cosine, inner product 관계

quadrantChart
  title Distance Metric Selection
  x-axis "크기 정보 약함" --> "크기 정보 중요"
  y-axis "의미/방향 약함" --> "의미/방향 중요"
  quadrant-1 "혼합 판단"
  quadrant-2 "Cosine"
  quadrant-3 "Preprocess First"
  quadrant-4 "L2"
  "Text Embedding": [0.25, 0.85]
  "Geo/Sensor Vector": [0.85, 0.35]
  "Normalized Embedding": [0.35, 0.75]
  "Raw Numeric Features": [0.75, 0.45]

8.7 간단 예제: Python 계산

import numpy as np
from sklearn.metrics.pairwise import cosine_distances, euclidean_distances

x = np.array([[3.0, 4.0]])
y = np.array([[6.0, 8.0]])
z = np.array([[4.0, 3.0]])

print("L2 x-y:", euclidean_distances(x, y)[0, 0])
print("Cosine x-y:", cosine_distances(x, y)[0, 0])
print("L2 x-z:", euclidean_distances(x, z)[0, 0])
print("Cosine x-z:", cosine_distances(x, z)[0, 0])

x와 y는 방향은 같지만 크기가 다르다. Cosine distance는 거의 0이지만, L2 distance는 크기 차이를 거리로 본다. 이 예제가 두 지표의 철학 차이를 가장 잘 보여준다.

8.8 실용 예제: pgvector metric 선택

-- L2 distance
SELECT id, content
FROM documents
ORDER BY embedding <-> '[0.1,0.2,0.3]'
LIMIT 5;

-- cosine distance
SELECT id, content
FROM documents
ORDER BY embedding <=> '[0.1,0.2,0.3]'
LIMIT 5;

-- negative inner product
SELECT id, content
FROM documents
ORDER BY embedding <#> '[0.1,0.2,0.3]'
LIMIT 5;

pgvector는 L2, inner product, cosine distance 연산자를 제공한다. 같은 데이터라도 어떤 연산자로 인덱스를 만들고 검색하느냐에 따라 결과와 성능이 달라질 수 있다.

8.9 MongoDB Atlas Vector Search 예시

{
  "fields": [
    {
      "type": "vector",
      "path": "embedding",
      "numDimensions": 1536,
      "similarity": "cosine"
    }
  ]
}

MongoDB Atlas Vector Search는 vector index에서 euclidean, cosine, dotProduct 같은 similarity 설정을 사용한다. 텍스트 RAG에서는 보통 embedding 모델 권장 metric 또는 cosine/dotProduct 계열을 먼저 확인한다.

8.10 실무 선택 가이드

8.11 리스크/반례

8.12 근거 URL

• scikit-learn cosine distances: https://scikit-learn.org/stable/modules/generated/sklearn.metrics.pairwise.cosine_distances.html
• scikit-learn euclidean distances: https://scikit-learn.org/stable/modules/generated/sklearn.metrics.pairwise.euclidean_distances.html
• pgvector distance operators: https://github.com/pgvector/pgvector
• MongoDB Atlas Vector Search index fields: https://www.mongodb.com/docs/atlas/atlas-vector-search/vector-search-type/
• Pinecone vector similarity metrics: https://docs.pinecone.io/guides/indexes/understanding-indexes#similarity-metrics
• Qdrant distance metrics: https://qdrant.tech/documentation/concepts/search/

9. RCPS 보통주 전환

9.1 소개

RCPS는 Redeemable Convertible Preferred Shares, 즉 상환전환우선주다. 보통주에는 없는 상환권과 전환권을 포함하는 우선주 형태로, 스타트업/벤처 투자에서 자주 사용된다.

상환권은 일정 조건에서 회사에 투자금 상환을 청구할 수 있는 권리이고, 전환권은 우선주를 보통주 등 다른 종류주식으로 바꿀 수 있는 권리다.

주의: 이 장은 일반 리서치 요약이며 법률/세무/회계 자문이 아니다. 실제 전환은 정관, 투자계약서, 주주간계약, 회계기준, 등기 실무를 전문가와 확인해야 한다.

9.2 전환 관련 법적 체크

flowchart LR
  A["RCPS 발행"] --> B["전환 조건 도래"]
  B --> C["투자자 전환청구"]
  C --> D["보통주 발행/전환 효력"]
  D --> E["변경등기"]
  D --> F["부채/자본 재분류 검토"]
  D --> G["지분율 희석 분석"]

9.3 재무/지배구조 영향

9.4 간단 계산 예시

전환 전 보통주가 1,000,000주이고, RCPS 100,000주가 1:1로 보통주 전환된다면:

기존 100,000주 보유자의 지분율은:

9.5 실무 체크리스트

9.6 근거 URL

• 상법 제345조~제350조 국가법령정보센터 PDF: https://www.law.go.kr/lbook/lbFileDownload.do?flExt=pdf&lbookConflSeq=37001&lbookSeq=44683
• 전환 절차 조문: https://law.go.kr/LSW/lsLawLinkInfo.do?chrClsCd=010202&lsJoLnkSeq=900424871
• NABO TIPS 평가 RCPS 설명: https://korea.nabo.go.kr/board/file/down.do?fid=33318010
• ZUZU RCPS 설명: https://zuzu.network/resource/guide/rcps/
• PwC M&A Guide Book 회계 영향: https://www.pwc.com/kr/ko/services/samilpwc_mna_guide-book.pdf

10. 저지연 파이프라인 최적화: STT → LLM → TTS

10.1 소개

음성 agent의 체감 품질은 전체 응답 완료 시간보다 Time To First Token 또는 Time To First Audio가 더 중요하다. 사용자는 첫 반응이 빠르면 전체 답변이 길어도 자연스럽게 느낀다.

sequenceDiagram
  participant Mic
  participant STT
  participant LLM
  participant TTS
  participant Speaker
  Mic->>STT: 20~100ms audio chunks
  STT-->>LLM: interim/final transcript
  LLM-->>TTS: token/text chunks
  TTS-->>Speaker: first audio chunk
  LLM-->>TTS: remaining stream
  TTS-->>Speaker: continuous playback

10.2 지연 요소

10.3 TTFT/TTFA 최적화 원칙

10.4 실용 예제: 비동기 스트리밍 파이프라인 스케치

import asyncio

async def stt_stream(audio_chunks):
    async for chunk in audio_chunks:
        # 실제 구현에서는 STT WebSocket으로 전송하고 interim/final 이벤트를 받는다.
        yield {"type": "interim", "text": "안녕하세요"}

async def llm_stream(text: str):
    for token in ["네, ", "도와드릴게요."]:
        await asyncio.sleep(0.03)
        yield token

async def tts_stream(text_chunks):
    async for text in text_chunks:
        # 실제 구현에서는 TTS WebSocket으로 text chunk를 보내고 audio bytes를 받는다.
        yield text.encode()

async def pipeline(audio_chunks):
    async for stt_event in stt_stream(audio_chunks):
        if stt_event["type"] in {"interim", "final"}:
            async def text_chunks():
                async for token in llm_stream(stt_event["text"]):
                    yield token

            async for audio in tts_stream(text_chunks()):
                yield audio

10.5 Realtime API vs Cascaded Pipeline

10.6 근거 URL

• OpenAI Realtime API: https://developers.openai.com/api/docs/guides/realtime
• OpenAI realtime conversation: https://developers.openai.com/api/docs/guides/realtime-conversations
• OpenAI latency optimization: https://developers.openai.com/api/docs/guides/latency-optimization
• OpenAI prompt caching: https://developers.openai.com/api/docs/guides/prompt-caching
• Google Speech-to-Text best practices: https://docs.cloud.google.com/speech-to-text/docs/v1/best-practices
• Deepgram endpointing/interim results: https://developers.deepgram.com/docs/understand-endpointing-interim-results
• ElevenLabs latency optimization: https://elevenlabs.io/docs/eleven-api/guides/how-to/best-practices/latency-optimization
• ElevenLabs audio streaming: https://elevenlabs.io/docs/eleven-api/concepts/audio-streaming

11. GStreamer

11.1 소개

GStreamer는 오디오/비디오/임의 데이터 흐름을 처리하기 위한 멀티미디어 프레임워크다. 핵심은 source, filter, encoder, muxer, sink 같은 element를 pipeline으로 연결해 데이터 흐름을 구성하는 것이다.

GStreamer는 media player뿐 아니라 실시간 transcoding, camera capture, RTP/WebRTC pipeline, AI inference 전처리, STT/TTS audio bridge에도 사용할 수 있다.

11.2 특징

flowchart LR
  S["filesrc / rtspsrc"] --> D["decodebin"]
  D --> C["audioconvert"]
  C --> R["audioresample"]
  R --> E["opusenc"]
  E --> M["rtpopuspay"]
  M --> K["udpsink / appsink"]

11.3 장단점

11.4 간단 예제 Python: 테스트 영상 출력

#!/usr/bin/env python3
import sys
import gi

gi.require_version("Gst", "1.0")
from gi.repository import Gst

Gst.init(sys.argv[1:])

pipeline = Gst.parse_launch("videotestsrc pattern=smpte ! videoconvert ! autovideosink")
pipeline.set_state(Gst.State.PLAYING)

bus = pipeline.get_bus()
bus.timed_pop_filtered(Gst.CLOCK_TIME_NONE, Gst.MessageType.ERROR | Gst.MessageType.EOS)

pipeline.set_state(Gst.State.NULL)

11.5 실용 예제 Python: 마이크 PCM을 appsink로 받아 STT에 전달

#!/usr/bin/env python3
import sys
import gi

gi.require_version("Gst", "1.0")
gi.require_version("GLib", "2.0")
from gi.repository import Gst, GLib

Gst.init(sys.argv[1:])

pipeline = Gst.parse_launch(
    "autoaudiosrc ! audioconvert ! audioresample "
    "! audio/x-raw,format=S16LE,rate=16000,channels=1 "
    "! appsink name=sink emit-signals=true sync=false max-buffers=10 drop=true"
)

appsink = pipeline.get_by_name("sink")

def on_sample(sink):
    sample = sink.emit("pull-sample")
    buffer = sample.get_buffer()
    ok, map_info = buffer.map(Gst.MapFlags.READ)
    if ok:
        pcm_bytes = bytes(map_info.data)
        # 여기에서 STT WebSocket으로 pcm_bytes를 전송한다.
        buffer.unmap(map_info)
    return Gst.FlowReturn.OK

appsink.connect("new-sample", on_sample)
pipeline.set_state(Gst.State.PLAYING)

loop = GLib.MainLoop()
try:
    loop.run()
finally:
    pipeline.set_state(Gst.State.NULL)

이 예제는 저지연 STT 파이프라인에서 유용하다. max-buffers=10 drop=true는 소비가 밀릴 때 latency가 계속 늘어나는 것을 막는 backpressure 전략이다.

11.6 근거 URL

• GStreamer 소개: https://gstreamer.freedesktop.org/documentation/application-development/introduction/gstreamer.html
• GStreamer basic tutorials: https://gstreamer.freedesktop.org/documentation/tutorials/basic/index.html
• GStreamer concepts tutorial: https://gstreamer.freedesktop.org/documentation/tutorials/basic/concepts.html
• GStreamer hello world Python: https://gstreamer.freedesktop.org/documentation/tutorials/basic/hello-world.html
• GStreamer dynamic pipeline/demuxer: https://gstreamer.freedesktop.org/documentation/tutorials/basic/dynamic-pipelines.html

12. Docker

12.1 소개

Docker는 애플리케이션과 실행 환경을 container image로 묶어 개발, 테스트, 배포를 일관되게 만드는 플랫폼이다. Docker container는 image의 실행 인스턴스이며, host OS kernel을 공유하면서 namespace/cgroup 등으로 격리된다.

12.2 특징

flowchart LR
  C["Source Code"] --> D["Dockerfile"]
  D --> I["Docker Image"]
  I --> R["Registry"]
  R --> T["Container Runtime"]
  T --> A["Running App"]

12.3 장단점

12.4 간단 예제

app.py

from flask import Flask

app = Flask(__name__)

@app.get("/")
def home():
    return {"message": "hello docker"}

Dockerfile

FROM python:3.12-slim

WORKDIR /app
RUN pip install --no-cache-dir flask
COPY app.py .

EXPOSE 8000
CMD ["flask", "--app", "app", "run", "--host", "0.0.0.0", "--port", "8000"]

docker build -t hello-docker:latest .
docker run --rm -p 8000:8000 hello-docker:latest

12.5 실용 예제: API + Redis Compose

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      REDIS_URL: redis://redis:6379/0
    depends_on:
      - redis

  redis:
    image: redis:7-alpine
    volumes:
      - redis-data:/data

volumes:
  redis-data:

docker compose up --build
docker compose logs -f api
docker compose down

12.6 Docker in AWS 사용 사례

12.7 AWS 주의사항

12.8 근거 URL

• Docker overview: https://docs.docker.com/get-started/docker-overview/
• Dockerfile overview: https://docs.docker.com/build/concepts/dockerfile/
• Docker Compose model: https://docs.docker.com/compose/intro/compose-application-model/
• Docker build best practices: https://docs.docker.com/build/building/best-practices/
• Amazon ECR: https://docs.aws.amazon.com/AmazonECR/latest/userguide/what-is-ecr.html
• Amazon ECS: https://docs.aws.amazon.com/AmazonECS/latest/developerguide/Welcome.html
• Amazon ECS image best practices: https://docs.aws.amazon.com/AmazonECS/latest/developerguide/container-considerations.html
• Amazon ECS security: https://aws.amazon.com/blogs/security/security-considerations-for-running-containers-on-amazon-ecs/
• Amazon EKS: https://docs.aws.amazon.com/eks/latest/userguide/what-is-eks.html

13. 멀티플렉싱

13.1 소개

멀티플렉싱은 여러 독립적인 데이터 흐름을 하나의 물리적/논리적 연결 위에 함께 실어 보내는 기술이다. 네트워크에서는 하나의 TCP/QUIC 연결 위에 여러 stream을 태우는 방식으로 나타나고, 미디어에서는 여러 audio/video/subtitle stream을 하나의 container에 muxing하는 방식으로 나타난다.

13.2 네트워크 멀티플렉싱

flowchart TD
  C["One Connection"] --> S1["Stream 1 frames"]
  C --> S2["Stream 2 frames"]
  C --> S3["Stream 3 frames"]
  S1 --> R["Receiver reassembles by stream id"]
  S2 --> R
  S3 --> R

13.3 미디어 멀티플렉싱

미디어에서 multiplexing은 audio, video, metadata, subtitle을 하나의 container stream으로 합치는 작업이다. 반대로 demultiplexing은 container에서 각 stream을 분리한다.

13.4 장단점

13.5 실용 예제: GStreamer mux/demux

# 비디오/오디오를 MPEG-TS로 muxing
gst-launch-1.0 \
  videotestsrc ! x264enc tune=zerolatency ! h264parse ! queue ! mux. \
  audiotestsrc ! audioconvert ! avenc_aac ! aacparse ! queue ! mux. \
  mpegtsmux name=mux ! filesink location=out.ts

# MPEG-TS demuxing
gst-launch-1.0 filesrc location=out.ts ! tsdemux name=demux \
  demux. ! queue ! decodebin ! autovideosink \
  demux. ! queue ! decodebin ! autoaudiosink

13.6 근거 URL

• HTTP/2 RFC 9113: https://www.ietf.org/rfc/rfc9113.html
• QUIC RFC 9000: https://www.ietf.org/rfc/rfc9000.html
• QUIC Working Group: https://quicwg.org/
• GStreamer dynamic pipelines/demuxer: https://gstreamer.freedesktop.org/documentation/tutorials/basic/dynamic-pipelines.html

14. 실시간 방송송출, 화상통화: VBR vs CBR

14.1 핵심 결론

실시간 방송송출과 화상통화는 둘 다 “실시간 영상”이지만 권장 rate control이 다르다.

짧게 말하면:

방송 플랫폼에 “밀어 넣는 송출”은 CBR이 기본값이고, 양방향 화상통화는 CBR보다 adaptive bitrate가 기본 전략이다.

14.2 왜 방송송출은 CBR이 자주 권장되는가?

YouTube Live 공식 encoder 설정은 bitrate encoding을 CBR로 안내한다. Twitch도 방송 가이드에서 CBR 사용을 권장한다. 이유는 단순히 “화질이 좋아서”가 아니라, 전체 서비스 품질이 더 예측 가능하기 때문이다.

flowchart LR
  E["Encoder CBR"] --> I["Platform Ingest"]
  I --> T["Transcoding"]
  T --> C["CDN"]
  C --> V["Viewers"]
  E -. "bitrate spike 감소" .-> I

14.3 방송송출에서 VBR/QVBR이 유리한 예외

AWS Elemental Live 문서는 CBR이 지정 bitrate를 유지하지만 장면 복잡도에 따라 화질이 변한다고 설명한다. 반대로 VBR은 평균 bitrate와 최대 bitrate를 지정하고, QVBR은 품질 수준과 최대 bitrate를 지정한다. AWS 문서 기준으로는 장비/시청자가 variable bitrate를 처리할 수 있고 대역폭 비용이 중요하면 VBR/QVBR을 고려할 수 있다.

실무적으로는 “VBR이 화질 효율은 좋지만, live ingest에는 burst를 막는 max bitrate와 buffer 설정이 필수”라고 보는 편이 안전하다.

14.4 왜 화상통화는 Adaptive VBR/ABR이 권장되는가?

화상통화는 방송송출과 다르게 지연 시간이 매우 짧아야 하고, 사용자 네트워크가 Wi-Fi, LTE/5G, 사내망, 해외망 등으로 계속 변한다. 이때 고정 CBR을 유지하면 네트워크가 나빠진 순간 큐가 쌓이고 latency가 커지거나 packet loss가 늘어난다.

WebRTC는 RTCRtpSender.setParameters()에서 maxBitrate, maxFramerate, scaleResolutionDownBy 같은 전송 제어를 제공한다. W3C/MDN 설명처럼 maxBitrate는 “상한”이지 항상 그 bitrate로 보내라는 뜻이 아니다. 실제 전송량은 transport/network 한계, codec, frame drop, resolution scaling에 따라 바뀐다.

flowchart TD
  N["Network changes"] --> BWE["Bandwidth Estimation"]
  BWE --> P["Sender Parameters"]
  P --> BR["Bitrate"]
  P --> FPS["Frame Rate"]
  P --> RES["Resolution"]
  BR --> L["Low Latency Playout"]
  FPS --> L
  RES --> L

14.5 실무 권장안

14.6 간단 예제: WebRTC 송신 bitrate 상한 설정

async function setSenderLimit(sender, maxBitrateBps, maxFps) {
  const params = sender.getParameters();
  params.encodings ??= [{}];
  params.encodings[0].maxBitrate = maxBitrateBps;
  params.encodings[0].maxFramerate = maxFps;
  await sender.setParameters(params);
}

이 코드는 CBR을 강제하는 코드가 아니다. WebRTC encoder가 네트워크 상태에 따라 동적으로 움직이되, 지정한 상한을 넘지 않도록 제한하는 용도다.

14.7 근거 URL

• YouTube Live encoder settings: https://support.google.com/youtube/answer/2853702?hl=en
• Twitch broadcasting guidelines: https://help.twitch.tv/s/article/broadcasting-guidelines?language=en_US
• AWS Elemental Live QVBR/VBR/CBR: https://docs.aws.amazon.com/elemental-live/latest/ug/qvbr-and-rate-control-mode.html
• MDN RTCRtpSender.setParameters(): https://developer.mozilla.org/en-US/docs/Web/API/RTCRtpSender/setParameters
• W3C WebRTC RTCRtpEncodingParameters: https://w3c.github.io/webrtc-pc/
• WebRTC transport-wide congestion control: https://webrtc.googlesource.com/src/+/main/docs/native-code/rtp-hdrext/transport-wide-cc-02/README.md

15. Fuzzy Search

15.1 소개

Fuzzy Search는 사용자의 검색어가 정확히 일치하지 않아도 “비슷한 문자열”을 찾아주는 검색 방식이다. 예를 들어 iphnoe로 검색해도 iphone을 찾거나, hundai로 검색해도 hyundai를 찾게 만드는 기술이다.

실무에서는 fuzzy search라는 말을 넓게 쓰지만, 실제 구현은 여러 계열로 나뉜다.

flowchart LR
  Q["User Query<br/>iphnoe"] --> N["Normalize<br/>lowercase, trim"]
  N --> F["Fuzzy Matching<br/>edit distance / trigram"]
  F --> R["Candidate Results"]
  R --> S["Ranking<br/>exact > prefix > fuzzy"]
  S --> O["iphone, iphone case, ..."]

15.2 핵심 원리

가장 대표적인 기준은 edit distance다. 두 문자열을 같게 만들기 위해 필요한 최소 편집 횟수를 계산한다.

예를 들어 kitten과 sitting의 Levenshtein distance는 3이다.

Elasticsearch/OpenSearch의 fuzziness: AUTO는 단어 길이에 따라 허용 edit distance를 정한다. 보통 짧은 단어는 정확히 맞추고, 긴 단어는 1~2회 오타를 허용한다.

15.3 장단점

15.4 주요 구현 선택지

15.5 Fuzzy Search vs Vector Search

실무 검색 UX는 보통 하나만 쓰지 않는다. exact match > prefix/autocomplete > fuzzy > semantic 순으로 후보를 만들고, 업무 중요도에 맞춰 점수를 섞는다.

15.6 간단 예제: Python RapidFuzz

from rapidfuzz import process, fuzz

choices = ["Samsung Electronics", "Hyundai Motor", "NVIDIA", "Microsoft"]

result = process.extract(
    "samsng electornics",
    choices,
    scorer=fuzz.WRatio,
    limit=3,
)

print(result)

RapidFuzz는 검색엔진이라기보다 문자열 유사도 계산/후보 재정렬에 가깝다. 대용량 전체 검색에는 DB index나 검색엔진으로 후보를 줄인 뒤 RapidFuzz로 reranking하는 구조가 더 안전하다.

15.7 실용 예제: PostgreSQL pg_trgm

CREATE EXTENSION IF NOT EXISTS pg_trgm;

CREATE TABLE companies (
  id bigserial PRIMARY KEY,
  name text NOT NULL
);

CREATE INDEX companies_name_trgm_idx
ON companies
USING gin (name gin_trgm_ops);

SELECT id, name, similarity(name, 'samsng') AS score
FROM companies
WHERE name % 'samsng'
ORDER BY score DESC
LIMIT 10;

pg_trgm은 문자열을 trigram으로 쪼개 유사도를 계산한다. GIN/GiST index를 함께 쓰면 Postgres 안에서 실용적인 fuzzy search를 만들 수 있다.

15.8 실용 예제: Elasticsearch fuzzy query

{
  "query": {
    "match": {
      "company_name": {
        "query": "samsng",
        "fuzziness": "AUTO",
        "prefix_length": 1,
        "max_expansions": 50
      }
    }
  }
}

prefix_length는 앞부분 몇 글자를 반드시 정확히 맞출지 정한다. 이 값을 키우면 엉뚱한 후보와 비용을 줄일 수 있지만, 앞글자 오타에는 약해진다.

15.9 실용 예제: MongoDB Atlas Search autocomplete fuzzy

db.products.aggregate([
  {
    $search: {
      autocomplete: {
        path: "name",
        query: "iphno",
        fuzzy: {
          maxEdits: 1,
          prefixLength: 1,
          maxExpansions: 256
        }
      }
    }
  },
  { $limit: 10 },
  { $project: { name: 1, score: { $meta: "searchScore" } } }
])

MongoDB Atlas Search의 fuzzy.maxEdits는 1 또는 2를 사용한다. 자동완성과 함께 쓸 때는 prefixLength와 maxExpansions가 품질과 비용을 좌우한다.

15.10 실무 설계 가이드

15.11 리스크/반례

15.12 근거 URL

• Elasticsearch fuzzy query: https://www.elastic.co/docs/reference/query-languages/query-dsl/query-dsl-fuzzy-query
• Elasticsearch fuzziness common option: https://www.elastic.co/guide/en/elasticsearch/reference/8.19/common-options.html
• OpenSearch fuzzy query: https://docs.opensearch.org/latest/query-dsl/term/fuzzy/
• MongoDB Atlas Search autocomplete fuzzy: https://www.mongodb.com/docs/atlas/atlas-search/autocomplete/
• PostgreSQL pg_trgm: https://www.postgresql.org/docs/current/static/pgtrgm.html
• RapidFuzz documentation: https://rapidfuzz.github.io/RapidFuzz/
• Fuse.js fuzzy search: https://www.fusejs.io/fuzzy-search.html
• Meilisearch typo tolerance: https://www.meilisearch.com/docs/resources/internals/typo_tolerance

16. 종합 설계 제안

16.1 AI SaaS를 만든다면 추천 아키텍처

flowchart TD
  U["Browser / Mobile"] --> API["FastAPI async API"]
  API --> TC["Tenant Middleware"]
  TC --> AG["PydanticAI / LangChain Agent"]
  AG --> WF["LangGraph for workflows"]
  AG --> VDB["MongoDB Vector / pgvector"]
  API --> VOICE["Realtime Voice Pipeline"]
  VOICE --> GST["GStreamer audio pipeline"]
  API --> Q["Async Queue"]
  API --> LOG["Tenant-aware logs/metrics"]
  subgraph AWS
    ECR["ECR"]
    ECS["ECS Fargate"]
    SM["Secrets Manager"]
  end
  ECR --> ECS
  ECS --> API
  SM --> API

16.2 우선순위

17. 호환성 체크리스트

18. 사실검증 메모

19. 작성 시 사용한 사용자 질문 프롬프트

아래 내용들을 자세히 조사후 md 파일로 정리해주세요

멀티테넌트

파이썬 비동기 프로그래밍
- 소개
- 장단점
- 특징
- 간단예제
- 실용예제

LangChain, LangGraph
- 소개, 특징, 장단점
- 간단예제, 실용예제

PydanticAI
- 소개, 특징, 장단점
- 간단예제, 실용예제

LangChain, LangGraph vs PydanticAI
- 비교분석

vectorDB in MongoDB
- 소개, 특징, 장단점
- 간단예제, 실용예제

vectorDB in other SQL DB
- 소개, 특징, 장단점
- 간단예제, 실용예제

rcps 보통주 전환

저지연 파이프라인 최적화
stt llm tts 전체파이프라인의 ttft 단축 및 스트리밍 응답 최적화

리서치 항목 추가
gstreamer
- 소개
- 장단점
- 특징
- 간단예제 python
- 실용예제 python

리서치 항목추가
docker
- 소개
- 장단점
- 특징
- 간단예제
- 실용예제
- in AWS 에서 사용 사례, 주의사항

리서치 항목 추가
멀티플렉싱

리서치 주제 추가
실시간 방송송출, 화상통화에서는 VBR or CBR 중 어느것이 더 권장 되는가?
그 이유는?

리서치 주제 추가
fuzzy search

주제 추가
L2 distance vs cosine distance