with get_openai_callback() as cb:는 Python의 컨텍스트 관리자(context manager)를 사용하여 get_openai_callback 함수가 반환하는 객체(cb)를 생성하고, 그 객체를 사용하는 블록을 정의하는 구문입니다. 이 구문을 이해하기 위해서는 Python의 컨텍스트 관리자가 어떻게 작동하는지와 get_openai_callback이 어떤 역할을 하는지를 아는 것이 중요합니다.

1. 컨텍스트 관리자 (Context Manager)

컨텍스트 관리자는 with 블록의 시작과 종료 시 특정 코드를 자동으로 실행하게 해줍니다. 일반적으로, 컨텍스트 관리자는 자원(resource)을 할당하고 해제하는 작업에 사용됩니다. 예를 들어, 파일을 열고 작업을 한 후 자동으로 파일을 닫는 데 사용할 수 있습니다.

• __enter__(): with 블록이 시작될 때 호출됩니다. 이 메서드는 일반적으로 어떤 자원을 할당하거나 초기화합니다.

• __exit__(): with 블록이 끝날 때 호출됩니다. 이 메서드는 자원을 해제하거나, 예외가 발생했을 때 이를 처리합니다.

2. get_openai_callback의 역할

get_openai_callback은 OpenAI API 호출과 관련된 메트릭을 수집하는 콜백 객체를 반환합니다. 이 콜백 객체는 컨텍스트 관리자에서 사용될 때 API 호출 동안의 토큰 사용량, 비용 등을 추적합니다.

3. with get_openai_callback() as cb:의 의미

• get_openai_callback()은 컨텍스트 관리자 역할을 하는 객체를 반환합니다.

• with 블록이 시작되면, cb 변수에 이 객체가 할당됩니다.

• with 블록 내에서 OpenAI API 호출이 이루어지면, cb 객체는 API 호출 관련 데이터를 수집합니다.

• with 블록이 종료되면, cb 객체는 수집한 데이터를 자동으로 정리하고, 필요한 경우 자원을 해제합니다.

예시 코드 분석

from langchain.llms import OpenAI
from langchain.callbacks import get_openai_callback

llm = OpenAI(model="text-davinci-003")

with get_openai_callback() as cb:
    response = llm("What is the capital of France?")
    print(response)
    print(f"Total Tokens: {cb.total_tokens}")
    print(f"Total Cost: {cb.total_cost}")

• get_openai_callback(): 콜백 객체를 생성하여 반환합니다.

• with ... as cb:: cb 변수에 콜백 객체를 할당하고, with 블록 내에서 이 객체를 사용합니다.

• cb.total_tokens, cb.total_cost: with 블록이 끝난 후, API 호출 동안 사용된 총 토큰 수와 총 비용을 출력합니다.

이 구문을 사용함으로써 개발자는 OpenAI API 호출의 성능을 모니터링하고 리소스 사용량을 효율적으로 관리할 수 있습니다.

get_openai_callback

소스: langchain_community/callbacks/manager.py

from contextlib import contextmanager

@contextmanager
def get_openai_callback() -> Generator[OpenAICallbackHandler, None, None]:
    """Get the OpenAI callback handler in a context manager.
    which conveniently exposes token and cost information.

    Returns:
        OpenAICallbackHandler: The OpenAI callback handler.

    Example:
        >>> with get_openai_callback() as cb:
        ...     # Use the OpenAI callback handler
    """
    cb = OpenAICallbackHandler()
    openai_callback_var.set(cb)
    yield cb
    openai_callback_var.set(None)

Pydantic은 Python에서 데이터 유효성 검사 및 설정 관리를 위한 라이브러리입니다. 주로 FastAPI와 같은 웹 프레임워크와 함께 사용되며, 데이터를 구조화하고 검증하는 데 유용합니다. BaseModel은 Pydantic의 핵심 클래스 중 하나로, 데이터 모델을 정의하는 데 사용됩니다.

Pydantic의 주요 기능

1. 유효성 검사 및 변환: 필드에 대해 타입을 지정하면, 입력 데이터가 자동으로 그 타입으로 변환되며, 유효성 검사가 수행됩니다.

2. 자동 완성 및 타입 힌팅 지원: IDE의 자동 완성과 타입 힌팅을 통해 개발 생산성을 높입니다.

3. 데이터 직렬화 및 역직렬화: 모델 인스턴스를 JSON으로 직렬화하거나 JSON으로부터 역직렬화할 수 있습니다.

4. 데이터 검증 오류 관리: 잘못된 데이터를 입력하면, Pydantic이 자동으로 유효성 검사 오류를 생성합니다.

BaseModel 사용 예시

다음은 Pydantic의 BaseModel을 사용하여 간단한 사용자 데이터를 관리하는 예제입니다.

from pydantic import BaseModel, EmailStr, Field
from typing import Optional

class User(BaseModel):
    id: int
    name: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    age: Optional[int] = Field(None, ge=18)
    is_active: bool = True

# Example usage
user_data = {
    "id": 1,
    "name": "John Doe",
    "email": "johndoe@example.com",
    "age": 25,
}

user = User(**user_data)
print(user)
print(user.dict())

코드 설명

1. 필드 정의:

• id: 정수형 필드.

• name: 길이가 3에서 50 사이인 문자열 필드.

• email: 이메일 형식의 문자열을 요구하는 필드. EmailStr 타입은 이메일 주소가 올바른 형식인지 검증합니다.

• age: 선택적 필드로, 값이 주어지면 18 이상이어야 합니다.

• is_active: 기본값이 True인 불리언 필드.

2. 필드 유효성 검사:

• Field를 사용하여 각 필드에 대한 추가적인 제약 조건을 지정합니다.

3. 데이터 생성 및 출력:

• user_data 딕셔너리를 통해 User 객체를 생성합니다. 생성된 객체를 출력하거나, .dict() 메서드를 사용하여 객체를 딕셔너리 형태로 변환할 수 있습니다.

이와 같이 Pydantic을 사용하면 데이터 모델을 간단하고 명확하게 정의할 수 있으며, 자동으로 타입 변환과 유효성 검사를 수행할 수 있습니다. 이를 통해 데이터 처리의 신뢰성과 안정성을 높일 수 있습니다.

BaseModel은 자동으로 __init__ 을 실행

Pydantic의 BaseModel을 사용하면 클래스 수준에서 필드를 정의할 수 있으며, 이러한 필드는 마치 __init__ 메서드에서 self.name과 같이 인스턴스 변수로 설정된 것처럼 동작합니다. Pydantic은 이러한 필드를 기반으로 자동으로 __init__ 메서드를 생성하고, 필드에 대한 타입 검사를 수행합니다.

이 방식은 일반적인 Python 클래스에서의 인스턴스 변수 설정과는 약간 다릅니다. 일반 Python 클래스에서는 인스턴스 변수를 __init__ 메서드 내에서 self를 통해 설정해야 하지만, Pydantic의 BaseModel을 사용하면 클래스 정의 시 필드의 타입과 기본값을 지정하여 더 간결하고 명확하게 모델을 정의할 수 있습니다.

예시 비교

일반 Python 클래스

class User:
    def __init__(self, id: int, name: str, email: str, age: int, is_active: bool = True):
        self.id = id
        self.name = name
        self.email = email
        self.age = age
        self.is_active = is_active

Pydantic BaseModel

from pydantic import BaseModel, EmailStr, Field
from typing import Optional

class User(BaseModel):
    id: int
    name: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    age: Optional[int] = Field(None, ge=18)
    is_active: bool = True

차이점 설명

• 일반 클래스에서는 __init__ 메서드 내에서 self를 사용하여 인스턴스 변수를 직접 설정합니다.

• Pydantic BaseModel에서는 클래스 정의 시 필드를 직접 설정하고, Pydantic이 자동으로 __init__ 메서드를 생성하여 필드 초기화, 타입 검사, 유효성 검사를 수행합니다.

이렇게 Pydantic의 BaseModel을 사용하면 코드가 더 간결해지며, 데이터 유효성 검사가 자동으로 처리되므로 안전하고 유지보수하기 쉬운 코드를 작성할 수 있습니다.

RunnableConfig는 체인생성후 Runnble의 필드값을 변경하거나, Runnable을 교체하라 수 있다. 

configurable_fields 메서드 이해

Runnable의 특정 필드값을 설정할 수 있도록 해준다. runnable의 .bind 메서드와 관련이 있다. 

  - RunnableSerializable 클래스의 configurable_fileds 메서드 (소스)

  - Model 생성시 configurable_fileds 사설정 -> Model 인스턴스 with_config 런타임 설정

class RunnableSerializable(Serializable, Runnable[Input, Output]):
    """Runnable that can be serialized to JSON."""
    
    def configurable_fields(
        self, **kwargs: AnyConfigurableField
    ) -> RunnableSerializable[Input, Output]:
        """Configure particular Runnable fields at runtime.

        Args:
            **kwargs: A dictionary of ConfigurableField instances to configure.

        Returns:
            A new Runnable with the fields configured.

        .. code-block:: python

            from langchain_core.runnables import ConfigurableField
            from langchain_openai import ChatOpenAI

            model = ChatOpenAI(max_tokens=20).configurable_fields(
                max_tokens=ConfigurableField(
                    id="output_token_number",
                    name="Max tokens in the output",
                    description="The maximum number of tokens in the output",
                )
            )

            # max_tokens = 20
            print(
                "max_tokens_20: ",
                model.invoke("tell me something about chess").content
            )

            # max_tokens = 200
            print("max_tokens_200: ", model.with_config(
                configurable={"output_token_number": 200}
                ).invoke("tell me something about chess").content
            )
        """
        from langchain_core.runnables.configurable import RunnableConfigurableFields

        for key in kwargs:
            if key not in self.__fields__:
                raise ValueError(
                    f"Configuration key {key} not found in {self}: "
                    f"available keys are {self.__fields__.keys()}"
                )

        return RunnableConfigurableFields(default=self, fields=kwargs)

Runnable 클래스의 with_config 메서드 

class Runnable(Generic[Input, Output], ABC):
    ... 중략 ...
    def with_config(
        self,
        config: Optional[RunnableConfig] = None,
        # Sadly Unpack is not well-supported by mypy so this will have to be untyped
        **kwargs: Any,
    ) -> Runnable[Input, Output]:
        """
        Bind config to a Runnable, returning a new Runnable.

        Args:
            config: The config to bind to the Runnable.
            kwargs: Additional keyword arguments to pass to the Runnable.

        Returns:
            A new Runnable with the config bound.
        """
        return RunnableBinding(
            bound=self,
            config=cast(
                RunnableConfig,
                {**(config or {}), **kwargs},
            ),  # type: ignore[misc]
            kwargs={},
        )

실제 예제 

  - Model 생성시 ConfigurableField 통해 설정

  - model 인스턴스 사용시 with_config 통해 설정

from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI

model = ChatOpenAI(temperature=0).configurable_fields(
    temperature=ConfigurableField(
        id="llm_temperature",
        name="LLM Temperature",
        description="The temperature of the LLM",
    )
)

//--- case-1
model.invoke("pick a random number")
// 결과
AIMessage(content='17', response_metadata={'token_usage': {'completion_tokens': 1, 'prompt_tokens': 11, 'total_tokens': 12}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_c2295e73ad', 'finish_reason': 'stop', 'logprobs': None}, id='run-ba26a0da-0a69-4533-ab7f-21178a73d303-0')

//--- case-2
model.with_config(configurable={"llm_temperature": 0.9}).invoke("pick a random number")
// 결과 
AIMessage(content='12', response_metadata={'token_usage': {'completion_tokens': 1, 'prompt_tokens': 11, 'total_tokens': 12}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_c2295e73ad', 'finish_reason': 'stop', 'logprobs': None}, id='run-ba8422ad-be77-4cb1-ac45-ad0aae74e3d9-0')

//--- case-3
prompt = PromptTemplate.from_template("Pick a random number above {x}")
chain = prompt | model
chain.invoke({"x": 0})
// 결과 
AIMessage(content='27', response_metadata={'token_usage': {'completion_tokens': 1, 'prompt_tokens': 14, 'total_tokens': 15}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_c2295e73ad', 'finish_reason': 'stop', 'logprobs': None}, id='run-ecd4cadd-1b72-4f92-b9a0-15e08091f537-0')

//--- case-4
chain.with_config(configurable={"llm_temperature": 0.9}).invoke({"x": 0})
// 결과 
AIMessage(content='35', response_metadata={'token_usage': {'completion_tokens': 1, 'prompt_tokens': 14, 'total_tokens': 15}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_c2295e73ad', 'finish_reason': 'stop', 'logprobs': None}, id='run-a916602b-3460-46d3-a4a8-7c926ec747c0-0')

configurable_alternatives 메서드 이해

Chain에 연결할 Runnable을 교체할 수 있게 한다. 

class RunnableSerializable(Serializable, Runnable[Input, Output]):
    ... 중략 ...
    def configurable_alternatives(
        self,
        which: ConfigurableField,
        *,
        default_key: str = "default",
        prefix_keys: bool = False,
        **kwargs: Union[Runnable[Input, Output], Callable[[], Runnable[Input, Output]]],
    ) -> RunnableSerializable[Input, Output]:
        """Configure alternatives for Runnables that can be set at runtime.

        Args:
            which: The ConfigurableField instance that will be used to select the
                alternative.
            default_key: The default key to use if no alternative is selected.
                Defaults to "default".
            prefix_keys: Whether to prefix the keys with the ConfigurableField id.
                Defaults to False.
            **kwargs: A dictionary of keys to Runnable instances or callables that
                return Runnable instances.

        Returns:
            A new Runnable with the alternatives configured.

        .. code-block:: python

            from langchain_anthropic import ChatAnthropic
            from langchain_core.runnables.utils import ConfigurableField
            from langchain_openai import ChatOpenAI

            model = ChatAnthropic(
                model_name="claude-3-sonnet-20240229"
            ).configurable_alternatives(
                ConfigurableField(id="llm"),
                default_key="anthropic",
                openai=ChatOpenAI()
            )

            # uses the default model ChatAnthropic
            print(model.invoke("which organization created you?").content)

            # uses ChatOpenAI
            print(
                model.with_config(
                    configurable={"llm": "openai"}
                ).invoke("which organization created you?").content
            )
        """
        from langchain_core.runnables.configurable import (
            RunnableConfigurableAlternatives,
        )

        return RunnableConfigurableAlternatives(
            which=which,
            default=self,
            alternatives=kwargs,
            default_key=default_key,
            prefix_keys=prefix_keys,
        )

실제 예제

    - Anthropic 모델 생성시, OpenAI 모델을 alternative로 설정한다. 

    - 상황에 따라 OpenAI 모델을 사용한다. 

from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI

llm = ChatAnthropic(
    model="claude-3-haiku-20240307", temperature=0
).configurable_alternatives(
    # This gives this field an id
    # When configuring the end runnable, we can then use this id to configure this field
    ConfigurableField(id="llm"),
    # This sets a default_key.
    # If we specify this key, the default LLM (ChatAnthropic initialized above) will be used
    default_key="anthropic",
    # This adds a new option, with name `openai` that is equal to `ChatOpenAI()`
    openai=ChatOpenAI(),
    # This adds a new option, with name `gpt4` that is equal to `ChatOpenAI(model="gpt-4")`
    gpt4=ChatOpenAI(model="gpt-4"),
    # You can add more configuration options here
)
prompt = PromptTemplate.from_template("Tell me a joke about {topic}")
chain = prompt | llm


//--- case-1 
# By default it will call Anthropic
chain.invoke({"topic": "bears"})
// 결과
AIMessage(content="Here's a bear joke for you:\n\nWhy don't bears wear socks? \nBecause they have bear feet!\n\nHow's that? I tried to come up with a simple, silly pun-based joke about bears. Puns and wordplay are a common way to create humorous bear jokes. Let me know if you'd like to hear another one!", response_metadata={'id': 'msg_018edUHh5fUbWdiimhrC3dZD', 'model': 'claude-3-haiku-20240307', 'stop_reason': 'end_turn', 'stop_sequence': None, 'usage': {'input_tokens': 13, 'output_tokens': 80}}, id='run-775bc58c-28d7-4e6b-a268-48fa6661f02f-0')

//--- case-2
# We can use `.with_config(configurable={"llm": "openai"})` to specify an llm to use
chain.with_config(configurable={"llm": "openai"}).invoke({"topic": "bears"})
// 결과
AIMessage(content="Why don't bears like fast food?\n\nBecause they can't catch it!", response_metadata={'token_usage': {'completion_tokens': 15, 'prompt_tokens': 13, 'total_tokens': 28}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_c2295e73ad', 'finish_reason': 'stop', 'logprobs': None}, id='run-7bdaa992-19c9-4f0d-9a0c-1f326bc992d4-0')

//--- case-3
# If we use the `default_key` then it uses the default
chain.with_config(configurable={"llm": "anthropic"}).invoke({"topic": "bears"})
// 결과
AIMessage(content="Here's a bear joke for you:\n\nWhy don't bears wear socks? \nBecause they have bear feet!\n\nHow's that? I tried to come up with a simple, silly pun-based joke about bears. Puns and wordplay are a common way to create humorous bear jokes. Let me know if you'd like to hear another one!", response_metadata={'id': 'msg_01BZvbmnEPGBtcxRWETCHkct', 'model': 'claude-3-haiku-20240307', 'stop_reason': 'end_turn', 'stop_sequence': None, 'usage': {'input_tokens': 13, 'output_tokens': 80}}, id='run-59b6ee44-a1cd-41b8-a026-28ee67cdd718-0')

bind 메서드 이해 

Runnable 클래스의 bind 메서드

  - RunnableBinding 클래스 인스턴스를 리턴한다. 

class Runnable(Generic[Input, Output], ABC):
   ... 중략 ...
   def bind(self, **kwargs: Any) -> Runnable[Input, Output]:
        """
        Bind arguments to a Runnable, returning a new Runnable.

        Useful when a Runnable in a chain requires an argument that is not
        in the output of the previous Runnable or included in the user input.

        Args:
            kwargs: The arguments to bind to the Runnable.

        Returns:
            A new Runnable with the arguments bound.

        Example:

        .. code-block:: python

            from langchain_community.chat_models import ChatOllama
            from langchain_core.output_parsers import StrOutputParser

            llm = ChatOllama(model='llama2')

            # Without bind.
            chain = (
                llm
                | StrOutputParser()
            )

            chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
            # Output is 'One two three four five.'

            # With bind.
            chain = (
                llm.bind(stop=["three"])
                | StrOutputParser()
            )

            chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
            # Output is 'One two'

        """
        return RunnableBinding(bound=self, kwargs=kwargs, config={})

RunnableBinding 클래스

    - bind: Bind kwargs to pass to the underlying Runnable when running it.
    - with_config: Bind config to pass to the underlying Runnable when running it.
    - with_listeners:  Bind lifecycle listeners to the underlying Runnable.
    - with_types: Override the input and output types of the underlying Runnable.
    - with_retry: Bind a retry policy to the underlying Runnable.
    - with_fallbacks: Bind a fallback policy to the underlying Runnable.

class RunnableBinding(RunnableBindingBase[Input, Output]):
    """Wrap a Runnable with additional functionality.

    A RunnableBinding can be thought of as a "runnable decorator" that
    preserves the essential features of Runnable; i.e., batching, streaming,
    and async support, while adding additional functionality.

    Any class that inherits from Runnable can be bound to a `RunnableBinding`.
    Runnables expose a standard set of methods for creating `RunnableBindings`
    or sub-classes of `RunnableBindings` (e.g., `RunnableRetry`,
    `RunnableWithFallbacks`) that add additional functionality.

    These methods include:
    - `bind`: Bind kwargs to pass to the underlying Runnable when running it.
    - `with_config`: Bind config to pass to the underlying Runnable when running it.
    - `with_listeners`:  Bind lifecycle listeners to the underlying Runnable.
    - `with_types`: Override the input and output types of the underlying Runnable.
    - `with_retry`: Bind a retry policy to the underlying Runnable.
    - `with_fallbacks`: Bind a fallback policy to the underlying Runnable.

    Example:

    `bind`: Bind kwargs to pass to the underlying Runnable when running it.

        .. code-block:: python

            # Create a Runnable binding that invokes the ChatModel with the
            # additional kwarg `stop=['-']` when running it.
            from langchain_community.chat_models import ChatOpenAI
            model = ChatOpenAI()
            model.invoke('Say "Parrot-MAGIC"', stop=['-']) # Should return `Parrot`
            # Using it the easy way via `bind` method which returns a new
            # RunnableBinding
            runnable_binding = model.bind(stop=['-'])
            runnable_binding.invoke('Say "Parrot-MAGIC"') # Should return `Parrot`

        Can also be done by instantiating a RunnableBinding directly (not recommended):

        .. code-block:: python

            from langchain_core.runnables import RunnableBinding
            runnable_binding = RunnableBinding(
                bound=model,
                kwargs={'stop': ['-']} # <-- Note the additional kwargs
            )
            runnable_binding.invoke('Say "Parrot-MAGIC"') # Should return `Parrot`
    """

<참조>

- LangChain Configurable: https://python.langchain.com/v0.2/docs/how_to/configure/

- LangChain KR: https://wikidocs.net/235704

LCEL로 연결되는 인스턴스는 Runnable을 상속받아야 한다. 공식 문서의 Runnable을 살펴본다. 

Custom Chain 생성을 하려면 Runnable을 상속 구현

  - invoke: abstractmethod로 반드시 구현해야 한다. call the chain on an input (ainvoke)

  - stream: stream back chunks of the response (astream)

  - batch: call the chain on a list of inputs (abatch)

invoke의 Input, Output 타입은 Generic으로 구현체의 타입을 따른다. 현재 Runnable의 invoke로 호출된 Output은 다음 Runnable의 Input 값이 된다. 

class Runnable(Generic[Input, Output], ABC):
    @abstractmethod
    def invoke(self, input: Input, config: Optional[RunnableConfig] = None) -> Output:
        """Transform a single input into an output. Override to implement.

        Args:
            input: The input to the Runnable.
            config: A config to use when invoking the Runnable.
               The config supports standard keys like 'tags', 'metadata' for tracing
               purposes, 'max_concurrency' for controlling how much work to do
               in parallel, and other keys. Please refer to the RunnableConfig
               for more details.

        Returns:
            The output of the Runnable.
        """

Generic[Input, Output] 의 타입표이고, LangChain의 컴포넌트들이 Runable을 상속받고 있다.

  - Prompt, ChatModel, LLM, OutputParser, Retriever, Tool

| pipe operator 이해

Runnable 체인의 연결은 pipe operatore( | ) 또는 Runnable의 .pipe() 메서드를 사용한다. 이는 마치 RxJS의 pipe 연결과 유사하다. 

파이썬은 클래스 인스턴스를 + 기호로 쓸 수 있다. 이때 클래스의 __add__ 를 구현한다. 마찬가지로 | 기호를 사용하기 위해서 __or__ 와 __ror__ 을 구현하고 있다. 

    // a | b -> a.__or__(b)
    def __or__(
        self,
        other: Union[
            Runnable[Any, Other],
            Callable[[Any], Other],
            Callable[[Iterator[Any]], Iterator[Other]],
            Mapping[str, Union[Runnable[Any, Other], Callable[[Any], Other], Any]],
        ],
    ) -> RunnableSerializable[Input, Other]:
        """Compose this Runnable with another object to create a RunnableSequence."""
        return RunnableSequence(self, coerce_to_runnable(other))
        
    // a | b (if not working) -> b.__ror_(a)
    def __ror__(
        self,
        other: Union[
            Runnable[Other, Any],
            Callable[[Other], Any],
            Callable[[Iterator[Other]], Iterator[Any]],
            Mapping[str, Union[Runnable[Other, Any], Callable[[Other], Any], Any]],
        ],
    ) -> RunnableSerializable[Other, Output]:
        """Compose this Runnable with another object to create a RunnableSequence."""
        return RunnableSequence(coerce_to_runnable(other), self)

    // 끝에 dictionary 타입을 Runnable로 만들어 준다. 
    def coerce_to_runnable(thing: RunnableLike) -> Runnable[Input, Output]:
    """Coerce a Runnable-like object into a Runnable.

    Args:
        thing: A Runnable-like object.

    Returns:
        A Runnable.

    Raises:
        TypeError: If the object is not Runnable-like.
    """
    if isinstance(thing, Runnable):
        return thing
    elif is_async_generator(thing) or inspect.isgeneratorfunction(thing):
        return RunnableGenerator(thing)
    elif callable(thing):
        return RunnableLambda(cast(Callable[[Input], Output], thing))
    elif isinstance(thing, dict):
        return cast(Runnable[Input, Output], RunnableParallel(thing))
    else:
        raise TypeError(
            f"Expected a Runnable, callable or dict."
            f"Instead got an unsupported type: {type(thing)}"
        )

이 코드에서 정의된 __or__ 메서드는 Python의 특수 메서드 중 하나로, 비트 연산자 | 를 오버로딩하여 객체 간의 조합을 가능하게 합니다. 이 메서드는 RunnableSerializable 클래스에 정의되어 있으며, 이 클래스의 인스턴스를 다른 객체와 조합하여 새로운 RunnableSequence를 생성합니다.

__or__ 메서드의 의미와 역할

1. __or__ 메서드 오버로딩

• 이 메서드는 | 연산자를 사용했을 때 호출됩니다. 예를 들어, a | b와 같이 사용할 때, a.__or__(b)가 호출됩니다.
• self는 __or__ 메서드가 호출되는 객체(RunnableSerializable의 인스턴스)를 가리킵니다.
• other는 | 연산자의 오른쪽에 있는 객체를 가리킵니다.

2. 매개변수 other의 타입

other는 다음 중 하나일 수 있습니다:

• Runnable[Any, Other]: 제네릭 Runnable 객체로, 임의의 입력(Any)을 받아서 Other 타입의 출력을 반환하는 실행 가능한 객체.
• Callable[[Any], Other]: 함수 또는 람다와 같이, 임의의 입력을 받아 Other 타입의 출력을 반환하는 호출 가능한 객체.
• Callable[[Iterator[Any]], Iterator[Other]]: 이터레이터를 받아 이터레이터를 반환하는 호출 가능한 객체.
• Mapping[str, Union[Runnable[Any, Other], Callable[[Any], Other], Any]]: Runnable이나 Callable 또는 다른 값으로 구성된 맵핑 객체(예: 딕셔너리).

3. 리턴 타입

• __or__ 메서드는 RunnableSerializable[Input, Other] 타입의 객체를 반환합니다. 이 반환 객체는 RunnableSequence로, 두 개의 Runnable 객체를 순차적으로 연결한 것입니다.

4. 코드 동작

• RunnableSequence(self, coerce_to_runnable(other))를 반환하여 두 객체를 조합합니다. 여기서 self는 현재 객체이고, other는 조합할 다른 객체입니다.
• coerce_to_runnable(other)는 other를 Runnable 객체로 변환하는 함수입니다. 만약 other가 Runnable이 아니더라도, 이 함수가 이를 적절한 Runnable 객체로 변환합니다.

__ror__ 메서드의 의미와 역할

1. __ror__ 메서드 오버로딩

• __ror__는 | 연산자가 왼쪽 객체에서 작동하지 않을 때 오른쪽 객체에 의해 호출됩니다. 즉, a | b에서 a가 __or__ 메서드를 정의하지 않았거나 NotImplemented를 반환하면, b.__ror__(a)가 호출됩니다.
• self는 이 메서드가 정의된 클래스의 인스턴스(즉, RunnableSerializable)를 가리킵니다.
• other는 | 연산자의 왼쪽에 있는 객체를 가리킵니다.

2. 매개변수 other의 타입

• other는 다음 중 하나일 수 있습니다:
  • Runnable[Other, Any]: Other 타입의 입력을 받아 Any 타입의 출력을 반환하는 실행 가능한 객체.
  • Callable[[Other], Any]: Other 타입의 입력을 받아 Any 타입의 출력을 반환하는 호출 가능한 객체.
  • Callable[[Iterator[Other]], Iterator[Any]]: Other 타입의 이터레이터를 받아 Any 타입의 이터레이터를 반환하는 호출 가능한 객체.
  • Mapping[str, Union[Runnable[Other, Any], Callable[[Other], Any], Any]]: Runnable이나 Callable 또는 다른 값으로 구성된 맵핑 객체(예: 딕셔너리).

3. 리턴 타입

• 이 메서드는 RunnableSerializable[Other, Output] 타입의 객체를 반환합니다. 반환된 객체는 RunnableSequence로, 이는 두 개의 Runnable 객체를 순차적으로 연결한 것입니다.

4. 코드 동작

• RunnableSequence(coerce_to_runnable(other), self)를 반환하여, other 객체와 self 객체를 결합한 새로운 RunnableSequence를 생성합니다.
• coerce_to_runnable(other)는 other 객체가 Runnable이 아니더라도 이를 Runnable 객체로 변환하는 함수입니다.

이 코드의 동작은 other | self 연산이 수행될 때, other를 Runnable로 변환하고, 이를 self와 결합하여 순차적인 실행 체인을 만드는 것입니다.

pipe 메서드의 의미와 역할

Runnable 의 pipe 메서드 구현의 반환값이 __or__ 의 반환값이 동일하다.

    def pipe(
        self,
        *others: Union[Runnable[Any, Other], Callable[[Any], Other]],
        name: Optional[str] = None,
    ) -> RunnableSerializable[Input, Other]:
        """Compose this Runnable with Runnable-like objects to make a RunnableSequence.

        Equivalent to `RunnableSequence(self, *others)` or `self | others[0] | ...`

        Example:
            .. code-block:: python

                from langchain_core.runnables import RunnableLambda

                def add_one(x: int) -> int:
                    return x + 1

                def mul_two(x: int) -> int:
                    return x * 2

                runnable_1 = RunnableLambda(add_one)
                runnable_2 = RunnableLambda(mul_two)
                sequence = runnable_1.pipe(runnable_2)
                # Or equivalently:
                # sequence = runnable_1 | runnable_2
                # sequence = RunnableSequence(first=runnable_1, last=runnable_2)
                sequence.invoke(1)
                await sequence.ainvoke(1)
                # -> 4

                sequence.batch([1, 2, 3])
                await sequence.abatch([1, 2, 3])
                # -> [4, 6, 8]
        """
        return RunnableSequence(self, *others, name=name)

1. pipe 메서드의 역할

• pipe 메서드는 현재 객체와 다른 Runnable 또는 Callable 객체들을 연결하여 하나의 실행 체인을 만듭니다.
• 이 체인은 여러 단계를 거쳐 데이터를 처리하며, 각 단계는 이전 단계의 출력을 받아 다음 단계의 입력으로 사용합니다.

2. 매개변수

• self: pipe 메서드가 호출되는 객체(즉, RunnableSerializable 클래스의 인스턴스)를 가리킵니다.
• *others: 가변 인자 형태로, 여러 개의 Runnable[Any, Other] 또는 Callable[[Any], Other] 객체들을 받아들입니다. 이 객체들은 순차적으로 연결되어 실행됩니다.
  • Runnable[Any, Other]: 임의의 입력(Any)을 받아 Other 타입의 출력을 생성하는 실행 가능한 객체.
  • Callable[[Any], Other]: 함수를 포함하여 입력을 받아 출력을 반환하는 호출 가능한 객체.
• name (선택적): 실행 체인의 이름을 지정하는 선택적인 매개변수입니다. 이 이름은 디버깅 또는 로깅 목적으로 사용할 수 있습니다.

3. 리턴 타입

• RunnableSerializable[Input, Other]: 여러 단계를 연결한 새로운 RunnableSerializable 객체를 반환합니다. 이 객체는 입력(Input)을 받아 최종 출력(Other)을 생성하는 실행 체인입니다.

4. 코드 동작

• 이 메서드는 현재 객체(self)와 others로 전달된 모든 Runnable 또는 Callable 객체들을 연결하여 하나의 RunnableSequence 객체를 생성합니다.
• RunnableSequence는 각 단계의 출력을 다음 단계의 입력으로 사용하여 최종 결과를 생성합니다.

사용예

from langchain_core.runnables import RunnableParallel

composed_chain_with_pipe = (
    RunnableParallel({"joke": chain})
    .pipe(analysis_prompt)
    .pipe(model)
    .pipe(StrOutputParser())
)

composed_chain_with_pipe.invoke({"topic": "battlestar galactica"})

// 결과
"I cannot reproduce any copyrighted material verbatim, but I can try to analyze the humor in the joke you provided without quoting it directly.\n\nThe joke plays on the idea that the Cylon raiders, who are the antagonists in the Battlestar Galactica universe, failed to locate the human survivors after attacking their home planets (the Twelve Colonies) due to using an outdated and poorly performing operating system (Windows Vista) for their targeting systems.\n\nThe humor stems from the juxtaposition of a futuristic science fiction setting with a relatable real-world frustration – the use of buggy, slow, or unreliable software or technology. It pokes fun at the perceived inadequacies of Windows Vista, which was widely criticized for its performance issues and other problems when it was released.\n\nBy attributing the Cylons' failure to locate the humans to their use of Vista, the joke creates an amusing and unexpected connection between a fictional advanced race of robots and a familiar technological annoyance experienced by many people in the real world.\n\nOverall, the joke relies on incongruity and relatability to generate humor, but without reproducing any copyrighted material directly."

// 또는 일렇게 사용도 가능하다. 
composed_chain_with_pipe = RunnableParallel({"joke": chain}).pipe(
    analysis_prompt, model, StrOutputParser()
)

파이프 예제 및 Runnable 구현체들

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

// model runnable
model = ChatOpenAI(model="gpt-4o-mini")
// prompt runnable
prompt = ChatPromptTemplate.from_template("tell me a joke about {topic}")
// output parser runnable
chain = prompt | model | StrOutputParser()

chain.invoke({"topic": "bears"})

// 결과
"Here's a bear joke for you:\n\nWhy did the bear dissolve in water?\nBecause it was a polar bear!"

Runnable 구현체들은 langchain_core.runnables 폴더의  base.py에 구현되어 있다.

  - RunnableSerializable : Serialize 는 보통 파일, 네트워크 전송을 가능하게 한다. 여기서는 json 으로 serialization이 가능하다. 

  - RunnableSequence : Runnable을 순차 처리한다. 

from langchain_core.runnables import RunnableLambda

def add_one(x: int) -> int:
    return x + 1

def mul_two(x: int) -> int:
    return x * 2

runnable_1 = RunnableLambda(add_one)
runnable_2 = RunnableLambda(mul_two)
sequence = runnable_1 | runnable_2
# Or equivalently:
# sequence = RunnableSequence(first=runnable_1, last=runnable_2)
sequence.invoke(1)
await sequence.ainvoke(1)

// 결과 
4

sequence.batch([1, 2, 3])
await sequence.abatch([1, 2, 3])

// 결과
[4, 6, 8]

  - RunnableParallel  : 예 소스 Runnable을 병렬 처리한다. 

from langchain_core.runnables import RunnableLambda

def add_one(x: int) -> int:
    return x + 1

def mul_two(x: int) -> int:
    return x * 2

def mul_three(x: int) -> int:
    return x * 3

runnable_1 = RunnableLambda(add_one)
runnable_2 = RunnableLambda(mul_two)
runnable_3 = RunnableLambda(mul_three)

sequence = runnable_1 | {  # this dict is coerced to a RunnableParallel
    "mul_two": runnable_2,
    "mul_three": runnable_3,
}
# Or equivalently:
# sequence = runnable_1 | RunnableParallel(
#     {"mul_two": runnable_2, "mul_three": runnable_3}
# )
# Also equivalently:
# sequence = runnable_1 | RunnableParallel(
#     mul_two=runnable_2,
#     mul_three=runnable_3,
# )

sequence.invoke(1)
await sequence.ainvoke(1)

// 결과
{'mul_two': 4, 'mul_three': 6}


sequence.batch([1, 2, 3])
await sequence.abatch([1, 2, 3])

// 결과
[{'mul_two': 4, 'mul_three': 6},
 {'mul_two': 6, 'mul_three': 9},
 {'mul_two': 8, 'mul_three': 12}]

LLM 호출 RunnableParallel 예제

from langchain_core.output_parsers import StrOutputParser

analysis_prompt = ChatPromptTemplate.from_template("is this a funny joke? {joke}")
composed_chain = {"joke": chain} | analysis_prompt | model | StrOutputParser()

// RunnableParallel 로 전환된다. 
composed_chain.invoke({"topic": "bears"})

// 결과 
'Haha, that\'s a clever play on words! Using "polar" to imply the bear dissolved or became polar/polarized when put in water. Not the most hilarious joke ever, but it has a cute, groan-worthy pun that makes it mildly amusing. I appreciate a good pun or wordplay joke.'

chain 의 output 을 lambda 의 input으로 받아 dictionary 를 만들어 체인을 구성할 수도 있다. 

composed_chain_with_lambda = (
    chain
    | (lambda input: {"joke": input})
    | analysis_prompt
    | model
    | StrOutputParser()
)

composed_chain_with_lambda.invoke({"topic": "beets"})

// 결과
"Haha, that's a cute and punny joke! I like how it plays on the idea of beets blushing or turning red like someone blushing. Food puns can be quite amusing. While not a total knee-slapper, it's a light-hearted, groan-worthy dad joke that would make me chuckle and shake my head. Simple vegetable humor!"

  - RunnableLambda  

from langchain_core.runnables import RunnableLambda

def add_one(x: int) -> int:
    return x + 1

runnable = RunnableLambda(add_one)

runnable.invoke(1) # returns 2
runnable.batch([1, 2, 3]) # returns [2, 3, 4]

# Async is supported by default by delegating to the sync implementation
await runnable.ainvoke(1) # returns 2
await runnable.abatch([1, 2, 3]) # returns [2, 3, 4]


# Alternatively, can provide both synd and sync implementations
async def add_one_async(x: int) -> int:
    return x + 1

runnable = RunnableLambda(add_one, afunc=add_one_async)
runnable.invoke(1) # Uses add_one
await runnable.ainvoke(1) # Uses add_one_async

  - RunnableGenerator

from typing import Any, AsyncIterator, Iterator

from langchain_core.runnables import RunnableGenerator


def gen(input: Iterator[Any]) -> Iterator[str]:
    for token in ["Have", " a", " nice", " day"]:
        yield token


runnable = RunnableGenerator(gen)
runnable.invoke(None)  # "Have a nice day"
list(runnable.stream(None))  # ["Have", " a", " nice", " day"]
runnable.batch([None, None])  # ["Have a nice day", "Have a nice day"]


# Async version:
async def agen(input: AsyncIterator[Any]) -> AsyncIterator[str]:
    for token in ["Have", " a", " nice", " day"]:
        yield token

runnable = RunnableGenerator(agen)
await runnable.ainvoke(None)  # "Have a nice day"
[p async for p in runnable.astream(None)] # ["Have", " a", " nice", " day"]

  - 그외 RunnableEachBase, RunableEach, RunnableBindingBase, RunnableBinding

모든 Runnable은 input 과 output schema를 노출한다. 

   - input_schema: input pydantic model

   - output_schema: output pydantic model

예제는 주로 invoke, ainvoke를 한번에 output을 받는 것이고, LLM을 통해 응답을 받을때는 UX개선을 위하 stream으로 받아 처리한다. Runnable stream 처리에 대한 다양한 접근법은 공식문서를 참조한다. 

Runnable 개념 정리

LangChain Runnable과 LangChain Expression Language (LCEL)은 LLM(대형 언어 모델)의 기능을 활용하여 견고한 프로덕션 수준의 프로그램을 만들 수 있는 강력한 도구입니다. 다음은 이들의 동작 방식과 중요성에 대한 개요입니다.

LangChain Runnable

LangChain Runnable은 실행할 수 있는 작업이나 기능을 캡슐화한 개념으로, 일반적으로 일련의 계산 단계를 나타냅니다. Runnable은 LangChain에서 복잡한 워크플로우 또는 체인을 구성할 수 있는 기본 구성 요소입니다. 각 Runnable은 유연하게 설계되어 동기, 비동기, 배치, 스트리밍 작업을 처리할 수 있습니다.

LangChain Expression Language (LCEL)

LangChain Expression Language (LCEL)은 이러한 워크플로우 또는 프로그램을 구성할 수 있는 선언적 언어입니다. LCEL을 사용하면 여러 Runnable을 명확하고 구조적인 방식으로 결합할 수 있습니다. LCEL을 통해 복잡한 작업 체인을 정의하고 여러 Runnable을 하나의 프로그램으로 결합할 수 있습니다.

LCEL과 Runnables의 주요 기능

1. 동기 작업: 전통적인 차단 작업으로, 각 단계가 완료된 후 다음 단계로 이동합니다.
2. 비동기 작업: 차단되지 않는 작업으로, 특정 작업이 완료되기를 기다리는 동안 프로그램이 다른 작업을 처리할 수 있습니다. 이는 서버 환경에서 여러 요청을 동시에 처리할 때 특히 중요합니다.
3. 배치 처리: 여러 입력을 동시에 처리할 수 있는 기능입니다. 대량의 데이터를 처리하거나 여러 사용자의 쿼리를 동시에 처리할 때 유용합니다.
4. 스트리밍: 중간 결과가 생성됨에 따라 이를 스트리밍할 수 있는 기능으로, 사용자 경험을 개선할 수 있습니다. 예를 들어, LLM으로 텍스트를 생성할 때 전체 텍스트가 생성되기 전에 부분적인 응답을 사용자에게 스트리밍할 수 있습니다.

사용 예시

• 배치 처리: LLM을 통해 고객 쿼리 데이터를 분류해야 하는 시나리오를 상상해 보십시오. LCEL과 Runnables을 사용하여 이 쿼리들을 배치 처리하고 병렬로 처리할 수 있어 작업 속도가 크게 향상됩니다.
• 비동기 작업: 여러 클라이언트 요청을 처리하는 웹 서버에서 LCEL을 사용하여 이러한 요청을 비동기로 관리함으로써, 서버가 여러 트래픽을 동시에 처리할 수 있게 합니다.
• 스트리밍 결과: 실시간으로 사용자와 상호작용하는 챗봇의 경우, LCEL을 사용하여 부분적으로 생성된 텍스트로 바로 응답을 시작함으로써 더 인터랙티브하고 반응성이 좋은 사용자 경험을 제공할 수 있습니다.

프로덕션 수준 프로그램에서의 LCEL의 역할

LCEL의 설계는 LLM의 기능을 활용하는 복잡한 워크플로우를 보다 쉽게 구축하고 관리할 수 있도록 하는 데 중점을 둡니다. 이는 다양한 작업 모드(동기, 비동기, 배치, 스트리밍)를 처리하는 데 있어 복잡성을 추상화하여, 개발자가 워크플로우의 논리 정의에 집중할 수 있도록 합니다. 이러한 접근 방식은 특히 신뢰성, 확장성, 성능이 중요한 프로덕션 환경에서 매우 가치가 있습니다.

구현 및 스키마

LangChain Runnables 및 LCEL의 구현은 프로그램의 다양한 구성 요소가 어떻게 상호작용하는지를 설명하는 스키마를 정의하는 것입니다. 이러한 스키마는 체인의 각 구성 요소가 올바른 유형의 입력을 받고 예상되는 유형의 출력을 생성하도록 보장합니다. 이러한 구조화된 접근 방식은 신뢰할 수 있는 프로그램을 구축하는 데 도움이 될 뿐만 아니라, 프로그램을 유지하고 확장하는 데도 유리합니다.

결론

LangChain Runnables과 LCEL은 LLM을 통합한 워크플로우를 구축하고 관리할 수 있는 강력한 패러다임을 제공합니다. 이들은 다양한 작업 모드를 지원하는 유연하고 확장 가능한 프레임워크를 제공하여, 복잡한 데이터 처리 작업을 간단하게 구성하고 강력한 처리 체인을 만들 수 있습니다. LCEL과 Runnables을 활용하여 효율적이고 반응성이 뛰어난 확장 가능한 솔루션을 만들 수 있습니다.

<참조>

- Runnable Conceptual Guide: https://python.langchain.com/v0.2/docs/concepts/#runnable-interface

- How-to Guides Runnable Sequence: https://python.langchain.com/v0.2/docs/how_to/sequence/

- How-to Guides Runnable stream: https://python.langchain.com/v0.2/docs/how_to/streaming/

Python에서 패키지(package)와 모듈(module)은 모두 코드를 조직화하고 재사용성을 높이기 위해 사용되는 구조적 요소입니다. 이 둘은 Python 코드베이스를 더 잘 구조화하고 관리하기 위한 단위이지만, 각각 다른 개념을 나타냅니다.

모듈 (Module)

• 모듈은 Python 코드의 파일로, 함수, 클래스, 변수, 또는 다른 코드 블록들이 포함된 단위입니다.

• 각 모듈은 .py 확장자를 가진 파일로 저장되며, 이 파일의 이름이 모듈의 이름이 됩니다.

• 모듈은 다른 모듈에서 import 문을 사용하여 불러올 수 있습니다.

예시

# my_module.py
def greet(name):
    return f"Hello, {name}!"

# 다른 파일에서 모듈을 불러오기
import my_module

print(my_module.greet("Alice"))  # 출력: Hello, Alice!

위의 예에서 my_module.py 파일이 모듈입니다. 이 모듈은 greet라는 함수를 포함하고 있으며, 다른 Python 파일에서 import를 통해 이 함수를 사용할 수 있습니다.

패키지 (Package)

• 패키지는 여러 모듈을 포함할 수 있는 디렉토리로, Python에서는 __init__.py 파일을 포함한 디렉토리를 패키지로 인식합니다.

• 패키지는 서브 모듈을 계층적으로 조직할 수 있게 해줍니다. 즉, 패키지 안에 또 다른 패키지를 포함하여 하위 모듈을 관리할 수 있습니다.

• __init__.py 파일은 해당 디렉토리를 패키지로 인식하게 합니다. Python 3.3 이후로는 이 파일이 없어도 패키지로 인식되지만, 여전히 패키지 초기화 작업을 위해 종종 사용됩니다.

예시

my_package/
    __init__.py
    module1.py
    module2.py
    sub_package/
        __init__.py
        module3.py

위의 구조에서 my_package는 패키지입니다. 이 패키지에는 module1.py와 module2.py라는 두 개의 모듈과, sub_package라는 하위 패키지가 포함되어 있습니다. 하위 패키지 sub_package에는 module3.py라는 모듈이 포함되어 있습니다.

이러한 구조에서는 다음과 같이 모듈을 가져올 수 있습니다:

from my_package import module1
from my_package.sub_package import module3

패키지와 모듈의 차이점

• 구조:

    • 모듈은 단일 .py 파일로 구성됩니다.

    • 패키지는 여러 모듈(그리고 하위 패키지)을 포함할 수 있는 디렉토리입니다.

• 범위:

    • 모듈은 단일 파일에 포함된 코드 단위입니다.

    • 패키지는 모듈과 하위 패키지를 조직화하는 더 큰 단위입니다.

• 내포 관계:

    • 모듈은 패키지 안에 포함될 수 있습니다.

    • 패키지는 모듈들을 그룹화하는 역할을 하며, 패키지 내에 또 다른 패키지를 포함할 수도 있습니다.

요약

• 모듈: Python 코드가 담긴 단일 .py 파일. 모듈은 함수, 클래스, 변수 등의 코드를 포함할 수 있으며, 다른 Python 코드에서 import를 통해 재사용할 수 있습니다.

• 패키지: 모듈들을 조직화하는 디렉토리 구조. 패키지는 모듈과 다른 하위 패키지를 포함할 수 있으며, 패키지로 인식되기 위해서는 일반적으로 __init__.py 파일이 필요합니다.

LangChain의 ChatPromptTemplate의 __init__ 인수에 대한 부분을 보다 * 가 중간에 오길래 무언가 궁금해졌다. 

class ChatPromptTemplate(BaseChatPromptTemplate):
    def __init__(
        self,
        messages: Sequence[MessageLikeRepresentation],
        *,
        template_format: Literal["f-string", "mustache", "jinja2"] = "f-string",
        **kwargs: Any,
    ) -> None:

이 코드 스니펫은 Python에서 클래스의 생성자 메소드인 __init__ 메소드를 정의하는 부분입니다. 이 메소드는 객체가 생성될 때 호출되며, 객체를 초기화하는 역할을 합니다. 각 매개변수의 의미를 살펴보면 다음과 같습니다.

파라미터 설명

1. self:

• 모든 인스턴스 메소드에서 첫 번째 인수로 self가 사용됩니다. 이는 객체 자신을 참조하며, 생성된 인스턴스에 접근할 수 있게 합니다.

• __init__ 메소드 내에서 self를 통해 객체의 속성을 설정할 수 있습니다.

2. messages: Sequence[MessageLikeRepresentation]:

• messages는 생성자에 전달되는 첫 번째 인수이며, Sequence 타입으로, MessageLikeRepresentation 타입의 객체들로 이루어진 순차적 자료형입니다.

• Sequence는 리스트, 튜플과 같은 순차적 데이터 구조를 포함하는 추상화된 타입입니다. 이 타입 힌트는 messages가 리스트나 튜플과 같은 자료형일 것이라는 것을 의미합니다.

• MessageLikeRepresentation은 사용자 정의 클래스이거나, 메시지를 표현하는 타입일 가능성이 있습니다.

3. template_format: Literal["f-string", "mustache", "jinja2"] = "f-string":

• template_format는 키워드 인수로, 세 가지 문자열 값 중 하나를 가질 수 있습니다: "f-string", "mustache", "jinja2".

• Literal은 Python의 typing 모듈에서 제공하는 기능으로, 특정 값의 집합 중 하나를 선택하도록 제한합니다.

• 이 인수는 기본값으로 "f-string"을 가지며, 따라서 사용자가 특별히 지정하지 않으면 "f-string"이 사용됩니다.

• 이 매개변수는 아마도 템플릿을 처리하는 방식을 지정하는 데 사용될 것입니다.

4. **kwargs: Any:

• **kwargs는 임의의 추가적인 키워드 인수들을 받아들이는 데 사용됩니다.

• kwargs는 딕셔너리 형태로 전달되며, 키는 문자열이고 값은 임의의 타입을 가질 수 있습니다(Any).

• **kwargs는 종종 유연성을 제공하여 함수나 메소드가 예상치 못한 추가 인수를 받아들일 수 있게 합니다.

5. -> None:

• 이 부분은 함수의 반환 타입 힌트를 나타내며, None이 반환됨을 의미합니다.

• 생성자 메소드인 __init__은 객체를 초기화할 뿐, 별도의 값을 반환하지 않으므로 항상 None을 반환합니다.

이 메소드의 역할

• 이 __init__ 메소드는 객체가 생성될 때 호출되며, messages, template_format, 그리고 추가적인 키워드 인수(kwargs)를 사용하여 객체의 초기 상태를 설정합니다.

• messages는 메시지들을 담은 시퀀스 타입의 인수이며, template_format은 메시지를 템플릿으로 처리하는 방식(예: "f-string", "mustache", "jinja2")을 지정합니다.

• kwargs는 추가적인 옵션이나 설정을 받아들일 수 있도록 하여 메소드를 유연하게 만듭니다.

class MyTemplate:
    def __init__(
        self,
        messages: Sequence[str],
        *,
        template_format: Literal["f-string", "mustache", "jinja2"] = "f-string",
        **kwargs: Any,
    ) -> None:
        self.messages = messages
        self.template_format = template_format
        self.options = kwargs

# 사용 예시
template = MyTemplate(
    messages=["Hello, {name}!", "Goodbye, {name}!"],
    template_format="jinja2",
    option1="value1",
    option2="value2"
)

print(template.messages)  # ['Hello, {name}!', 'Goodbye, {name}!']
print(template.template_format)  # 'jinja2'
print(template.options)  # {'option1': 'value1', 'option2': 'value2'}

아래 사용법 3가지 중 마지막번의 사용법임을 알 수 있다. 

* 인수 사용법

함수 정의에서 파라미터 앞에 붙는 *는 여러 가지 의미를 가질 수 있지만, 일반적으로 **위치 인수(variable-length positional arguments)**를 나타내는 데 사용됩니다. 이 기호는 함수가 호출될 때 임의의 개수의 위치 인수를 받아들이도록 하는 역할을 합니다.

*의 주요 의미

1. 가변 위치 인수 (*args):

• 함수 정의에서 *args와 같은 형태로 사용되며, 이 함수는 호출 시 여러 개의 위치 인수를 하나의 튜플로 받아들일 수 있습니다.

• args는 이름일 뿐이며, 어떤 이름으로도 사용할 수 있습니다.

def example_function(*args):
    print(args)

example_function(1, 2, 3)  # (1, 2, 3)
example_function('a', 'b', 'c')  # ('a', 'b', 'c')

여기서 args는 튜플이며, 함수에 전달된 모든 위치 인수들을 포함합니다.

2. 위치 인수 언패킹:

• 함수 호출 시, 이미 정의된 시퀀스(리스트나 튜플 등)를 개별 위치 인수로 전달할 때 사용됩니다.

def add(a, b, c):
    return a + b + c

numbers = (1, 2, 3)
print(add(*numbers))  # 6

여기서 *numbers는 튜플 numbers를 개별 인수 1, 2, 3으로 언패킹하여 함수에 전달합니다.

3. 키워드 전용 인수:

• 함수 정의에서 *는 특정 위치 이후의 모든 인수들이 키워드 인수로만 전달될 수 있음을 나타냅니다. 즉, 이 인수들은 반드시 키워드=값 형태로 지정해야 합니다.

def example_function(a, b, *, c, d):
    print(a, b, c, d)

example_function(1, 2, c=3, d=4)  # 올바름
# example_function(1, 2, 3, 4)  # 오류: c와 d는 키워드 인수로만 사용 가능

여기서 *는 c와 d 인수가 키워드 인수로만 전달되어야 함을 의미합니다.

요약

• *는 함수 정의에서 임의의 개수의 위치 인수를 받아들이거나, 키워드 전용 인수를 지정할 때 사용됩니다.

• 또한 함수 호출 시 시퀀스를 언패킹하여 개별 위치 인수로 전달할 때도 사용됩니다.

• 이 기호는 Python의 함수 정의 및 호출에서 매우 유연하고 강력한 기능을 제공합니다.

LangChain 소스 코드를 분석하면서 Python도 함께 배우고 있습니다. 이해되지 않는 것으 GPT를 통해 열심히 개념을 알아가고 있습니다.

1. Union에 대한 궁금증

Union은 Python의 타입 힌팅(type hinting)에서 사용되며, 함수의 매개변수나 반환 값이 둘 이상의 서로 다른 타입을 가질 수 있음을 나타낼 때 사용됩니다. Python 3.10 이전에는 typing 모듈의 Union을 사용했으며, Python 3.10부터는 더 간단한 | 연산자를 사용하여 동일한 기능을 제공할 수 있습니다.

예시

1. Python 3.9 이하에서의 Union 사용:

from typing import Union

def process_value(value: Union[int, float]) -> Union[int, float]:
    return value * 2

이 함수 process_value는 int나 float 타입의 값을 받아서 두 배로 증가시킨 후 반환합니다. 반환 값도 int나 float이 될 수 있음을 Union을 통해 명시하고 있습니다.

2. Python 3.10 이후에서의 | 연산자 사용:

def process_value(value: int | float) -> int | float:
    return value * 2

위와 동일한 기능을 | 연산자를 사용하여 간단히 표현할 수 있습니다.

Union 사용이 적합한 경우

• 다양한 타입을 처리할 때: 함수나 메소드가 여러 타입을 받을 수 있고, 그 타입에 따라 다르게 동작할 때 Union을 사용합니다.

• 점진적 타입 검사를 통해 오류를 줄일 때: 코드에 타입 힌트를 명확히 하여, 코드의 가독성과 안정성을 높이고, 코드 편집기나 정적 분석 도구가 더 나은 오류 검출을 할 수 있도록 도울 수 있습니다.

주의사항

• Union을 사용할 때, 가능한 한 사용 범위를 좁히는 것이 좋습니다. 너무 많은 타입을 허용하면 코드가 복잡해지고 유지보수가 어려워질 수 있습니다.

Union은 주로 함수의 입력과 출력에 다양한 타입을 허용할 때 매우 유용한 도구입니다. 필요에 따라 적절히 사용하는 것이 중요합니다.

2. ininstance 에 대한 궁금증

isinstance는 Python에서 특정 객체가 특정 클래스나 타입의 인스턴스인지 확인할 때 사용되는 내장 함수입니다. 이 함수는 객체와 클래스(또는 클래스의 튜플)를 인수로 받아, 객체가 해당 클래스의 인스턴스이거나 그 클래스의 서브클래스의 인스턴스인 경우 True를 반환하고, 그렇지 않으면 False를 반환합니다.

isinstance의 기본 사용법

result = isinstance(object, classinfo)

• object: 타입을 확인하려는 객체.

• classinfo: 클래스, 클래스의 튜플, 또는 타입들.

예시

1. 단일 클래스에 대한 검사:

# 객체가 정수형인지 확인
x = 10
print(isinstance(x, int))  # True

2. 여러 클래스에 대한 검사:

# 객체가 정수형 또는 실수형인지 확인
x = 10.5
print(isinstance(x, (int, float)))  # True

4. 상속 관계에서의 검사:

class Animal:
    pass

class Dog(Animal):
    pass

dog = Dog()
print(isinstance(dog, Animal))  # True
print(isinstance(dog, Dog))     # True

이 예제에서 Dog 클래스는 Animal 클래스를 상속받기 때문에, dog 객체는 Animal 클래스의 인스턴스이기도 합니다.

isinstance의 주요 용도

• 타입 확인: 함수나 메소드 내부에서 인수의 타입을 확인하여, 타입에 따라 다른 로직을 수행할 때 유용합니다.

• 상속 관계 검사: 객체가 특정 클래스에서 파생된 클래스의 인스턴스인지 확인할 수 있습니다.

• 입력 검증: 함수에 전달된 인수의 타입이 예상한 타입과 일치하는지 검증하여, 타입 오류를 미리 방지할 수 있습니다.

isinstance와 type의 차이점

isinstance는 상속 관계를 고려하는 반면, type 함수는 객체의 정확한 타입을 확인할 때 사용됩니다. 예를 들어, 상속받은 클래스의 인스턴스는 type을 사용하면 부모 클래스와 매칭되지 않습니다.

class Animal:
    pass

class Dog(Animal):
    pass

dog = Dog()

# isinstance를 사용하면 True
print(isinstance(dog, Animal))  # True

# type을 사용하면 False
print(type(dog) == Animal)  # False, Dog != Animal

이와 같이, isinstance는 더 유연하게 타입을 확인할 수 있는 반면, type은 정확한 타입 비교에 사용됩니다.

1. Dictionary 병합

LangChain 소스를 분석하다 이런 코드를 보았다.

class ChatPromptTemplate(BaseChatPromptTemplate):
    def __init__(
        self,
        messages: Sequence[MessageLikeRepresentation],
        *,
        template_format: Literal["f-string", "mustache", "jinja2"] = "f-string",
        **kwargs: Any,
    ) -> None:
        ...
        // 파이썬 초보자가 봤을 때 이게 뭐지?  
        kwargs = {
            **dict(
                input_variables=sorted(input_vars),
                optional_variables=sorted(optional_variables),
                partial_variables=partial_vars,
            ),
            **kwargs,
        }

이 코드에서 kwargs 변수를 정의하는 방식은 Python에서 딕셔너리를 병합하는 기법 중 하나입니다. 이 코드의 주요 목적은 두 개의 딕셔너리를 결합하여 하나의 딕셔너리를 만드는 것입니다. 여기서 kwargs는 일반적으로 함수나 메소드에서 사용되는 키워드 인수들을 담고 있는 딕셔너리입니다.

코드의 의미 분석

kwargs = {
    **dict(
        input_variables=sorted(input_vars),
        optional_variables=sorted(optional_variables),
        partial_variables=partial_vars,
    ),
    **kwargs,
}

1. dict(...):
   • dict(...) 표현식은 새로운 딕셔너리를 생성합니다.
   • 이 딕셔너리는 세 개의 키-값 쌍을 가지고 있습니다:
     • input_variables: input_vars를 정렬한 리스트.
     • optional_variables: optional_variables를 정렬한 리스트.
     • partial_variables: partial_vars라는 변수에 담긴 값.
2. sorted(input_vars)와 sorted(optional_variables):
   • sorted(input_vars)와 sorted(optional_variables)는 각각 input_vars와 optional_variables 리스트를 정렬한 결과를 반환합니다.
   • 이 결과는 새로운 딕셔너리의 값으로 사용됩니다.
3. **dict(...):
   • ** 연산자는 딕셔너리 언패킹 연산자입니다. 이 연산자를 사용하면 딕셔너리의 키-값 쌍을 다른 딕셔너리로 풀어서 넣을 수 있습니다.
   • **dict(...)는 input_variables, optional_variables, partial_variables라는 키와 그에 대응하는 값들을 풀어서 {}로 감싸인 새로운 딕셔너리로 만듭니다.
4. **kwargs:
   • 기존의 kwargs 딕셔너리도 **kwargs로 언패킹되어 추가됩니다.
   • 이때, 만약 kwargs 딕셔너리에 input_variables, optional_variables, partial_variables와 동일한 키가 존재하면, kwargs 딕셔너리의 값이 우선하여 기존 값을 덮어씁니다.
5. 최종 kwargs 딕셔너리:
   • 최종적으로, 첫 번째 dict(...)로 생성된 딕셔너리와 기존의 kwargs 딕셔너리가 병합된 새로운 kwargs 딕셔너리가 만들어집니다.
   • 이 딕셔너리는 함수나 메소드에 전달될 키워드 인수들로 사용될 수 있습니다.

예시

input_vars = ['var3', 'var1', 'var2']
optional_variables = ['opt3', 'opt1', 'opt2']
partial_vars = {'part1': 'value1'}

kwargs = {'input_variables': ['override_var'], 'new_key': 'new_value'}

kwargs = {
    **dict(
        input_variables=sorted(input_vars),
        optional_variables=sorted(optional_variables),
        partial_variables=partial_vars,
    ),
    **kwargs,
}

print(kwargs)

위 코드를 실행하면, 최종 kwargs 딕셔너리는 다음과 같이 출력됩니다:

{
    'input_variables': ['override_var'],  # kwargs에서 덮어씌워짐
    'optional_variables': ['opt1', 'opt2', 'opt3'],
    'partial_variables': {'part1': 'value1'},
    'new_key': 'new_value'
}

요약

• 이 코드는 dict(...)로 만든 딕셔너리와 기존 kwargs 딕셔너리를 병합하여 새로운 kwargs 딕셔너리를 생성합니다.
• **dict(...)와 **kwargs를 사용하여 두 딕셔너리를 결합합니다. 중복되는 키가 있을 경우, kwargs에 있는 값이 우선합니다.
• 이 기법은 키워드 인수를 동적으로 처리할 때 유용합니다.

2. sorted 용법에 대한 궁금증

sorted는 Python의 내장 함수 중 하나로, 주어진 iterable(리스트, 튜플, 문자열 등)의 요소들을 정렬하여 새로운 리스트로 반환하는 함수입니다. sorted는 원본 데이터를 변경하지 않고, 정렬된 새로운 리스트를 반환합니다.

sorted 함수의 기본 사용법

sorted(iterable, key=None, reverse=False)

• iterable: 정렬할 대상이 되는 iterable 객체(예: 리스트, 튜플, 문자열 등).

• key: 정렬 기준을 지정하는 함수. 각 요소에 대해 이 함수가 호출된 결과를 기준으로 정렬합니다.

• reverse: True로 설정하면 내림차순으로 정렬하고, 기본값인 False는 오름차순으로 정렬합니다.

예시

1. 리스트 정렬:

numbers = [3, 1, 4, 1, 5, 9, 2, 6, 5]
sorted_numbers = sorted(numbers)
print(sorted_numbers)  # [1, 1, 2, 3, 4, 5, 5, 6, 9]

2. 문자열 정렬:

word = "python"
sorted_letters = sorted(word)
print(sorted_letters)  # ['h', 'n', 'o', 'p', 't', 'y']

3. 튜플 정렬:

tuples = [(1, 'b'), (3, 'a'), (2, 'c')]
sorted_tuples = sorted(tuples)
print(sorted_tuples)  # [(1, 'b'), (2, 'c'), (3, 'a')]

4. key 매개변수 사용:

• key 매개변수는 정렬 기준을 정의하는 함수입니다. 각 요소에 대해 이 함수가 호출된 결과를 기준으로 정렬됩니다.

words = ["apple", "banana", "cherry", "date"]
sorted_words = sorted(words, key=len)
print(sorted_words)  # ['date', 'apple', 'banana', 'cherry']

5. reverse=True 사용:

• 정렬 결과를 내림차순으로 반환합니다.

numbers = [3, 1, 4, 1, 5, 9, 2, 6, 5]
sorted_numbers_desc = sorted(numbers, reverse=True)
print(sorted_numbers_desc)  # [9, 6, 5, 5, 4, 3, 2, 1, 1]

요약

• sorted 함수는 iterable의 요소를 정렬하여 새로운 리스트로 반환합니다.

• key 매개변수를 사용하여 정렬 기준을 설정할 수 있으며, reverse 매개변수를 사용하여 정렬 방향(오름차순/내림차순)을 설정할 수 있습니다.

• 원본 데이터는 변경되지 않으며, 정렬된 새로운 리스트가 반환됩니다.

3. cast 에 대한 궁금증

cast는 Python의 타입 힌팅(type hinting)과 관련된 기능 중 하나로, typing 모듈에서 제공하는 함수입니다. cast는 특정 값을 지정된 타입으로 “캐스팅”한다고 표시하는 역할을 합니다. 그러나 실제로 값을 변환하거나 변경하지는 않으며, 주로 타입 힌팅을 통해 코드의 가독성을 높이고, 정적 분석 도구들이 올바른 타입을 추론하도록 돕기 위한 목적으로 사용됩니다.

cast의 기본 사용법

from typing import cast

cast(typ, val)

• typ: 캐스팅할 타입. Python의 타입(예: int, str, List[int] 등)이나 사용자 정의 클래스 등을 지정할 수 있습니다.

• val: 실제 값. typ로 캐스팅할 값입니다.

cast의 역할

• 정적 분석 지원: cast는 Python 코드에서 변수나 표현식의 타입을 명시적으로 지정하는 데 사용됩니다. 정적 분석 도구나 IDE에서 코드의 타입을 더 잘 이해하고, 타입 관련 경고나 오류를 감지하는 데 도움이 됩니다.

• 런타임에는 아무 영향 없음: cast는 런타임에 아무런 영향을 미치지 않습니다. 즉, cast를 사용해도 실제로 값의 타입이 변경되거나 변환되지 않습니다.

예시

1. 기본 사용법:

from typing import cast, List

def get_items() -> List[str]:
    return ["apple", "banana", "cherry"]

items = cast(List[str], get_items())
print(items)  # ['apple', 'banana', 'cherry']

여기서 cast(List[str], get_items())는 get_items()의 반환값이 List[str] 타입임을 명시적으로 지정합니다.

2. 변수 타입 힌트:

from typing import cast

value: int = cast(int, "42")  # 정적 분석 도구에게 value가 int임을 알려줌
print(value)  # "42", 실제로는 str 타입임

이 예시에서 cast(int, "42")는 "42"가 int 타입이라고 명시하지만, 실제로는 str 타입입니다. 이 코드에서 cast는 단지 정적 타입 힌팅을 위한 것이며, value의 타입이 실제로 int로 변환되지 않습니다.

3. 사용자 정의 클래스:

from typing import cast

class Animal:
    def speak(self):
        pass

class Dog(Animal):
    def speak(self):
        return "Woof!"

class Cat(Animal):
    def speak(self):
        return "Meow!"

def get_animal() -> Animal:
    return Dog()

animal = cast(Dog, get_animal())
print(animal.speak())  # "Woof!", 정적 분석 도구는 animal을 Dog로 간주함

여기서 cast(Dog, get_animal())는 get_animal()이 반환하는 객체가 Dog 타입임을 명시적으로 지정하여, animal 변수가 Dog로 간주되도록 합니다. 실제로는 get_animal()이 반환하는 객체가 Dog이기 때문에 문제가 없지만, 정적 분석 도구에 명시적으로 알려주기 위해 사용됩니다.

요약

• cast는 Python에서 타입 힌팅을 명시적으로 지정하기 위한 도구로, 런타임에는 영향을 미치지 않습니다.

• 주로 정적 분석 도구나 IDE에서 타입 추론을 돕기 위해 사용됩니다.

• cast는 값의 실제 타입을 변경하지 않으며, 코드의 가독성과 안전성을 높이는 역할을 합니다.

return cls(messages, template_format=template_format)는 Python에서 클래스 메소드나 다른 클래스 내부 메소드에서 새로운 클래스 인스턴스를 생성하여 반환할 때 사용되는 패턴입니다. 이 문장에서 cls는 현재 클래스 자체를 가리키고, messages와 template_format은 그 클래스의 생성자 __init__ 메소드에 전달되는 인수입니다.

이 코드의 의미

1. cls:
   • cls는 클래스 메소드 또는 클래스 내부의 다른 메소드에서 해당 클래스 자체를 참조하는 키워드입니다. cls를 사용하면 현재 클래스의 인스턴스를 생성할 수 있습니다.
   • 일반적으로 @classmethod 데코레이터가 붙은 메소드 내에서 사용되며, 이 메소드가 호출될 때 해당 클래스 자체가 첫 번째 인수로 cls에 전달됩니다.
2. cls(messages, template_format=template_format):
   • 이 구문은 현재 클래스의 인스턴스를 생성하는 표현입니다.
   • messages와 template_format=template_format은 생성자에 전달되는 인수입니다. 여기서 messages는 위치 인수로, template_format은 키워드 인수로 전달됩니다.
   • 이는 클래스의 __init__ 메소드가 messages와 template_format이라는 두 개의 인수를 받을 것으로 예상된다는 것을 의미합니다.
3. return:
   • 새로 생성된 클래스 인스턴스를 반환합니다. 이 반환된 객체는 이 메소드를 호출한 코드에서 사용할 수 있게 됩니다.

예시 코드

아래는 이 패턴이 어떻게 사용되는지에 대한 예시입니다:

class PromptTemplate:
    def __init__(self, messages, template_format=None):
        self.messages = messages
        self.template_format = template_format

    @classmethod
    def from_messages(cls, messages, template_format="default"):
        # 새로운 PromptTemplate 인스턴스를 생성하여 반환
        return cls(messages, template_format=template_format)

# 사용 예시
messages = ["Hello, World!", "How are you?"]
template = PromptTemplate.from_messages(messages, template_format="custom")
print(template.messages)  # ['Hello, World!', 'How are you?']
print(template.template_format)  # 'custom'

요약

• return cls(messages, template_format=template_format)는 현재 클래스의 인스턴스를 생성하여 반환하는 코드입니다.
• cls는 해당 클래스 자체를 참조하며, 이 코드 구문은 클래스 메소드 또는 클래스 내부의 다른 메소드에서 새로운 인스턴스를 생성하기 위해 사용됩니다.
• 이 패턴은 클래스의 __init__ 메소드가 해당 인수들을 받아들일 것으로 예상합니다. 따라서 이 코드가 사용되는 클래스는 __init__ 메소드가 messages와 template_format을 인수로 받아야 합니다.

Retriever은 VectorStore에서 데이터를 검색하는 인터페이스이다. Retriever는 VectorStore에서 as_retriever() 메서드 호출로 생성한다. 리트리버의 두가지 쿼리 방식에 대해 알아보자. 

  - Sematic Similarity

     - k: relevance 관련성 있는 것을 결과에 반영할 갯수

  - Maxium Marginal relavance

     - k: relevance 관련성 있는 것을 결과에 반영할 갯수

     - fetch_k: diverse 다양성 있는 것을 결과에 반영할 갯수

패키지

Semantic Similarity는 의미적으로 유사한 것을 찾아준다. Reriever를 VectorStore를 통해 얻게 되면 VectoreStore 클래스의 similarity_search() 를 사용한다. 그리고, VectoreStore 클래스를 상속받은 구현체는 similarity_search가 abstractmethod 이기에 구현을 해야 한다. 예로)  Chroma는 VectorStore를 상속받아 구현하고 있다.

  - 관련성있는 것을 찾기 위하여 k 갯수를 설정한다. 

class Chroma(VectorStore):
    def similarity_search(
        self,
        query: str,
        k: int = DEFAULT_K,
        filter: Optional[Dict[str, str]] = None,
        **kwargs: Any,
    ) -> List[Document]:
        """Run similarity search with Chroma.

        Args:
            query (str): Query text to search for.
            k (int): Number of results to return. Defaults to 4.
            filter (Optional[Dict[str, str]]): Filter by metadata. Defaults to None.

        Returns:
            List[Document]: List of documents most similar to the query text.
        """
        docs_and_scores = self.similarity_search_with_score(
            query, k, filter=filter, **kwargs
        )
        return [doc for doc, _ in docs_and_scores]

MMR(Maximum Marginal Relevance) 는 관련성과 다양성을 동시에 검색하여 제공한다. VectorStore의 max_marginal_relevance_search() 를 사용한다.  Chroma 구현내용을 보자.

   - k: 관련성 결과반영 갯수

   - fetch_k: 다양성 결과반영 갯수 

class Chroma(VectorStore):
    def max_marginal_relevance_search(
        self,
        query: str,
        k: int = DEFAULT_K,
        fetch_k: int = 20,
        lambda_mult: float = 0.5,
        filter: Optional[Dict[str, str]] = None,
        where_document: Optional[Dict[str, str]] = None,
        **kwargs: Any,
    ) -> List[Document]:
        """Return docs selected using the maximal marginal relevance.
        Maximal marginal relevance optimizes for similarity to query AND diversity
        among selected documents.

        Args:
            query: Text to look up documents similar to.
            k: Number of Documents to return. Defaults to 4.
            fetch_k: Number of Documents to fetch to pass to MMR algorithm.
            lambda_mult: Number between 0 and 1 that determines the degree
                        of diversity among the results with 0 corresponding
                        to maximum diversity and 1 to minimum diversity.
                        Defaults to 0.5.
            filter (Optional[Dict[str, str]]): Filter by metadata. Defaults to None.

        Returns:
            List of Documents selected by maximal marginal relevance.
        """
        if self._embedding_function is None:
            raise ValueError(
                "For MMR search, you must specify an embedding function on" "creation."
            )

        embedding = self._embedding_function.embed_query(query)
        docs = self.max_marginal_relevance_search_by_vector(
            embedding,
            k,
            fetch_k,
            lambda_mult=lambda_mult,
            filter=filter,
            where_document=where_document,
        )
        return docs

VectorStore에서 from_documents 또는 from_texts 메서도에 content와 embedding을 파라미터 설정하고 호출하면 VST 타입의 VectorStore 객체를 반환 받을 수 있다. 이를 통해 similarity_search와 max_marginal_relevance_search 를 사용할 수 있다. 

class VectorStore(ABC):
    @classmethod
    @abstractmethod
    def from_texts(
        cls: Type[VST],
        texts: List[str],
        embedding: Embeddings,
        metadatas: Optional[List[dict]] = None,
        **kwargs: Any,
    ) -> VST:
        """Return VectorStore initialized from texts and embeddings.

        Args:
            texts: Texts to add to the vectorstore.
            embedding: Embedding function to use.
            metadatas: Optional list of metadatas associated with the texts.
                Default is None.
            **kwargs: Additional keyword arguments.

        Returns:
            VectorStore: VectorStore initialized from texts and embeddings.
        """

similarity_search와 max_marginal_relevance_search 를 사용예를 보자.

from langchain.vectorstores import Chroma
from langchain.openai import OpenAIEmbeddings

persistence_path = 'db/chroma'
embeddings = OpenAIEmbeddings()
vectorestore = Chroma(persist_directory=persistence_path, embedding_function=embeddings)

texts = [
    """홍길동을 홍길동이라 부르지 못하는 이유는 무엇인가""",
    """심청이가 물에 빠진것을 효라고 말할 수 있는가, 그것이 진심이었나""",
    """춘향이가 이몽룡을 기다른 것은 진심이었나""",
]

smalldb = Chroma.from_texts(texts, embedding=embedding)
question = "진심이라는 의미"

smalldb.similarity_search(question, k=1)
// 결과 
[Document(page_content='춘향이가 이몽룡을 기다른 것은 진심이었나', metadata={})]


smalldb.max_marginal_relevance_search(question,k=2, fetch_k=3)
// 결과
[Document(page_content='춘향이가 이몽룡을 기다른 것은 진심이었나', metadata={}),
 Document(page_content='심청이가 물에 빠진것을 효라고 말할 수 있는가, 그것이 진심이었나', metadata={})]

위의 두가지외에 다른 검색 타입까지 합쳐 놓은 것이 langchain_core VectorStore의 as_retriever() 이다. from_documents 또는 from_texts를 호출하면 VST (Vector Store Type) 인스턴스를 반환 받아 as_retriever()를 호출한다. as_retriever()는 langchain_community VectorStoreRetriever를 반환하고, 이는 langchain_core의 retrievers.py 파일의 BaseRetriever 클래스를 상속받아 구현하고 있다. 

  - 0.1.46 이후 get_relevant_documents 사용하지 않고 invoke로 대체한다. 

  - _get_relevant_documents 를 구현해야 한다.

  - search_type 

     - similarity

     - similarity_score_threshold : score_threshold 값 설정 이상만 결과에 반영

     - mmr 

  - 결과값: List[Document] 

class VectorStoreRetriever(BaseRetriever):
    """Base Retriever class for VectorStore."""

    vectorstore: VectorStore
    """VectorStore to use for retrieval."""
    search_type: str = "similarity"
    """Type of search to perform. Defaults to "similarity"."""
    search_kwargs: dict = Field(default_factory=dict)
    """Keyword arguments to pass to the search function."""
    allowed_search_types: ClassVar[Collection[str]] = (
        "similarity",
        "similarity_score_threshold",
        "mmr",
    )
    ...
    
 class BaseRetriever(RunnableSerializable[RetrieverInput, RetrieverOutput], ABC):
    def invoke(
        self, input: str, config: Optional[RunnableConfig] = None, **kwargs: Any
    ) -> List[Document]:
        """Invoke the retriever to get relevant documents.
        ...
        """
        ...
        
    
    // 하위 클래스 생성시 호출이 된다. cls는 하위 클래스이다. get_relevant_documents를 할당함.
    def __init_subclass__(cls, **kwargs: Any) -> None:
        super().__init_subclass__(**kwargs)
        # Version upgrade for old retrievers that implemented the public
        # methods directly.
        if cls.get_relevant_documents != BaseRetriever.get_relevant_documents:
            warnings.warn(
                "Retrievers must implement abstract `_get_relevant_documents` method"
                " instead of `get_relevant_documents`",
                DeprecationWarning,
            )
            swap = cls.get_relevant_documents
            cls.get_relevant_documents = (  # type: ignore[assignment]
                BaseRetriever.get_relevant_documents
            )
            cls._get_relevant_documents = swap  # type: ignore[assignment]
        if (
            hasattr(cls, "aget_relevant_documents")
            and cls.aget_relevant_documents != BaseRetriever.aget_relevant_documents
        ):
            warnings.warn(
                "Retrievers must implement abstract `_aget_relevant_documents` method"
                " instead of `aget_relevant_documents`",
                DeprecationWarning,
            )
            aswap = cls.aget_relevant_documents
            cls.aget_relevant_documents = (  # type: ignore[assignment]
                BaseRetriever.aget_relevant_documents
            )
            cls._aget_relevant_documents = aswap  # type: ignore[assignment]
        parameters = signature(cls._get_relevant_documents).parameters
        cls._new_arg_supported = parameters.get("run_manager") is not None
        # If a V1 retriever broke the interface and expects additional arguments
        cls._expects_other_args = (
            len(set(parameters.keys()) - {"self", "query", "run_manager"}) > 0
        )
    
    @abstractmethod
    def _get_relevant_documents(
        self, query: str, *, run_manager: CallbackManagerForRetrieverRun
    ) -> List[Document]:
        """Get documents relevant to a query.

        Args:
            query: String to find relevant documents for.
            run_manager: The callback handler to use.
        Returns:
            List of relevant documents.
        """
        ...
        
    @deprecated(since="0.1.46", alternative="invoke", removal="0.3.0")
    def get_relevant_documents(
       ....

API

langchain_core의 BaseRetriever --> langchain_core의 VectorStoreRetriever 또는  langchain_community에서 다양한 <name>Retriever 를 구현한다. 

  - invoke, ainvoke 호출

  - get_relevant_documents 호출은 invoke 로 대체되고, v0.3.0 에서 삭제될 예정이다. 

<참조>

- 공식문서: https://python.langchain.com/v0.2/docs/integrations/retrievers/

- API: https://api.python.langchain.com/en/latest/community_api_reference.html#module-langchain_community.retrievers

- LangChain KR: https://wikidocs.net/234016

import와 from 키워드는 파이썬에서 모듈과 그 모듈 내의 특정 항목을 가져오는 데 사용됩니다. 각각의 기능과 사용할 수 있는 것들을 정리하면 다음과 같습니다:

import 키워드

import는 전체 모듈을 가져옵니다. 가져온 모듈의 이름을 통해 해당 모듈에 정의된 함수, 클래스, 변수 등을 사용할 수 있습니다.

import module_name

import로 가져올 수 있는 것들:

• 파이썬 표준 라이브러리 모듈: 예를 들어, import os, import sys, import math 등.
• 사용자 정의 모듈: 사용자가 만든 .py 파일을 가져올 수 있습니다. 예를 들어, import my_module.
• 서드 파티 라이브러리 모듈: 설치된 서드 파티 패키지를 가져올 수 있습니다. 예를 들어, import numpy, import pandas.

from 키워드

from은 모듈 내의 특정 항목을 직접 가져올 때 사용됩니다. 이를 통해 모듈 이름 없이도 해당 항목을 직접 사용할 수 있습니다.

from module_name import specific_item

from으로 정의할 수 있는 것들:

• 모듈 내의 특정 함수: 예를 들어, from math import sqrt는 sqrt 함수를 직접 사용할 수 있게 합니다.
• 모듈 내의 특정 클래스: 예를 들어, from datetime import datetime은 datetime 클래스를 직접 사용할 수 있게 합니다.
• 모듈 내의 특정 변수: 예를 들어, from config import config_value는 config_value라는 변수를 가져올 수 있게 합니다.
• 모듈 내의 모든 항목: from module_name import *를 사용하면 모듈 내의 모든 공개된 항목을 가져올 수 있습니다. 하지만, 이 방식은 권장되지 않습니다.

예시

import math  # math 모듈을 가져옴
print(math.sqrt(16))  # math 모듈을 통해 sqrt 함수 호출

from math import sqrt  # math 모듈에서 sqrt 함수만 가져옴
print(sqrt(16))  # 모듈 이름 없이 sqrt 함수 호출

from my_module import my_function  # my_module 모듈에서 my_function 함수만 가져옴
my_function()  # 직접 함수 호출

import와 from 키워드를 적절히 사용하면 코드의 가독성과 효율성을 높일 수 있습니다. written by GPT

클래스의 속성을 정의하는 여러 가지 방법을 소개해 드리겠습니다. 파이썬에서 클래스의 속성을 정의하는 방법에는 몇 가지가 있습니다. 각 방법은 특정 상황에서 유용할 수 있습니다.

클래스 변수와 속성의미

1. 클래스 변수 (Class Variables)

클래스 변수는 클래스 자체에 속하며, 모든 인스턴스에서 공유됩니다.

   class MyClass:
       class_variable = "I am a class variable"

   print(MyClass.class_variable)
   obj1 = MyClass()
   obj2 = MyClass()
   print(obj1.class_variable)
   print(obj2.class_variable)

2. 인스턴스 변수 (Instance Variables)

인스턴스 변수는 각 객체마다 개별적으로 유지되며, __init__ 메서드에서 초기화됩니다.

   class MyClass:
       def __init__(self, value):
           self.instance_variable = value

   obj1 = MyClass(1)
   obj2 = MyClass(2)
   print(obj1.instance_variable)  # 1
   print(obj2.instance_variable)  # 2

3. 프로퍼티 (Properties)

프로퍼티는 getter와 setter 메서드를 사용하여 속성의 접근 및 수정 방법을 제어할 수 있습니다.

   class MyClass:
       def __init__(self, value):
           self._value = value

       @property
       def value(self):
           return self._value

       @value.setter
       def value(self, new_value):
           if new_value < 0:
               raise ValueError("Value cannot be negative")
           self._value = new_value

   obj = MyClass(10)
   print(obj.value)  # 10
   obj.value = 20
   print(obj.value)  # 20
   # obj.value = -10  # ValueError 발생

4. 데코레이터를 사용한 클래스 속성 정의

@classmethod 및 @staticmethod 데코레이터를 사용하여 클래스 메서드 및 정적 메서드를 정의할 수 있습니다.

   class MyClass:
       class_variable = 0

       @classmethod
       def increment_class_variable(cls):
           cls.class_variable += 1

       @staticmethod
       def static_method():
           print("This is a static method")

   MyClass.increment_class_variable()
   print(MyClass.class_variable)  # 1
   MyClass.static_method()  # This is a static method

5. 데이터 클래스 (Data Classes)

파이썬 3.7부터 도입된 데이터 클래스는 기본적인 속성 및 메서드를 자동으로 생성해줍니다.

   from dataclasses import dataclass

   @dataclass
   class MyClass:
       attribute1: int
       attribute2: str

   obj = MyClass(1, "hello")
   print(obj.attribute1)  # 1
   print(obj.attribute2)  # hello

이와 같은 방법들을 통해 파이썬 클래스의 속성을 정의하고 사용할 수 있습니다. 각 방법은 특정 상황에서 유용하며, 필요에 따라 적절한 방법을 선택하여 사용할 수 있습니다.

인스턴스 변수 와 속성의 차이점

클래스의 인스턴스 변수와 속성은 비슷하지만, 약간의 차이점이 있습니다. 인스턴스 변수는 클래스의 각 인스턴스에서 개별적으로 유지되는 데이터를 의미하며, 속성은 주로 접근 제어 및 추가적인 로직을 포함하는 특수한 메서드를 통해 접근되는 데이터를 의미합니다. 여기서 두 개념의 차이점을 더 자세히 설명하겠습니다.

인스턴스 변수 (Instance Variables)

인스턴스 변수는 클래스의 각 인스턴스마다 독립적으로 존재하는 변수를 의미합니다. 이는 보통 __init__ 메서드 내에서 정의되며, 직접적으로 접근하고 수정할 수 있습니다.

예시:

class MyClass:
    def __init__(self, value):
        self.instance_variable = value  # 인스턴스 변수

obj = MyClass(10)
print(obj.instance_variable)  # 10
obj.instance_variable = 20
print(obj.instance_variable)  # 20

속성 (Properties)

속성은 메서드를 통해 간접적으로 접근하고 수정할 수 있는 인스턴스 변수와 비슷한 개념입니다. 속성은 @property 데코레이터를 사용하여 정의되며, 이로 인해 getter, setter, deleter 메서드를 통해 접근과 수정이 가능합니다. 속성은 접근 제어, 유효성 검사, 추가 로직을 추가하는 데 유용합니다.

예시:

class MyClass:
    def __init__(self, value):
        self._value = value  # 실제 데이터를 저장하는 변수 (보통 이름 앞에 _를 붙임)

    @property
    def value(self):
        return self._value

    @value.setter
    def value(self, new_value):
        if new_value < 0:
            raise ValueError("Value cannot be negative")
        self._value = new_value

obj = MyClass(10)
print(obj.value)  # 10
obj.value = 20
print(obj.value)  # 20
# obj.value = -10  # ValueError 발생

주요 차이점

1. 직접 접근 vs 간접 접근:
   • 인스턴스 변수는 직접 접근하고 수정할 수 있습니다 (obj.instance_variable).
   • 속성은 간접적으로 접근하고 수정할 수 있으며, 접근 메서드를 통해 제어됩니다 (obj.value).
2. 추가 로직:
   • 인스턴스 변수는 단순히 데이터를 저장합니다.
   • 속성은 getter, setter 메서드를 통해 접근 로직을 추가할 수 있습니다 (예: 유효성 검사, 로그 작성).
3. 읽기 전용 속성:
   • 인스턴스 변수는 읽기/쓰기가 자유롭습니다.
   • 속성은 @property만 정의하면 읽기 전용으로 만들 수 있습니다. (setter를 정의하지 않음)

결론

인스턴스 변수는 클래스 인스턴스의 데이터를 저장하는 데 사용되며, 속성은 데이터를 보호하고 접근을 제어하기 위해 사용됩니다. 속성은 특히 추가적인 로직이 필요한 경우 유용합니다.

1) 클래스내에서 사용하는 경우

__getattr__ 메서드는 Python에서 객체의 속성에 접근할 때 호출되는 특별 메서드입니다. 객체의 속성을 조회할 때 해당 속성이 존재하지 않으면 __getattr__ 메서드가 호출됩니다. 이는 동적 속성 접근을 가능하게 하며, 클래스에서 존재하지 않는 속성에 대한 요청을 처리할 수 있습니다.

__getattr__의 의미와 사용법

• 의미: __getattr__은 객체에서 속성에 접근할 때 해당 속성이 존재하지 않는 경우 호출됩니다. 이는 객체의 속성을 동적으로 생성하거나 계산된 속성을 반환하는 데 유용합니다.
• 사용법: __getattr__ 메서드는 클래스 내부에 정의되며, 단 하나의 매개변수를 가집니다. 이 매개변수는 존재하지 않는 속성의 이름을 나타내는 문자열입니다. 이 메서드는 존재하지 않는 속성에 대해 반환할 값을 반환해야 합니다.

예시 코드

아래는 __getattr__ 메서드를 사용하여 동적으로 속성을 생성하는 예시입니다:

from typing import Any

class DynamicAttributes:
    def __init__(self):
        self.existing_attribute = "This attribute exists"

    def __getattr__(self, name: str) -> Any:
        """
        This method is called when an attribute is not found.

        Args:
            name (str): The name of the attribute that is being accessed.

        Returns:
            Any: The value to return for the non-existing attribute.
        """
        return f"The attribute '{name}' does not exist, but here is a default value."

# Example usage:
obj = DynamicAttributes()
print(obj.existing_attribute)  # Output: This attribute exists
print(obj.non_existing_attribute)  # Output: The attribute 'non_existing_attribute' does not exist, but here is a default value.

주요 포인트

1. 속성이 존재하는 경우: __getattr__는 호출되지 않습니다. 예를 들어, obj.existing_attribute에 접근하면 __getattr__가 호출되지 않고, existing_attribute의 실제 값이 반환됩니다.
2. 속성이 존재하지 않는 경우: __getattr__가 호출됩니다. 예를 들어, obj.non_existing_attribute에 접근하면, __getattr__가 호출되어 해당 속성의 이름을 인수로 받아 반환값을 제공합니다.
3. 반환값: __getattr__ 메서드는 호출된 속성에 대해 반환할 값을 반환해야 합니다. 이 값은 문자열, 숫자, 객체 등 어떤 유형도 될 수 있습니다.

요약

__getattr__ 메서드는 Python 객체에서 존재하지 않는 속성에 대한 접근을 처리하기 위한 메서드입니다. 이 메서드를 사용하면 동적 속성 접근, 계산된 속성 반환, 기본값 제공 등의 작업을 수행할 수 있습니다. 이를 통해 객체의 유연성을 높이고, 동적으로 속성을 관리할 수 있습니다.

2) init.py 파일 내에서 사용

__init__.py 파일에 __getattr__ 메서드를 정의하는 것은 패키지 수준에서의 동적 속성 접근을 가능하게 합니다. 이 메서드를 사용하면 모듈 또는 패키지에서 존재하지 않는 속성에 접근할 때 동적 동작을 정의할 수 있습니다.

패키지 수준에서의 __getattr__ 사용 예시

Python 3.7부터 패키지의 __init__.py 파일에 __getattr__를 정의할 수 있습니다. 이를 통해 패키지에서 존재하지 않는 속성에 접근할 때 원하는 동작을 수행할 수 있습니다.

예시: 동적 속성 접근

다음은 mypackage라는 패키지에서 __getattr__를 정의하는 예시입니다:

mypackage/
    __init__.py
    module1.py
    module2.py

__init__.py 파일에서 __getattr__를 정의합니다:

# mypackage/__init__.py

def __getattr__(name):
    if name == "special_attribute":
        return "This is a special attribute"
    raise AttributeError(f"module {__name__} has no attribute {name}")

이제 mypackage 패키지에서 special_attribute에 접근할 수 있습니다:

import mypackage

print(mypackage.special_attribute)  # Output: This is a special attribute
print(mypackage.non_existing_attribute)  # Raises AttributeError

주요 포인트

1. 동적 속성 접근: 패키지 수준에서 존재하지 않는 속성에 접근할 때 동적 동작을 정의할 수 있습니다. 예를 들어, 특정 속성 이름에 대해 동적으로 값을 반환하거나, 필요한 경우 예외를 발생시킬 수 있습니다.
2. 패키지 초기화: 패키지를 초기화할 때 __getattr__를 사용하면, 패키지에 포함되지 않은 속성에 대한 접근을 처리할 수 있습니다. 이는 패키지의 유연성을 높이고, 사용자에게 특정 속성 접근을 허용할 수 있습니다.
3. Python 3.7 이상: 패키지의 __init__.py 파일에 __getattr__를 정의하는 기능은 Python 3.7에서 도입되었습니다. 따라서 이 기능을 사용하려면 Python 3.7 이상 버전이 필요합니다.

예제 코드 설명

• __getattr__ 메서드는 name 매개변수를 받아, 접근하려는 속성의 이름을 나타냅니다.
• special_attribute에 접근할 때는 "This is a special attribute"를 반환합니다.
• special_attribute가 아닌 다른 속성에 접근하려고 하면 AttributeError를 발생시킵니다.

요약

• __init__.py 파일에 __getattr__를 정의하면 패키지 수준에서 동적 속성 접근을 처리할 수 있습니다.
• 이는 패키지의 유연성을 높이고, 사용자에게 특정 속성에 대한 동적 접근을 허용하는 데 유용합니다.
• Python 3.7 이상에서만 사용 가능합니다.

Google 스타일 docstring의 전체 포맷을 설명하기 위해, 모든 주요 섹션을 포함한 예시를 제공합니다. 이 예시는 함수와 클래스에 대한 포맷을 모두 다룹니다.

함수의 예시

def embed_query(text, model='default', verbose=False):
    """
    Embed query text using a specified model.

    This function takes a text string and converts it into an embedding
    using the specified model. It supports different models for embedding
    and can provide verbose output.

    Args:
        text (str): The text to embed. This should be a plain string without any special formatting.
        model (str, optional): The model to use for embedding. Defaults to 'default'.
        verbose (bool, optional): If True, the function will print detailed information during execution. Defaults to False.

    Returns:
        list: A list representing the text embedding.

    Raises:
        ValueError: If the text is empty or the model is not supported.
        RuntimeError: If the embedding process fails.

    Examples:
        >>> embed_query("Hello, world!")
        [0.1, 0.3, 0.5, ...]
        >>> embed_query("Hello, world!", model='advanced')
        [0.2, 0.4, 0.6, ...]

    Notes:
        The embedding process can take a significant amount of time
        depending on the length of the text and the complexity of the model.
    """
    if not text:
        raise ValueError("Text cannot be empty.")
    if model not in ['default', 'advanced']:
        raise ValueError("Unsupported model.")
    # Implementation of embedding process...
    embedding = [0.1, 0.3, 0.5]  # Dummy embedding
    if verbose:
        print(f"Embedding for '{text}' generated using model '{model}'.")
    return embedding

클래스의 예시

class EmbeddingModel:
    """
    A model for generating embeddings from text.

    This class provides methods to embed text using various models. It can
    handle different types of text inputs and supports multiple embedding
    techniques.

    Attributes:
        model_name (str): The name of the model.
        version (str): The version of the model.
        is_trained (bool): Indicates whether the model has been trained.

    Methods:
        train(data):
            Trains the model using the provided data.
        embed(text):
            Embeds the given text and returns the embedding.
        save(path):
            Saves the model to the specified path.
    """

    def __init__(self, model_name, version):
        """
        Initializes the EmbeddingModel with a name and version.

        Args:
            model_name (str): The name of the model.
            version (str): The version of the model.
        """
        self.model_name = model_name
        self.version = version
        self.is_trained = False

    def train(self, data):
        """
        Trains the model using the provided data.

        This method takes a dataset and trains the embedding model.
        It updates the is_trained attribute to True upon successful training.

        Args:
            data (list): A list of training data samples.

        Returns:
            None

        Raises:
            ValueError: If the data is empty or not in the expected format.

        Examples:
            >>> model = EmbeddingModel('text_model', '1.0')
            >>> model.train(['sample1', 'sample2'])
        """
        if not data:
            raise ValueError("Training data cannot be empty.")
        # Implementation of training process...
        self.is_trained = True

    def embed(self, text):
        """
        Embeds the given text and returns the embedding.

        Args:
            text (str): The text to embed.

        Returns:
            list: A list representing the text embedding.

        Raises:
            RuntimeError: If the model has not been trained.

        Examples:
            >>> model = EmbeddingModel('text_model', '1.0')
            >>> model.train(['sample1', 'sample2'])
            >>> model.embed('Hello, world!')
            [0.1, 0.3, 0.5, ...]
        """
        if not self.is_trained:
            raise RuntimeError("Model must be trained before embedding.")
        # Implementation of embedding process...
        return [0.1, 0.3, 0.5]  # Dummy embedding

    def save(self, path):
        """
        Saves the model to the specified path.

        Args:
            path (str): The file path to save the model to.

        Returns:
            None

        Examples:
            >>> model = EmbeddingModel('text_model', '1.0')
            >>> model.save('/path/to/save/model')
        """
        # Implementation of save process...
        pass

주요 구성 요소

1. 요약 설명(Summary):
   • 클래스나 함수의 첫 번째 줄에서 기능을 간략하게 설명합니다.
   • 간결하고 명확하게 작성하며, 마침표로 끝냅니다.
2. 확장 설명(Extended Description):
   • 요약 설명 이후 빈 줄을 두고, 기능에 대한 상세 설명을 작성합니다.
   • 설명이 길어질 경우 여러 문단으로 나눌 수 있습니다.
3. Args (인수):
   • Args: 섹션에서 함수나 메소드의 매개변수를 설명합니다.
   • 각 매개변수에 대해 이름, 유형, 설명을 포함합니다.
   • 선택적 매개변수는 (optional)로 표시합니다.
4. Attributes (속성):
   • 클래스의 속성을 설명합니다.
   • 각 속성의 이름과 설명을 포함합니다.
5. Methods (메소드):
   • 클래스의 메소드를 설명합니다.
   • 각 메소드의 이름과 기능을 간략하게 설명합니다.
6. Returns (반환값):
   • 함수나 메소드의 반환값을 설명합니다.
   • 반환값의 유형과 설명을 포함합니다.
   • 반환값이 없으면 생략할 수 있습니다.
7. Raises (예외):
   • 함수나 메소드가 발생시킬 수 있는 예외를 설명합니다.
   • 예외의 유형과 설명을 포함합니다.
8. Examples (예제):
   • 함수나 메소드의 사용 예제를 포함합니다.
   • 코드 블록을 사용하여 예제를 보여줍니다.
9. Notes (참고):
   • 함수나 메소드에 대한 추가적인 참고 사항을 작성합니다.
   • 필요한 경우에만 포함합니다.

이러한 구성 요소들을 사용하면, Google 스타일의 docstring을 통해 코드의 문서화를 일관성 있게 작성할 수 있습니다.

Python에서 lambda 키워드는 익명 함수(anonymous function)를 생성하는 데 사용됩니다. lambda 함수를 사용하면 이름 없이도 함수 객체를 생성할 수 있습니다. 일반적으로 lambda 함수는 간단한 기능을 수행하는 짧은 함수가 필요할 때 사용됩니다.

lambda 함수의 구문

lambda 키워드를 사용하여 함수를 정의하는 구문은 다음과 같습니다:

lambda arguments: expression

• arguments: 함수에 전달될 인수들입니다.
• expression: 함수가 반환할 표현식입니다.

예시

기본 사용 예시

# 일반 함수 정의 
def add(x, y): 
  return x + y 
  
# lambda 함수 정의 
add_lambda = lambda x, y: x + y 

# 함수 호출 
print(add(2, 3)) # 5 
print(add_lambda(2, 3)) # 5

리스트의 각 요소에 함수를 적용하는 예시

# lambda 함수를 사용하여 리스트의 각 요소에 2를 곱함 
numbers = [1, 2, 3, 4, 5] 
doubled = list(map(lambda x: x * 2, numbers)) 
print(doubled) # [2, 4, 6, 8, 10]

정렬 시에 키로 사용하는 예시

# lambda 함수를 사용하여 리스트를 정렬 
pairs = [(1, 'one'), (2, 'two'), (3, 'three'), (4, 'four')] 
# 두 번째 요소(문자열) 기준으로 정렬 
pairs.sort(key=lambda pair: pair[1]) 
print(pairs) # [(4, 'four'), (1, 'one'), (3, 'three'), (2, 'two')]

lambda 함수와 일반 함수 비교

lambda 함수는 익명 함수로, 보통 한 줄로 표현되며, 작은 연산이나 함수 객체가 필요한 곳에서 주로 사용됩니다. 일반 함수는 def 키워드를 사용하여 이름이 있는 함수로 정의되며, 여러 줄로 이루어질 수 있고 복잡한 논리를 포함할 수 있습니다.

예시: lambda 함수와 일반 함수의 비교

# lambda 함수 
multiply = lambda x, y: x * y 
print(multiply(2, 3)) # 6 

# 일반 함수 
def multiply_def(x, y): 
  return x * y 
  
print(multiply_def(2, 3)) # 6

사용 시 주의사항

• 간결성: lambda 함수는 단일 표현식만 포함할 수 있으며, 여러 문장을 포함할 수 없습니다. 따라서 복잡한 로직을 처리하기에는 적합하지 않습니다.
• 가독성: 짧고 간단한 경우에는 lambda 함수가 유용하지만, 너무 복잡한 경우 가독성을 해칠 수 있으므로 일반 함수를 사용하는 것이 좋습니다.

요약

• lambda 키워드는 익명 함수를 생성하는 데 사용됩니다.
• 간단한 함수나 일회성 함수 객체가 필요한 곳에서 유용합니다.
• 구문은 lambda arguments: expression 형태를 가지며, 단일 표현식만 포함할 수 있습니다.
• 일반 함수(def 사용)와 비교할 때, 더 간단하고 간결하지만 복잡한 로직에는 적합하지 않습니다.

이를 통해 lambda 키워드의 역할과 사용 방법을 이해할 수 있습니다.

Mac 기준으로

Python Version Manager 설치

pyenv 설치

brew install pyenv

.zshrc 설정 추가

# pyenv setting
eval "$(pyenv init --path)"
eval "$(pyenv init -)"

Python Virtual Environment 설치

pyenv-virtualenv 설치

brew install pyenv-virtualenv

.zshrc 설정 추가

# pyenv-virtualenv setting
eval "$(pyenv virtualenv-init -)"

Python Version 설치

python 버전 목록 확인

pyenv install --list

python 특정 버전 설치

pyenv install 3.12.4

현재 사용버전

pyenv versions

* system (set by /Users/dowon2yun/.pyenv/version)
  2.7.14
  3.6.2
  3.6.2/envs/mms
  3.12.4
  mms --> /Users/dowon2yun/.pyenv/versions/3.6.2/envs/mms

Python Version 적용 

전체 사용 설정

  - pyenv global [version]

pyenv global 3.12.4
pyenv versions
  system
  2.7.14
  3.6.2
  3.6.2/envs/mms
* 3.12.4 (set by /Users/dowon2yun/.pyenv/version)

로컬 프로젝트만 사용 설정

  - pyenv local [version]

~ mkdir test
~ cd test
~/test pyenv local 3.6.2
~/test ls -alrt
total 8
drwxr-xr-x    3 dowon2yun  staff    96 Jun 17 23:00 .
-rw-r--r--    1 dowon2yun  staff     6 Jun 17 23:00 .python-version
drwxr-xr-x+ 128 dowon2yun  staff  4096 Jun 17 23:00 ..
~/test cat .python-version
3.6.2

현재 shell만 사용 설정

  - pyenv shell [version]

~/test pyenv shell 3.12.4
~/test pyenv version
3.12.4 (set by PYENV_VERSION environment variable)

Python Virtual Env 설정

python 패키지의 버전 격리 환경을 만들어 준다. 

python 버전과 virtual 환경 생성

  - pyenv virtualenv [version] [virtual-name]

~ mkdir test2
~ cd test2
~/test2 pyenv virtualenv 3.12.4 venv
~/test2 pyenv versions
  system
  2.7.14
  3.6.2
  3.6.2/envs/mms
* 3.12.4 (set by PYENV_VERSION environment variable)
  3.12.4/envs/venv
  mms --> /Users/dowon2yun/.pyenv/versions/3.6.2/envs/mms
  venv --> /Users/dowon2yun/.pyenv/versions/3.12.4/envs/venv

Python Virtual Env 활성화/비활성화

  - pyenv activate [virtual-name]

~/test2 pyenv activate venv
(venv) ~/test2 pyenv versions
  system
  2.7.14
  3.6.2
  3.6.2/envs/mms
  3.12.4
  3.12.4/envs/venv
  mms --> /Users/dowon2yun/.pyenv/versions/3.6.2/envs/mms
* venv --> /Users/dowon2yun/.pyenv/versions/3.12.4/envs/venv (set by PYENV_VERSION environment variable)

  - pyenv deactivate [virtual-name]

(venv) ~/test2 pyenv deactivate venv
~/test2 pyenv versions
  system
  2.7.14
  3.6.2
  3.6.2/envs/mms
* 3.12.4 (set by /Users/dowon2yun/.pyenv/version)
  3.12.4/envs/venv
  mms --> /Users/dowon2yun/.pyenv/versions/3.6.2/envs/mms
  venv --> /Users/dowon2yun/.pyenv/versions/3.12.4/envs/venv
~/test2

Poetry 설치

전문적인 의존성 관리 및 패키지 배포관리 툴인 poetry를 설치한다. 

https://python-poetry.org/docs/

pipx 를 통해 설치힌다. 

https://pipx.pypa.io/stable/installation/

brew install pipx
pipx ensurepath
sudo pipx ensurepath --global

pipx를 통해 poetry를 설치한다. python 3.12.3 을 사용한다는 메세지가 출력된다. 

pipx install poetry

// result message
  installed package poetry 1.8.3, installed using Python 3.12.3
  These apps are now globally available
    - poetry

poetry를 실행한다. 

poetry

// result mesage
Poetry (version 1.8.3)

Poetry 다음 Tab으로 명령 목록 보기

oh-my-zsh 설정이 .zshrc 에 있음을 가정한다

// .zshrc 에서 ZSH_CUSTOM 주석 풀고 계정 폴더 밑으로 oh-my-zsh 설정
# Would you like to use another custom folder than $ZSH/custom?
ZSH_CUSTOM=/Users/peter/oh-my-zsh

// 저장후 변경 적용
. .zshrc

// 폴더 생성 
mkdir $ZSH_CUSTOM/plugins/poetry

oh-my-zsh 의 plugins 에 poetry 추가 

// .zshrc oh-my-zsh의 plugins 에 poetry 추가
plugins=(git poetry)

// .zshrc 변경 적용
. .zshrc

다음 명령 수행

poetry completions zsh > $ZSH_CUSTOM/plugins/poetry/_poetry

테스트 "peotry in" 까지 입력하고 tab key를 치면 아래와 같이 init, install 등의 poetry 명령 목록이 출력된다.

$ . .zshrc
$ poetry in
init     -- Creates a basic pyproject.toml file in the current directory.
install  -- Installs the project dependencies.

Poetry 통한  프로젝트, 패키지 추가

[1] Poetry 기반 프로젝트 생성

  - poetry new [project-name]

poetry new ai-agent

[2] 프로젝트로 이동해서 가상환경을 프로제트내로 설정한다.

poetry config virtualenvs.in-project true

[3] poetry 프로젝트 가상환경으로 변경

poetry shell

[4] ai_agent 패키지 폴더에 __main__.py 추가

  -  폴더 지정으로 run 하기 위해 __init__.py 위치에 __main__.py 파일을 추가한다.

  - .env 파일 생성후 KEY 값 설정

  -  명령

     - poetry shell 미수행시 : poetry run python [folder-name] 또는 [file-name]

     - poetry shell 수행시 : python [folder-name]

// .env 파일
OPENAI_API_KEY=sh-proj-xsdhfdrerjelrelreahahhahahaahaha

// __main__.py 내역 
import os
from dotenv import load_dotenv

load_dotenv()

print(f"[API KEY]\n{os.environ['OPENAI_API_KEY']}")

// 실행
poetry run python ai_agent
[API KEY]
sh-proj-xsdhfdrerjelrelreahahhahahaahaha

[5] Python version 변경

1) pyenv install [newVersion]

2) poetry env remove [oldVersion]

3) poetry env use [newVersion]

References

https://python-poetry.org/docs/

https://blog.flynnpark.dev/15

https://github.com/ohmyzsh/ohmyzsh/wiki/Plugins

구성파일은 TOML 파일 형식

https://www.itworld.co.kr/news/248128#:~:text=TOML(Tom's%20Obvious%20Minimal%20Language)%EC%9D%80%20%EA%B5%AC%EC%84%B1%20%EB%8D%B0%EC%9D%B4%ED%84%B0%20%EC%A0%80%EC%9E%A5%EC%9D%84,%EC%95%84%EB%8B%88%EB%8B%A4(pip%EC%97%90%EC%84%9C%20%ED%8C%A8%ED%82%A4%EC%A7%80%20%EB%B9%8C%EB%93%9C%EB%A5%BC%20%EC%9C%84%ED%95%B4%20%EC%82%AC%EC%9A%A9%EB%90%98%EB%8A%94%20pyproject.toml%20%EC%B0%B8%EC%A1%B0).

https://teddylee777.github.io/poetry/poetry-tutorial/

NMF(non-negative matrix factorization) 기법을 소개한다. 이 기법은 데이터내에 독립된 특성을 찾는데 사용한다. 많은 데이터 세트 내의 항목들은 미리 알기 어려운 다른 특성들과의 조합으로 생성되고 이러한 특성을 찾아 본다. 

이전 분류 방식

  - 베이지안 분류기 (Classification)

  - 의사결정트리

  - 지지벡터머신(SVM)

  - 군집 (Clustering)

비음수 행렬 인수분해 (NMF)

  - 행렬 준비 

    + 특성 행렬(feature matrix) : 가로줄에 각각의 특성을 가졌고 세로줄에 단어, 값들은 단어 특성의 중요도

    + 가중치 행렬(weights matrix) : 가로줄 타이틀, 세로줄 특성

    + 행렬 만들기 = 특성 행렬 * 가중치 행렬

  - 비음수 행렬 인수분해라 불리는 이유는 음수가 아닌 특성과 가중치를 리턴하기 때문이다. 모든 특성은 양수 값을 가져야 함을 의미한다.

  - 단어 출현 횟수와 같은 명사류 데이터뿐만 아니라 주식시장 데이터와 같은 숫자 데이터 문제에도 잘 어울린다.

  - 진행 : 행렬 준비 -> NMF 실행 -> 결과 출력    

     * 특성을 나누는 행렬을 만들고 필요없는 특성을 제거하면서 원하는 차원을 줄여서 원하는 것을 종합적인 결과를 출력함

<참조> 

 - 얼굴 인식

 - PCA, NMF

<참조>

- 파이썬 가상환경 만들기

  비오는 토요일 오전 강남 토즈타워에서 다섯분과 함께 머신 러닝 스터디를 시작한다. 

  모임 : https://www.facebook.com/groups/1511952519075217/  

  참여자 : 봉성주님, 서병선님, 김민기님, 이웅재님과 함께 한다. 

  기타 : 총 12주동안 http://www.it-ebooks.info/book/330/ 책을 1 챕터식 읽고 질문하고 답하기. 

  책의 서문에 나온 내용이다. 인터넷을 기반으로 하는 집단 지성 데이터를 수집하여 다양한 분야를 들여다 보는데 책의 목적이 있는 것 같다. 

1장 

  실제 생활에서 집단 지성은 어디에서 쓰는 것일까 예를 들어준다. 

  - 시장 예측

  - 금융 사기 탐지

  - 머신 비젼

  - 공급망 최적화

  - 주식 마켓 분석

  - 국가 안보 

  머신러닝

  - 클렌징이 중요하다. 

    + 클렌징은 누가 하는가? 데이터 마이닝

  - 이미 되어 있다고 가정하고 머신러닝을 수행한다. 

회고 

  하고 싶었던 것을 함께 할 수 있어서 좋다. 책이 나의 목적과 너무 잘 맞는다. 피부에 와 닿는 것을 경험할 수 있을 것 같다. 나에겐 신선한다. 

커뮤니케이션 방식 

  - slack 

  - github 저장소 : https://github.com/ML-Lounge/Collective-Intelligence