The Yellow Lion King 데이터와 함께 살아가기

최근 이슈가되는 DeepSeek에 대한 그 현상과 원인 설명을 통해 내용 이해에 도움을 드리고자 합니다.

동영상 설명은 아래링크 참고
https://youtu.be/uJuIuCvENQo?si=2JiU1WPhVu_4trpK

목차

• 현상
  • NVIDIA 주가 폭락
  • 나스닥 지수 폭락
• 원인분석
  1. DeepSeek App이 Download 1위해서
  2. (1위 한 이유는) 좋은 서비스를 무료로 제공해서
  3. (좋은 서비스를 무료로 제공할 수 있었던 이유는) 좋은 모델을 싸게 만들어서
  4. (좋은 모델을 싸게 만들 수 있었던 이유는) 엔지니어링을 통해 저사양의 서버를 효과적으로 사용해서 (엔지니어링 기법 등 별도 포스팅)
  5. (저사양의 서버를 효과적으로 사용할 수 있었던 이유는) 좋은 엔지니어들이 있어서
• 영향도 분석

현상

NVIDIA 주가 폭락

2025년 1월 27일 월요일 하루만에 NVIDIA의 주가가 약 17%가까이 급락했습니다. 거의 6000억 달러의 시장가격이 사라졌다고 합니다.

1월 27일 NVIDIA 주가 16.97% 급락

나스닥 지수 폭락

뿐만아니라 특기 기술 주식들이 상장되어있는 나스닥도 급락했습니다.

원인분석

1. DeepSeek App Download 1위

위에서 처럼 왜 이렇게 급락한 것 일까요?
여러가지가 있겠지만 그 주요 이유는 미국 App Store 에서 무료 App중에서 다운로드 1등을 찍은 앱이 DeepSeek라는 앱이었기 때문입니다. DeepSeek라는 앱은 ChatGPT 앱 또는 Perplexity 앱과 같은 AI Assistant App 입니다.

DeepSeek라는 회사나 이름은 잘 모르겠지만 다운로드 1등한 것이 뭐가 큰 문제여서 엔비디아 주가가 폭락한 걸까요?

2. (DeepSeek App이 1위 한 이유는) 좋은 서비스를 무료로 제공해서

먼저 왜 DeepSeek가 미국 앱스토어에서 다운로드 1등을 했는가를 알아볼 필요가 있습니다.  
DeepSeek가 성능 좋은 서비스를 무료로 제공했기 때문에 많은 앱 다운로드가 있었고 그래서 다운로드 1등을 했습니다.
DeepSeek 앱은 DeepSeek-V3를 기본으로 하는데 이 모델의 성능이 아래와 같이 다른 모델과 비교해서 동등하거나 보다 우수한 부분도 있거든요.

특히 놀라운 점은 1월 20일에 발표한 DeepSeek-R1의 경우 수학문제풀이와 코딩 밴치마크 테스트에서 OpenAI를 능가하는 성능을 보인 것 입니다.

이처럼 좋은 모델을 (ChatGPT 에서는 별도 비용을 지불해야하지만) DeepSeek 앱에서는 무료로 사용할 수 있기 때문에 DeepSeek App이 미국 App Store에서 다운로드 1위를 할 수 있었습니다.
 
조금 다른 이야기이지만 API 비용도 저렴합니다.
DeepSeek는 훈련도 싸게하고 추론(서비스)도  저렴하게 할 수 있기 때문에 서비스도 무료나 저렴하게 제공합니다. 예를들어 OpenAI ChatGPT에서는 유료 가입회원에게만 제공하는 o1 모델과 유사한 성능의 서비스를 무료로 제공하고 있습니다.

3. (좋은 서비스를 무료로 제공할 수 있었던 이유는) 좋은 모델을 싸게 만들어서

DeepSeek는 OpenAI의 Chat-GPT나 Meta의 Llama의 성능을 낼 수 있는 또 다른 모델(DeepSeek-R1)을 매우 적은 비용으로 만들었습니다. 무려 10분의 1 도 안되는 비용으로 말이지요. DeepSeek에서는 5.6M 달러가 들었다고 발표했는데 다른 회사에서는 수억 달러 또는 수십억 달러 정도 들었다고 했거든요.

4. (좋은 모델을 싸게 만들 수 있었던 이유는) 엔지니어링을 통해 저사양의 서버를 효과적으로 사용해서

이렇게 저비용으로 모델을 만들고 서비스할 수 있었던 이유에는 (데이터, 환경, 인프라, 리소스 등) 여려가지가 있겠지만 저는 한마디로 말하면

DeepSeek가 엔지니어링을 잘했다

라고 말할 수 있겠습니다. 모델을 개발하고 서비스할 때 하드웨어를 효율적으로 운영할 수 있도록 여러 테크닉을 적용해서 개발했기 때문입니다. DeepSeep에서 발표한 모델은 많은 제약과 한계를 극복하기 위해 다양한 아이디어의 기술을 적용했습니다.
어떻게 엔지니어링을 했는지는 별도 포스팅
엔지니어링을 잘해서 비용 효율적인 모델과 서비스를 만들었습니다.
이처럼 좋은 모델을 만든 것도 대단한데 더욱 놀라운 것은 이러한 모델을 NVIDIA H800이라는 서버를 이용해서 만들었다는 것 입니다.
DeepSeek는

NVIDIA H800서버 2048대를 클러스터로 만들어서 훈련

했습니다. 그 이유는 미국의 규제로 인하여 중국에 납품하는 NVIDIA 서버는 제한되어있기 때문입니다. 그래서 더 좋은 서버를 사용할 수 없었습니다. H800서버는 H100서버의 다운 스펙서버입니다. 그럼에도 불구하고 더 짧은 시간의 훈련시간과 좋은 성능을 달성했으니 놀라지 않을 수 없습니다.

위 클리앙의 내용을 보니 구체적으로, FP64일때의 성능과 NVLink, Interconnection bandwidth에서 차이가 나네요.

5. (저사양의 서버를 효과적으로 사용할 수 있었던 이유는) 좋은 엔지니어들이 있어서

그럴 수 있었던 것은 DeepSeek회사 직원들의 배경이 주식자동거래 프로그램을 만드는 퀀트들이었다고 하네요. 그래서 최적화를 위한 다양한 레벨의 인지니어링이 가능했다고 생각합니다.
 
 

영향도 분석(별도 포스팅)

미국 정부
먼저 이번 DeepSeek 이슈로 인하여 AI에 대해서 미국 리더십을 확실하게 가져 가려고 했던 트럼프 정부를 긴장시킨 것은 맞는 것 같습니다. 트럼프가 미국 기업에게 알람이 되어야 한다고 이야기 했으니까요. 기존에는 AI 개발을 위해서 인프라가 중요하고, 인프라에서 우위를 점하고 있는 미국이 당연히 계속해서 앞서 나갈 것으로 생각했는데, 그리고 그래서 더욱 앞서 나가기 위해서 향후 투자 계획도 발표했는데 안좋은 인프라에서도 AI개발이 가능하다고하는 사례가 나왔기 때문이지요. 그렇기 때문에 중국에 대한 견제는 더욱 심해질 것으로 예상됩니다. 새로운 규제가 더 나올 수도 있겠지요. 그리고 AI 경쟁에서의 스푸트니크 쇼크를(소련이 먼저 인공위성을 쏘아올려서 미국을 깜짝 놀라게 만들었던 사건을) 방지하기위해 AI 관련 투자에 더욱 노력할 수 있습니다. 
AI 관련 미국 기업
위에서 말한 것처럼 저사양의 서버로도 훈련하고 서비스할 수 있다면 굳이 비싼 서버를 사용할 필요가 없지요. 고사양의 비싼 서버를 구매할 필요가 당분간은 줄어들 것으로 예상됩니다. 그래서 엔비디아 같은 기업의 경우 매출 감소가 예상이되니 주가가 하락한 것이죠. 그러나 반대로 적은 비용으로도 AI 모델 개발이 가능해지니 더 많은 기업들이 뛰어들어서 전체적인 서버 수요는 늘어날 것이라는 예상도 있습니다.
Software 중심의 Big Tech기업의 경우 당연히 엔지니어링에 더욱 집중 할 것으로 예상됩니다. 좋은 사례를 보았으니 유사하거나 더 좋은 방법을 찾고자 노력할 것으로 예상됩니다. 
AI 관련 한국 기업
같은 맥락에서 한국의 Software 기업에게도 기회가 생겼다고 할수 있습니다. 적은 비용으로 고성능의 품질 좋은 AI서비스를 만들 수 있게되었으니 
SK하이닉스나 삼정전자 같은 Hardware 기업은 단기적으로 엔비디아와 같이 실적이 줄어들 수 있을 것 같습니다. 서버 수요가 늘어난다고 해도 저사양, 낮은 단가의 서버이기 때문에 높은 부가가치를 바라기는 어렵게 되는 거죠.
 

Source

NVIDIA 주가 조회
나스닥 지수 조회
DeepSeek 1등 뉴스

DeepSeek API Docs(버전 릴리즈 공지 내용)

관련 CNN 뉴스
 
나무위키 DeepSeek

클리앙 H100, H800 비교

지난 시간에는 LCEL Get-started를 통해서 LCEL의 간단하면서 다양한 예시를 알아보았습니다.

이번 시간에는 LCEL을 사용하는 이유에 대해서 알아보도록 하겠습니다. 즉, 사용하면 무엇이 좋은지 어떤 가치가 있는지 알아보겠습니다. 참고로 LCEL Get-started를 먼저 보고 오시면 이해에 도움이 됩니다. 혹시 안보셨다면 먼저 읽어보시기를 추천 드립니다.

LCEL은 간단한 콤포넌트 부터 복잡한 체인을 쉽게 만들 수 있게 합니다. 바로 다음과 같은 것들을 이용해서 만듭니다.

1. 통일된 인터페이스:

모든 LCEL 객체는 Runnable 인터페이스를 구현합니다. Runnable은 호출하는 메소드들의 공통 집합을 정의합니다.(invoke, batch, stream, ainvoke,...등과 같은...)  이것은 LCEL객체가 자동적으로 이러한 호출들을 지원할 수 있게 만듭니다. 즉, LCEL 객체로 만든 모든 체인은 하나의 LCEL객체라는 것을 말합니다. 

2. 원시도구들의 구성:

LCEL의 가치를 이해하기 위해서, LCEL 없이 비슷한 기능을 어떻게 개발하는 지 생각해보고 동작하는 지를 살펴보는 것은 도움이 됩니다. LCEL은 콤포넌트들을 병렬화하고, fallbacks을 추가하고, 체인 내부의 구성을 동적으로 바꾸는 등을 통해서 체인을 쉽게 구성하게 해줍니다. 이번 포스팅에서는 지난 번 LCEL Get-started의 내용에 있던 기본 예시를 다루어 보겠습니다. 간단한 프롬프트와 모델이 결합된 체인을 다루어 볼껀데요, 이것은 이미 많은 기능들이 정의되어 있습니다. 이것을 재생성하는 데 필요한 것이 무엇인지 살펴보겠습니다.

준비

먼저 공통적으로 필요한 체인 하나를 만들어 보겠습니다.

%pip install –upgrade –quiet langchain-core langchain-openai

관련된 패키지를 설치 합니다.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser


prompt = ChatPromptTemplate.from_template("Tell me a short joke about {topic}")
model = ChatOpenAI(model="gpt-3.5-turbo")
output_parser = StrOutputParser()

chain = prompt | model | output_parser

프롬프트, 모델, 아웃풋 파서를 연결한 체인을 만듭니다.

이제부터 다음 내용을 구현할 때 두가지 경우의 코드를 비교하겠습니다. 즉, LCEL 없이 구현할 때와 LCEL로 구현할 때의 코드를 보겠습니다.

• invoke, stream, batch, async, changing model provider, runtime configuration, logging, Fallback

Invoke

제일 간단한 사용예를 보면, 토픽(주제) 문자열을 전달해 주 농담 문자열을 받는 것 입니다. 아래는 이러한 내용을 LCEL 없이 개발했을 때와 LCEL로 개발했을 때의 차이를 보여 줍니다. 

LCEL 없이 만드는 경우

from typing import List

import openai


prompt_template = "Tell me a short joke about {topic}"
client = openai.OpenAI()

def call_chat_model(messages: List[dict]) -> str:
    response = client.chat.completions.create(
        model="gpt-3.5-turbo", 
        messages=messages,
    )
    return response.choices[0].message.content

def invoke_chain(topic: str) -> str:
    prompt_value = prompt_template.format(topic=topic)
    messages = [{"role": "user", "content": prompt_value}]
    return call_chat_model(messages)

invoke_chain("ice cream")

LCEL로 만드는 경우

from langchain_core.runnables import RunnablePassthrough


prompt = ChatPromptTemplate.from_template(
    "Tell me a short joke about {topic}"
)
output_parser = StrOutputParser()
model = ChatOpenAI(model="gpt-3.5-turbo")
chain = (
    {"topic": RunnablePassthrough()} 
    | prompt
    | model
    | output_parser
)

chain.invoke("ice cream")

Stream

만약에 결과로 스트림을 원한다면 위에서 LCEL 없이 만든 함수의 내용을 수정해야 합니다.

LCEL 없이 만드는 경우

from typing import Iterator


def stream_chat_model(messages: List[dict]) -> Iterator[str]:
    stream = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=messages,
        stream=True,
    )
    for response in stream:
        content = response.choices[0].delta.content
        if content is not None:
            yield content

def stream_chain(topic: str) -> Iterator[str]:
    prompt_value = prompt.format(topic=topic)
    return stream_chat_model([{"role": "user", "content": prompt_value}])


for chunk in stream_chain("ice cream"):
    print(chunk, end="", flush=True)

LCEL로 만드는 경우

for chunk in chain.stream("ice cream"):
    print(chunk, end="", flush=True)

LCEL로 만드는 경우는 그저 두 줄의 코드면 충분합니다.

Batch

병렬적으로 여러 입력을 배치로 실행하기 원한다면 또 다시 새로운 함수가 필요합니다.

LCEL 없이 만드는 경우

from concurrent.futures import ThreadPoolExecutor


def batch_chain(topics: list) -> list:
    with ThreadPoolExecutor(max_workers=5) as executor:
        return list(executor.map(invoke_chain, topics))

batch_chain(["ice cream", "spaghetti", "dumplings"])

LCEL로 만드는 경우

chain.batch(["ice cream", "spaghetti", "dumplings"])

Async

비동기 버전을 원한다면,

LCEL 없이 만드는 경우

async_client = openai.AsyncOpenAI()

async def acall_chat_model(messages: List[dict]) -> str:
    response = await async_client.chat.completions.create(
        model="gpt-3.5-turbo", 
        messages=messages,
    )
    return response.choices[0].message.content

async def ainvoke_chain(topic: str) -> str:
    prompt_value = prompt_template.format(topic=topic)
    messages = [{"role": "user", "content": prompt_value}]
    return await acall_chat_model(messages)

LCEL로 만드는 경우

chain.ainvoke("ice cream")

Chat Model 대신 LLM 사용

chat 엔드포인트 대신에 completion 엔드포인트를 사용하기 원한다면,

LCEL없이 만드는 경우

def call_llm(prompt_value: str) -> str:
    response = client.completions.create(
        model="gpt-3.5-turbo-instruct",
        prompt=prompt_value,
    )
    return response.choices[0].text

def invoke_llm_chain(topic: str) -> str:
    prompt_value = prompt_template.format(topic=topic)
    return call_llm(prompt_value)

invoke_llm_chain("ice cream")

LCEL로 만드는 경우

from langchain_openai import OpenAI

llm = OpenAI(model="gpt-3.5-turbo-instruct")
llm_chain = (
    {"topic": RunnablePassthrough()} 
    | prompt
    | llm
    | output_parser
)

llm_chain.invoke("ice cream")

모델 제공자가 다른 경우

OpenAI 대신에 Anthropic을 사용하기 원하면,

LCEL 없이 만드는 경우

import anthropic

anthropic_template = f"Human:\n\n{prompt_template}\n\nAssistant:"
anthropic_client = anthropic.Anthropic()

def call_anthropic(prompt_value: str) -> str:
    response = anthropic_client.completions.create(
        model="claude-2",
        prompt=prompt_value,
        max_tokens_to_sample=256,
    )
    return response.completion    

def invoke_anthropic_chain(topic: str) -> str:
    prompt_value = anthropic_template.format(topic=topic)
    return call_anthropic(prompt_value)

invoke_anthropic_chain("ice cream")

LCEL로 만드는 경우

from langchain_community.chat_models import ChatAnthropic

anthropic = ChatAnthropic(model="claude-2")
anthropic_chain = (
    {"topic": RunnablePassthrough()} 
    | prompt 
    | anthropic
    | output_parser
)

anthropic_chain.invoke("ice cream")

실행 환경 설정(Runtime configurability)

LCEL 없이 만드는 경우

def invoke_configurable_chain(
    topic: str, 
    *, 
    model: str = "chat_openai"
) -> str:
    if model == "chat_openai":
        return invoke_chain(topic)
    elif model == "openai":
        return invoke_llm_chain(topic)
    elif model == "anthropic":
        return invoke_anthropic_chain(topic)
    else:
        raise ValueError(
            f"Received invalid model '{model}'."
            " Expected one of chat_openai, openai, anthropic"
        )

def stream_configurable_chain(
    topic: str, 
    *, 
    model: str = "chat_openai"
) -> Iterator[str]:
    if model == "chat_openai":
        return stream_chain(topic)
    elif model == "openai":
        # Note we haven't implemented this yet.
        return stream_llm_chain(topic)
    elif model == "anthropic":
        # Note we haven't implemented this yet
        return stream_anthropic_chain(topic)
    else:
        raise ValueError(
            f"Received invalid model '{model}'."
            " Expected one of chat_openai, openai, anthropic"
        )

def batch_configurable_chain(
    topics: List[str], 
    *, 
    model: str = "chat_openai"
) -> List[str]:
    # You get the idea
    ...

async def abatch_configurable_chain(
    topics: List[str], 
    *, 
    model: str = "chat_openai"
) -> List[str]:
    ...

invoke_configurable_chain("ice cream", model="openai")
stream = stream_configurable_chain(
    "ice_cream", 
    model="anthropic"
)
for chunk in stream:
    print(chunk, end="", flush=True)

# batch_configurable_chain(["ice cream", "spaghetti", "dumplings"])
# await ainvoke_configurable_chain("ice cream")

LCEL로 만드는 경우

from langchain_core.runnables import ConfigurableField


configurable_model = model.configurable_alternatives(
    ConfigurableField(id="model"), 
    default_key="chat_openai", 
    openai=llm,
    anthropic=anthropic,
)
configurable_chain = (
    {"topic": RunnablePassthrough()} 
    | prompt 
    | configurable_model 
    | output_parser
)



configurable_chain.invoke(
    "ice cream", 
    config={"model": "openai"}
)
stream = configurable_chain.stream(
    "ice cream", 
    config={"model": "anthropic"}
)
for chunk in stream:
    print(chunk, end="", flush=True)

configurable_chain.batch(["ice cream", "spaghetti", "dumplings"])

# await configurable_chain.ainvoke("ice cream")

로깅(Logging)

LCEL 없이 만드는 경우

def invoke_anthropic_chain_with_logging(topic: str) -> str:
    print(f"Input: {topic}")
    prompt_value = anthropic_template.format(topic=topic)
    print(f"Formatted prompt: {prompt_value}")
    output = call_anthropic(prompt_value)
    print(f"Output: {output}")
    return output

invoke_anthropic_chain_with_logging("ice cream")

LCEL 로 만드는 경우

import os

os.environ["LANGCHAIN_API_KEY"] = "..."
os.environ["LANGCHAIN_TRACING_V2"] = "true"

anthropic_chain.invoke("ice cream")

Fallbacks

LCEL 없이 만드는 경우

def invoke_chain_with_fallback(topic: str) -> str:
    try:
        return invoke_chain(topic)
    except Exception:
        return invoke_anthropic_chain(topic)

async def ainvoke_chain_with_fallback(topic: str) -> str:
    try:
        return await ainvoke_chain(topic)
    except Exception:
        # Note: we haven't actually implemented this.
        return ainvoke_anthropic_chain(topic)

async def batch_chain_with_fallback(topics: List[str]) -> str:
    try:
        return batch_chain(topics)
    except Exception:
        # Note: we haven't actually implemented this.
        return batch_anthropic_chain(topics)

invoke_chain_with_fallback("ice cream")
# await ainvoke_chain_with_fallback("ice cream")
batch_chain_with_fallback(["ice cream", "spaghetti", "dumplings"]))

LCEL 로 만드는 경우

fallback_chain = chain.with_fallbacks([anthropic_chain])

fallback_chain.invoke("ice cream")
# await fallback_chain.ainvoke("ice cream")
fallback_chain.batch(["ice cream", "spaghetti", "dumplings"])

전체 코드 비교

이렇게 단순한 경우에도 LCEL 체인이 많은 기능들을 잘 압축해서 제공합니다.  체인들이 더 복잡해 줄 수록 이런 간단함은 더욱 특별한 가치를 가지게 됩니다.

LCEL 없이 만드는 경우

from concurrent.futures import ThreadPoolExecutor
from typing import Iterator, List, Tuple

import anthropic
import openai


prompt_template = "Tell me a short joke about {topic}"
anthropic_template = f"Human:\n\n{prompt_template}\n\nAssistant:"
client = openai.OpenAI()
async_client = openai.AsyncOpenAI()
anthropic_client = anthropic.Anthropic()

def call_chat_model(messages: List[dict]) -> str:
    response = client.chat.completions.create(
        model="gpt-3.5-turbo", 
        messages=messages,
    )
    return response.choices[0].message.content

def invoke_chain(topic: str) -> str:
    print(f"Input: {topic}")
    prompt_value = prompt_template.format(topic=topic)
    print(f"Formatted prompt: {prompt_value}")
    messages = [{"role": "user", "content": prompt_value}]
    output = call_chat_model(messages)
    print(f"Output: {output}")
    return output

def stream_chat_model(messages: List[dict]) -> Iterator[str]:
    stream = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=messages,
        stream=True,
    )
    for response in stream:
        content = response.choices[0].delta.content
        if content is not None:
            yield content

def stream_chain(topic: str) -> Iterator[str]:
    print(f"Input: {topic}")
    prompt_value = prompt.format(topic=topic)
    print(f"Formatted prompt: {prompt_value}")
    stream = stream_chat_model([{"role": "user", "content": prompt_value}])
    for chunk in stream:
        print(f"Token: {chunk}", end="")
        yield chunk

def batch_chain(topics: list) -> list:
    with ThreadPoolExecutor(max_workers=5) as executor:
        return list(executor.map(invoke_chain, topics))

def call_llm(prompt_value: str) -> str:
    response = client.completions.create(
        model="gpt-3.5-turbo-instruct",
        prompt=prompt_value,
    )
    return response.choices[0].text

def invoke_llm_chain(topic: str) -> str:
    print(f"Input: {topic}")
    prompt_value = promtp_template.format(topic=topic)
    print(f"Formatted prompt: {prompt_value}")
    output = call_llm(prompt_value)
    print(f"Output: {output}")
    return output

def call_anthropic(prompt_value: str) -> str:
    response = anthropic_client.completions.create(
        model="claude-2",
        prompt=prompt_value,
        max_tokens_to_sample=256,
    )
    return response.completion   

def invoke_anthropic_chain(topic: str) -> str:
    print(f"Input: {topic}")
    prompt_value = anthropic_template.format(topic=topic)
    print(f"Formatted prompt: {prompt_value}")
    output = call_anthropic(prompt_value)
    print(f"Output: {output}")
    return output

async def ainvoke_anthropic_chain(topic: str) -> str:
    ...

def stream_anthropic_chain(topic: str) -> Iterator[str]:
    ...

def batch_anthropic_chain(topics: List[str]) -> List[str]:
    ...

def invoke_configurable_chain(
    topic: str, 
    *, 
    model: str = "chat_openai"
) -> str:
    if model == "chat_openai":
        return invoke_chain(topic)
    elif model == "openai":
        return invoke_llm_chain(topic)
    elif model == "anthropic":
        return invoke_anthropic_chain(topic)
    else:
        raise ValueError(
            f"Received invalid model '{model}'."
            " Expected one of chat_openai, openai, anthropic"
        )

def stream_configurable_chain(
    topic: str, 
    *, 
    model: str = "chat_openai"
) -> Iterator[str]:
    if model == "chat_openai":
        return stream_chain(topic)
    elif model == "openai":
        # Note we haven't implemented this yet.
        return stream_llm_chain(topic)
    elif model == "anthropic":
        # Note we haven't implemented this yet
        return stream_anthropic_chain(topic)
    else:
        raise ValueError(
            f"Received invalid model '{model}'."
            " Expected one of chat_openai, openai, anthropic"
        )

def batch_configurable_chain(
    topics: List[str], 
    *, 
    model: str = "chat_openai"
) -> List[str]:
    ...

async def abatch_configurable_chain(
    topics: List[str], 
    *, 
    model: str = "chat_openai"
) -> List[str]:
    ...

def invoke_chain_with_fallback(topic: str) -> str:
    try:
        return invoke_chain(topic)
    except Exception:
        return invoke_anthropic_chain(topic)

async def ainvoke_chain_with_fallback(topic: str) -> str:
    try:
        return await ainvoke_chain(topic)
    except Exception:
        return ainvoke_anthropic_chain(topic)

async def batch_chain_with_fallback(topics: List[str]) -> str:
    try:
        return batch_chain(topics)
    except Exception:
        return batch_anthropic_chain(topics)

LCEL로 만드는 경우

import os

from langchain_community.chat_models import ChatAnthropic
from langchain_openai import ChatOpenAI
from langchain_openai import OpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough, ConfigurableField

os.environ["LANGCHAIN_API_KEY"] = "..."
os.environ["LANGCHAIN_TRACING_V2"] = "true"

prompt = ChatPromptTemplate.from_template(
    "Tell me a short joke about {topic}"
)
chat_openai = ChatOpenAI(model="gpt-3.5-turbo")
openai = OpenAI(model="gpt-3.5-turbo-instruct")
anthropic = ChatAnthropic(model="claude-2")
model = (
    chat_openai
    .with_fallbacks([anthropic])
    .configurable_alternatives(
        ConfigurableField(id="model"),
        default_key="chat_openai",
        openai=openai,
        anthropic=anthropic,
    )
)

chain = (
    {"topic": RunnablePassthrough()} 
    | prompt 
    | model 
    | StrOutputParser()
)

요약

이번 시간에는 LCEL을 실제로 사용할 때와 안했을 때의 차이 비교를 통해서 LCEL이 제공하는 가치에 대해서 알아보았습니다.  LCEL이 통일된 인터페이스를 제공하고 많은 원시 구성 요소들을 제공하기 때문에 간단한 메소드 호출 만으로 많은 기능을 대신할 수 있음을 사례로 살표 보았습니다. 이러한 사례를 통해서 단순하고 간결한 코드 작성으로 구현이 가능함을 확인 하였습니다. 

Next Steps

LCEL에 대한 학습을 계속하려면 다음을 권장합니다:

여기서 부분적으로 다룬 LCEL 인터페이스에 대해 전체 내용을 자세히 읽어보세요.

LCEL이 제공하는 추가적인 구성 기본 요소에 대해 학습하기 위해서 How-to 섹션을 탐험하세요.

공통 사용 사례에 대한 LCEL 동작을 보려면 Cookbook 섹션을 확인하세요. 그 다음에 살펴볼 좋은 사용 사례는 검색 증강 생성입니다.

LangChain 표현 언어 또는 LCEL은 체인을 쉽게 조립하는 선언적인 방법입니다. LCEL은 상용 서비스부터 프로토 타입을 구현하는 데 코드 변경 없이 지원하도록 처음부터 설계되었습니다. 가장 간단한 "프롬프트 + LLM" 체인에서부터 가장 복잡한 체인까지(사람들이 100개 이상의 단계로 구성된 LCEL 체인을 성공적으로 운영했습니다) 모두 지원합니다. LCEL을 사용하려는 이유 중에서 몇 가지를 강조해보겠습니다:

1. 스트리밍 지원: LCEL로 체인을 빌드하면 첫 번째 출력이 나올 때까지의 경과 시간 최대한 줄일 수 있습니다. 일부 체인의 경우, LLM에서 스트리밍 출력 파서로 토큰을 직접 전송하고, LLM이 원시 토큰을 출력하는 속도와 동일한 속도로 파싱된 증분 출력 묶음(청크)를 얻을 수 있습니다.
2. 비동기 지원: LCEL로 작성된 모든 체인은 동기식 API(예: 프로토타입을 만들 때 Jupyter 노트북에서) 및 비동기식 API(예: LangServe 서버에서)로 호출할 수 있습니다. 이는 프로토타입 및 프로덕션에서 동일한 코드를 사용하여 훌륭한 성능을 내고, 동일한 서버에서 많은 동시 요청을 처리할 수 있게 해줍니다.
3. 최적화된 병렬 실행: LCEL 체인에 병렬로 실행할 수 있는 단계가 있는 경우(예: 여러 Retriver에서 문서를 가져오는 경우), 대기 시간을 최소화 하기 위해서 동기 및 비동기 인터페이스를 자동으로 실행됩니다.
4. 재시도 및 예비 기능: LCEL 체인의 어떤 부분이든 재시도 및 예비 기능을 구성할 수 있습니다. 이것은 규모가 큰 체인을 더 신뢰할 수 있게 만드는 좋은 방법입니다. 현재 재시도/예비 기능에 대한 스트리밍 지원을 추가 중이므로 나중에는 추가된 신뢰성을 얻을 수 있으면서도 동시에 추가 대기 시간이 필요 없습니다.
5. 중간 결과에 대한 액세스: 더 복잡한 체인의 경우 종종 최종 출력이 생성되기 전에 중간 단계의 결과에 액세스하는 것이 매우 유용합니다. 이는 최종 사용자에게 무언가가 발생하고 있음을 알리거나 체인을 디버그하는 데 사용될 수 있습니다. 중간 결과를 스트리밍할 수 있으며 모든 LangServe 서버에서 사용 가능합니다.
6. 입력 및 출력 스키마: 입력 및 출력 스키마는 각 LCEL 체인에 대한 Pydantic 및 JSONSchema 스키마를 제공하며 체인의 구조에서 유추됩니다. 이는 입력 및 출력의 유효성 검사에 사용될 수 있으며 LangServe의 중요한 부분입니다.
7. 원활한 LangSmith 추적 통합: 체인이 더 복잡해질수록 모든 단계를 정확히 이해하는 것이 점점 중요해집니다. LCEL을 사용하면 모든 단계가 최대의 가시성과 디버깅 가능성을 위해 자동으로 LangSmith에 로그됩니다.
8. 원활한 LangServe 배포 통합: LCEL로 생성된 모든 체인은 LangServe를 사용하여 쉽게 배포할 수 있습니다.

LCEL 시작하기 : Get started

LCEL은 기본 콤포넌트를 가지고 복잡한 체인을 쉽게 만들 수 있게 합니다.  그리고 로깅과 병렬처리, 스트리밍과 같은 외부 기능으로의 확대를 지원 합니다.

기본 예제 : 프롬프트 + 모델 + 출력 파서

가장 기본적이고 일반적인 사용 예제는 프롬프트 템플릭과 모델을 함께 체인으로 묶는 것 입니다. 이것이 어떻게 동작하는지 보기 위해서 하나의 토픽을 받아서 하나의 농담을 생성하는 체인을 만들어 보시지요.

from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template("tell me a short joke about {topic}")
model = ChatOpenAI()
output_parser = StrOutputParser()

chain = prompt | model | output_parser

chain.invoke({"topic": "ice cream"})

"Why did the ice cream go to therapy?\n\nBecause it had too many toppings and couldn't find its cone-fidence!"

아래의 코드를 통해서 우리는 다른 콤포넌트들을 LCEL을 이용해서 하나로 묶을 수 있습니다.

chain = prompt | model | output_parser

여기서 | 기호는 유닉스의 파이프 연산자 입니다. 이것은 하나의 콤포넌트에서 나온 아웃풋을 다음 콤포넌트의 입력으로 제공하여 체인을 만듭니다.

이 체인의 입력은 프롬프트 템플릿으로 전달 됩니다. 그리고 그 프롬프트 템플릿의 출력은 모델로 전달 됩니다. 그리고 모델의 출력은 아웃풋 파서에게 전달 됩니다. 무슨일이 벌어지는지 정확히 이해하기 위해서 콤포넌트들을 각각 알아보시죠.

1. Prompt​

프롬프트는 BasePromptTemplate입니다. 이것은 템플릿 변수로 사전(dictionary)을 받아서 PromptValue를 생성합니다. PromptValue는 (입력으로 문자열을 받는) LLM 이나 (입력으로 연속된 메시지를 받는) ChatModel에 전달될 수 있는 완성된 프롬프트로 만드는 랩퍼입니다. 이것은 BaseMessages나 문자열을 만드는 로직을 정의하기 때문에 두 언어모델과 함께 사용 할 수 있습니다.

prompt_value = prompt.invoke({"topic": "ice cream"})
prompt_value

# 출력 내용
ChatPromptValue(messages=[HumanMessage(content='tell me a short joke about ice cream')])

prompt_value.to_messages()

# 출력 내용 : HumanMessage 객체
[HumanMessage(content='tell me a short joke about ice cream')]

prompt_value.to_string()

# 출력 내용: 문자열
'Human: tell me a short joke about ice cream'

2. Model​

그리고나서 PromptValue 는 모델에게 전달 됩니다. 이번에 우리가 사용한 모델이 ChatModel이기 때문에 아웃풋은 BaseMessage일 껍니다.

message = model.invoke(prompt_value)
message

# 출력 내용
AIMessage(content="Why did the ice cream go to therapy? \n\nBecause it had too many toppings and couldn't find its cone-fidence!")

만약에 우리의 모델이 LLM이라면 이것의 출력은 문자열일 껍니다.

from langchain.llms import OpenAI

llm = OpenAI(model="gpt-3.5-turbo-instruct")
llm.invoke(prompt_value)

3. Output parser​

그리고 마지막으로 우리는 우리의 모델 아웃풋을 아웃풋 파서에 전달 합니다. 아웃풋 파써가 BaseOutputParser라는 것은 문자열이나 BaseMessage를 입력으로 받는 다는 것을 의미합니다.

output_parser.invoke(message)

# 출력 내용
"Why did the ice cream go to therapy? \n\nBecause it had too many toppings and couldn't find its cone-fidence!"

4. 전체 Pipeline​

다음의 단계를을 따라 갑니다.:

1. {"topic": "ice cream"}과 같이 원하는 토픽을 사용자 입력으로 전달합니다.
2. 프롬프트는 사용자 입력을 받는데 이것은 토픽을 이용해서 프롬프트를 만든 다음 PromptValue를 생성하는데 사용합니다.
3. 모델 콤포넌트는 생성된 프롬프트를 받아서 OpenAI LLM 모델에 전달합니다. 이 모델의 출력으로 생성된 것은 ChatMessage 오브젝트입니다. 
4. 마지막으로, output_parser 콤포넌트는 ChatMessage 오브젝트를 입력 받고 파이썬 문자열로 변환하여 invoke 메소드로 반환합니다. 

만약에 어떤 콤포넌트든지 출력에 대해서 궁금한 점이 있다면 이 체인의 작은 버전을 언제든지 테스트 할 수 있습니다. 예를 들면 중간 결과를 보기 위해서 prompt 또는 prompt | model 같은 것을 테스트 할 수 있습니다. 

input = {"topic": "ice cream"}

# 프롬프트만 실행
prompt.invoke(input)
# > ChatPromptValue(messages=[HumanMessage(content='tell me a short joke about ice cream')])

# 프롬프트와 모델만 실행
(prompt | model).invoke(input)
# > AIMessage(content="Why did the ice cream go to therapy?\nBecause it had too many toppings and couldn't cone-trol itself!")

RAG 검색 예시

다음 예시로 우리는 질문에 대답할 때 어떤 내용을 추가하는 retrieval-augmented generation 체인을 만들기 원합니다.

# Requires:
# pip install langchain docarray tiktoken

from langchain.chat_models import ChatOpenAI
from langchain.embeddings import OpenAIEmbeddings
from langchain.prompts import ChatPromptTemplate
from langchain.vectorstores import DocArrayInMemorySearch
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel, RunnablePassthrough

vectorstore = DocArrayInMemorySearch.from_texts(
    ["harrison worked at kensho", "bears like to eat honey"],
    embedding=OpenAIEmbeddings(),
)
retriever = vectorstore.as_retriever()

template = """Answer the question based only on the following context:
{context}

Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)
model = ChatOpenAI()
output_parser = StrOutputParser()

setup_and_retrieval = RunnableParallel(
    {"context": retriever, "question": RunnablePassthrough()}
)
chain = setup_and_retrieval | prompt | model | output_parser

chain.invoke("where did harrison work?")

이 경우에 구성된 체인은 다음과 같습니다.

chain = setup_and_retrieval | prompt | model | output_parser

제일 첫번째 단계로 우리는 메모리 저장소를 이용해서 retriever를 설정할 수 있습니다. 이 retriever는 질문에 기반해서 문서를 가져올 수 있습니다. 이것은 다른 콤포넌트들과 함께 묶여질 수 있을 뿐만 아니라 실행될 수 있는 콤포넌트입니다. 뿐만아니라 여러분은 retriever만 분리해서 실행 할 수도 있습니다. 

retriever.invoke("where did harrison work?")

setup_and_retrieval = RunnableParallel(
    {"context": retriever, "question": RunnablePassthrough()}
)

setup_and_retrieval = RunnableParallel(
    {"context": retriever, "question": RunnablePassthrough()}
)
chain = setup_and_retrieval | prompt | model | output_parser

흐름은 다음과 같습니다:

1. 첫 번째 단계에서는 두 항목을 포함하는 RunnableParallel 객체를 만듭니다. 첫 번째 항목인 "context"에는 retriever에 의해 가져온 문서 결과가 포함됩니다. 두 번째 항목인 "question"에는 사용자의 원래 질문이 포함됩니다. 질문을 전달하기 위해 RunnablePassthrough를 사용하여 복사합니다.
2. 위 단계에서 생성된 사전을 프롬프트 컴포넌트에 전달합니다. 그런 다음 사용자 입력인 "question" 과 검색된 문서인 "context"를 사용하여 프롬프트를 구성하고 PromptValue를 출력합니다.
3. 모델 컴포넌트는 생성된 프롬프트를 가져와 OpenAI LLM 모델에 전달하여 실행 평가합니다. 모델에서 생성된 출력은 ChatMessage 객체입니다.
4. 마지막으로 output_parser 구성 요소는 ChatMessage를 입력으로 받아서 이를 Python 문자열로 변환하고, invoke 메서드에 반환합니다.

Next steps​

동일한 기능을 LCEL로 구현할 때의 코드와 LCEL 없이 구현할 때의, 두 코드를 나란히 놓고 비교하는 내용이 있는 '왜 LCEL을 사용하는지'를 읽어 보시기 바랍니다. 

이번 시간을 통해서 LCEL이 무엇인지 알아 보았습니다. 그리고 prompt + model  + output_parsert 형태의 기본 예제를 만들고 이해했으며, 마지막에는 RAG 예제를 통해서 관련된 문서를 입력 프롬프트에서 함께 사용하도록 하는 방법에 대해서 알아 보았습니다.

목차  LangChain - Get started

1. Introduction
2. Installation
3. Quickstart
4. Security (이번 포스팅 내용)

동영상 설명은 아래 링크를 참고해 주세요. 

https://youtu.be/pSjEz7PD_6s

이전 포스팅까지 잘 따라오셨다면 LangChain에 대해서 이미 많은 것을 이해하고 기본적인 내용을 활용할 수 있게 되었다고 할 수 있겠습니다. 직전 내용인 Quickstart의 내용이 살짝 많은 것 처럼 생각될 수 있는데, 이것은 다양한 형태의 사용을 설명하다 보니 조금 복잡해 진것 같습니다. 하여간 집중해서 보셨다면 이제 LangChain을 이용해서 어느정도 간단한 어플리케이션을 개발/ 구성을 하실 수 있게 됐을 것 같습니다.

이번 시간에는 보안(Security)에 대해서 다루어 보겠습니다.

보안(Security)

LangChain은 로컬 및 원격 파일 시스템, API 및 데이터베이스와 같은 다양한 외부 리소스와 통합된 대규모 생태계를 보유하고 있습니다. 이러한 통합은 개발자가 LLM의 강력한 기능을 활용하면서 외부 리소스(서비스)에 액세스하고 상호 작용 할 수 있는 다양한 응용 프로그램을 만들 수 있게 합니다.

예를 들면 아이언맨에 나오는 자비스 같은 것을 말합니다. 자비스는 랭체인을 기반으로 구현할 수 있는 언어 모델 어플리케이션의 좋은 예라고 할 수 있겠습니다. (LLM 이) 사용자(토니)의 말을 잘 알아 듣고 대답을 생성하며, 필요하면 (LangChain 이 하는 것 처럼) 외부의 시스템/서비스에 접속해서 필요한 정보를 가져오거나 필요한 동작을 실행 합니다. 이런 어플리케이션의 특징이 있기 때문에 전통적인 보안의 관점에서 좀더 확장된 시각의 접근이 필요한 것 같습니다. 왜냐하면 하나의 서비스 완성을 위해서 여러 기능의 연결점이 많고, 내부와 외부 서비스로 다양하기 때문입니다. 즉, LLM의 특징과 언제 어떻게 사용될 지 모르는 불확실 성에서 오는 잠재적 위험이라고 할 수 있겠습니다.

최적의 방법

이러한 응용 프로그램을 개발할 때 개발자는 다음과 같은 좋은 보안 관행을 따르는 것이 중요합니다.

• 권한 제한: 응용 프로그램이 필요한 권한을 구체적으로 제한하세요. 넓거나 과도한 권한 부여는 중대한 보안 취약점이 될 수 있습니다. 이러한 취약점을 피하려면 읽기 전용 자격 증명을 사용하거나 민감한 리소스에 대한 액세스를 금지하거나 적절한 샌드박싱 기술(컨테이너 내에서 실행)을 사용하는 것을 고려하십시오. 이처럼 권한을 제한하는 것은 어찌보면 당연한 일이지만 어플리케이션의 구성이 더 복잡해 짐에 따라 더 중요한 포인트입니다. 
• 잠재적인 남용 예측: 인간이 실수할 수 있듯이 대형 언어 모델(LLM)도 실수 할 수 있습니다. 어떤 시스템 액세스 또는 자격 증명이든지 간에 할당된 권한이 허용하는 모든 방식으로 사용될 수 있다고 가정하십시오. 예를 들어 데이터베이스 자격 증명이 데이터 삭제를 허용한다면 해당 자격 증명을 사용할 수 있는 모든 LLM이 실제로 데이터를 삭제할 수 있다고 가정하는 것이 가장 안전합니다.
• 깊이 있는 방어: 어떤 보안 기술도 완벽하지 않습니다. Fine-tuning 및 좋은 체인 디자인은 대형 언어 모델(LLM)이 실수를 할 확률을 줄일 수 있지만 완전히 제거할 수는 없습니다. 보안을 보장하기 위해 단일 방어 계층에 의존하는 대신 여러 겹의 보안 접근 방식을 결합하는 것이 가장 좋습니다. 예를 들어 읽기 전용 권한과 샌드박싱을 모두 사용하여 LLM이 명시적으로 사용하도록 허용된 데이터에만 액세스할 수 있도록 보장할 수 있습니다.

보안 관행을 따르지 않으면 다음과 같은 리스크가 발생할 수 있습니다(당연하지만 이보다 더할 수 있습니다.):

• 데이터 손상 또는 손실.
• 기밀 정보에 대한 무단 액세스.
• 핵심 리소스의 성능 또는 가용성에 대한 손상.

예시 시나리오 및 완화 전략:

• 사용자가 파일 시스템에 액세스 권한이 있는 에이전트에게 삭제해서는 안 될 파일을 삭제하거나 민감한 정보가 포함된 파일의 내용을 읽도록 요청할 수 있습니다. 완화를 위해 에이전트를 특정 디렉토리만 사용하도록 제한하고 읽거나 쓸 수 있는 안전한 파일만 허용하도록 합니다. 에이전트를 컨테이너 내에서 실행하여 추가로 샌드박스 처리하는 것도 고려합니다.
• 사용자가 외부 API에 쓰기 권한이 있는 에이전트에게 악의적인 데이터를 API에 기록하거나 해당 API에서 데이터를 삭제할 수 있습니다. 완화를 위해 에이전트에게 읽기 전용 API 키를 부여하거나 이미 그런 남용에 대한 견딜 수 있는 엔드포인트만 사용하도록 제한할 수 있습니다.
• 사용자가 데이터베이스에 액세스할 수 있는 에이전트에게 테이블을 삭제하거나 스키마를 변형할 수 있습니다. 완화를 위해 에이전트가 액세스해야 하는 테이블에 권한을 제한하고 읽기 전용 자격 증명을 발급하는 것을 고려합니다.

만약 여러분이 회사에서 외부 리소스에 액세스하는 응용 프로그램을 개발 중이라면 회사의 보안 팀과 상의하여 응용 프로그램을 최적으로 설계하고 보안을 강화하는 방법을 결정하는 것이 좋습니다.

취약점 보고하기

보안 취약점이 있다면 security@langchain.dev 로 이메일을 통해 신고해 주세요. 이렇게 하면 문제가 즉각적으로 처리되고 필요한 대응이 이루어집니다.  

기업 솔루션

LangChain은 추가적인 보안 요구사항을 가지고 있는 고객을 위해 기업 솔루션을 제공할 수 있습니다. sales@langchain.dev로 연락주시기 바랍니다.

기타

가장 기본적이면서, 중요하면서, 잊기 쉬운 것이 바로 Key 관리 아닌가 싶습니다. 자체 모델과 서비스를 이용해서 통합하고 어플리케이션을 만들면 Key 관리 Risk가 적겠지만, 쉽고, 빠르고, 높은 품질의 어플리케이션을 만들려면 외부 서비스와의 통합이 필수적이게 되고 이때 Key와 Credential을 사용하게됩니다. 그래서 위의 예시 시나리오에서도 나왔지만 API 사용시 적절한 Key 관리가 매우 중요합니다. 코드에 Key를 포함하지 않는 것은 기본이고, 용도와 최소한의 권한을 부여한 Key를 사용하고 주기적으로 점검하고 관리하는 것이 중요합니다. 

이것으로 LangChain을 사용할 때 보안 관점에서의 필요성과 보안 모범 사례를 알아보았고,

예상 시나리오에 대한 적절한 완화 전략과 기타 보안 관련 내용에 대해서 이야기해 보았습니다.

동영상 설명 : https://youtu.be/pSjEz7PD_6s