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

RAG 구성시에 마지막 chain을 만들 때 retriever를 설정할 때 RunnableParallel을 사용한다. 

from langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

vectorstore = FAISS.from_texts(
    ["harrison worked at kensho"], embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever()
template = """Answer the question based only on the following context:
{context}

Question: {question}
"""

# The prompt expects input with keys for "context" and "question"
prompt = ChatPromptTemplate.from_template(template)

model = ChatOpenAI()

retrieval_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

// 아래 3가지의 사용방식은 동일한다. 
// ref: https://python.langchain.com/v0.2/docs/how_to/parallel/
// {"context": retriever, "question": RunnablePassthrough()}
// RunnableParallel({"context": retriever, "question": RunnablePassthrough()})
// RunnableParallel(context=retriever, question=runnablePassthrough())

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

RunnablePassthrough 클래스는 RunnableSerializable을 상속받고 있다. 

class RunnablePassthrough(RunnableSerializable[Other, Other]):
    ... 중략 ..
    
    def invoke(
        self, input: Other, config: Optional[RunnableConfig] = None, **kwargs: Any
    ) -> Other:
        if self.func is not None:
            call_func_with_variable_args(
                self.func, input, ensure_config(config), **kwargs
            )
        return self._call_with_config(identity, input, config)

RunnablePassthrough 사용 예

from langchain_core.runnables import (
    RunnableLambda,
    RunnableParallel,
    RunnablePassthrough,
)

runnable = RunnableParallel(
    origin=RunnablePassthrough(),
    modified=lambda x: x+1
)

runnable.invoke(1) # {'origin': 1, 'modified': 2}


def fake_llm(prompt: str) -> str: # Fake LLM for the example
    return "completion"

chain = RunnableLambda(fake_llm) | {
    'original': RunnablePassthrough(), # Original LLM output
    'parsed': lambda text: text[::-1] # Parsing logic
}

chain.invoke('hello') # {'original': 'completion', 'parsed': 'noitelpmoc'}

여러 PromptTemplate을 병렬로 사용하기

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableParallel
from langchain_openai import ChatOpenAI

model = ChatOpenAI()
joke_chain = ChatPromptTemplate.from_template("tell me a joke about {topic}") | model
poem_chain = (
    ChatPromptTemplate.from_template("write a 2-line poem about {topic}") | model
)

map_chain = RunnableParallel(joke=joke_chain, poem=poem_chain)

map_chain.invoke({"topic": "bear"})

// 결과
{'joke': AIMessage(content="Why don't bears like fast food? Because 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_d9767fc5b9', 'finish_reason': 'stop', 'logprobs': None}, id='run-fe024170-c251-4b7a-bfd4-64a3737c67f2-0'),
 'poem': AIMessage(content='In the quiet of the forest, the bear roams free\nMajestic and wild, a sight to see.', response_metadata={'token_usage': {'completion_tokens': 24, 'prompt_tokens': 15, 'total_tokens': 39}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_c2295e73ad', 'finish_reason': 'stop', 'logprobs': None}, id='run-2707913e-a743-4101-b6ec-840df4568a76-0')}

RunnableParallel에 추가 dictionary 정보를 주고 싶을 경우 

RunnableAssign을 사용한다. 

from typing import Dict
from langchain_core.runnables.passthrough import (
    RunnableAssign,
    RunnableParallel,
)
from langchain_core.runnables.base import RunnableLambda

def add_ten(x: Dict[str, int]) -> Dict[str, int]:
    return {"added": x["input"] + 10}

mapper = RunnableParallel(
    {"add_step": RunnableLambda(add_ten),}
)

runnable_assign = RunnableAssign(mapper)

# Synchronous example
runnable_assign.invoke({"input": 5})
# returns {'input': 5, 'add_step': {'added': 15}}

# Asynchronous example
await runnable_assign.ainvoke({"input": 5})
# returns {'input': 5, 'add_step': {'added': 15}}

RunnablePassthrough.assign 에서도 사용한다. 

  class RunnablePassthrough(RunnableSerializable[Other, Other]):
    ... 중략 ...
    
    @classmethod
    def assign(
        cls,
        **kwargs: Union[
            Runnable[Dict[str, Any], Any],
            Callable[[Dict[str, Any]], Any],
            Mapping[
                str,
                Union[Runnable[Dict[str, Any], Any], Callable[[Dict[str, Any]], Any]],
            ],
        ],
    ) -> "RunnableAssign":
        """Merge the Dict input with the output produced by the mapping argument.

        Args:
            **kwargs: Runnable, Callable or a Mapping from keys to Runnables
                or Callables.

        Returns:
            A Runnable that merges the Dict input with the output produced by the
            mapping argument.
        """
        return RunnableAssign(RunnableParallel(kwargs))

Dyamic Chain 예를 보자. 

from operator import itemgetter

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import Runnable, RunnablePassthrough, chain

contextualize_instructions = """Convert the latest user question into a standalone question given the chat history. Don't answer the question, return the question and nothing else (no descriptive text)."""
contextualize_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", contextualize_instructions),
        ("placeholder", "{chat_history}"),
        ("human", "{question}"),
    ]
)
contextualize_question = contextualize_prompt | llm | StrOutputParser()

qa_instructions = (
    """Answer the user question given the following context:\n\n{context}."""
)
qa_prompt = ChatPromptTemplate.from_messages(
    [("system", qa_instructions), ("human", "{question}")]
)


@chain
def contextualize_if_needed(input_: dict) -> Runnable:
    if input_.get("chat_history"):
        # NOTE: This is returning another Runnable, not an actual output.
        return contextualize_question
    else:
        return RunnablePassthrough() | itemgetter("question")


@chain
def fake_retriever(input_: dict) -> str:
    return "egypt's population in 2024 is about 111 million"


full_chain = (
    RunnablePassthrough.assign(question=contextualize_if_needed).assign(
        context=fake_retriever
    )
    | qa_prompt
    | llm
    | StrOutputParser()
)

full_chain.invoke(
    {
        "question": "what about egypt",
        "chat_history": [
            ("human", "what's the population of indonesia"),
            ("ai", "about 276 million"),
        ],
    }
)

// 결과
"According to the context provided, Egypt's population in 2024 is estimated to be about 111 million."

Custom Function을 Runnable로 사용하기  - @chain == RunnableLambda

chaining 시에 custom function은 RunnableLambda 를 사용하거나, @chain 데코레이터를  사용한다. 

from operator import itemgetter

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda
from langchain_openai import ChatOpenAI


def length_function(text):
    return len(text)


def _multiple_length_function(text1, text2):
    return len(text1) * len(text2)


def multiple_length_function(_dict):
    return _multiple_length_function(_dict["text1"], _dict["text2"])


model = ChatOpenAI()
prompt = ChatPromptTemplate.from_template("what is {a} + {b}")

chain1 = prompt | model

chain = (
    {
        "a": itemgetter("foo") | RunnableLambda(length_function),
        "b": {"text1": itemgetter("foo"), "text2": itemgetter("bar")}
        | RunnableLambda(multiple_length_function),
    }
    | prompt
    | model
)

chain.invoke({"foo": "bar", "bar": "gah"})

// 결과
AIMessage(content='3 + 9 equals 12.', response_metadata={'token_usage': {'completion_tokens': 8, 'prompt_tokens': 14, 'total_tokens': 22}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_c2295e73ad', 'finish_reason': 'stop', 'logprobs': None}, id='run-73728de3-e483-49e3-ad54-51bd9570e71a-0')

@chain 데코레이터 사용하기 

from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import chain

prompt1 = ChatPromptTemplate.from_template("Tell me a joke about {topic}")
prompt2 = ChatPromptTemplate.from_template("What is the subject of this joke: {joke}")


@chain
def custom_chain(text):
    prompt_val1 = prompt1.invoke({"topic": text})
    output1 = ChatOpenAI().invoke(prompt_val1)
    parsed_output1 = StrOutputParser().invoke(output1)
    chain2 = prompt2 | ChatOpenAI() | StrOutputParser()
    return chain2.invoke({"joke": parsed_output1})


custom_chain.invoke("bears")

// 결과
'The subject of the joke is the bear and his girlfriend.'

chain 데코레이터 소스 코드를 보면, function을 RunnableLambda로 변환한다. 

def chain(
    func: Union[
        Callable[[Input], Output],
        Callable[[Input], Iterator[Output]],
        Callable[[Input], Coroutine[Any, Any, Output]],
        Callable[[Input], AsyncIterator[Output]],
    ],
) -> Runnable[Input, Output]:
    """Decorate a function to make it a Runnable.
    Sets the name of the Runnable to the name of the function.
    Any runnables called by the function will be traced as dependencies.

    Args:
        func: A callable.

    Returns:
        A Runnable.

    Example:

    .. code-block:: python

        from langchain_core.runnables import chain
        from langchain_core.prompts import PromptTemplate
        from langchain_openai import OpenAI

        @chain
        def my_func(fields):
            prompt = PromptTemplate("Hello, {name}!")
            llm = OpenAI()
            formatted = prompt.invoke(**fields)

            for chunk in llm.stream(formatted):
                yield chunk
    """
    return RunnableLambda(func)

또는 | 오프레이터를 통해 자동 RunnableLambda를 적용할 수 있다. 

prompt = ChatPromptTemplate.from_template("tell me a story about {topic}")

model = ChatOpenAI()

chain_with_coerced_function = prompt | model | (lambda x: x.content[:5])

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

// 결과 
'Once '

<참조>

- LangChain parallel 공식 문서: https://python.langchain.com/v0.2/docs/how_to/parallel/

- LangChain Runnable 소스: https://github.com/langchain-ai/langchain/blob/master/libs/core/langchain_core/runnables/base.py

- LangChain function 공식 문서: https://python.langchain.com/v0.2/docs/how_to/functions/

- LangChain LCEL how-to 공식문서: https://python.langchain.com/v0.2/docs/how_to/#langchain-expression-language-lcel

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/

LLM에 프롬프트시에 다양한 포멧을 정의하는 할 수 있게 도와주는 것이 LangChain의 PromptTemplate 이다. 

개념

PromptTemplate은 단순 질의시 사용한다. 

  - from_template -> PromptTemplate 의 인스턴스를 반환한다. (파이썬의 cls 용법 , 메서드 중간 * 인자 용법 참조)

class PromptTemplate(StringPromptTemplate):
    @classmethod
    def from_template(
        cls,
        template: str,
        *,
        template_format: str = "f-string",
        partial_variables: Optional[Dict[str, Any]] = None,
        **kwargs: Any,
    ) -> PromptTemplate:
        """Load a prompt template from a template.

        *Security warning*:
            Prefer using `template_format="f-string"` instead of
            `template_format="jinja2"`, or make sure to NEVER accept jinja2 templates
            from untrusted sources as they may lead to arbitrary Python code execution.

            As of LangChain 0.0.329, Jinja2 templates will be rendered using
            Jinja2's SandboxedEnvironment by default. This sand-boxing should
            be treated as a best-effort approach rather than a guarantee of security,
            as it is an opt-out rather than opt-in approach.

            Despite the sand-boxing, we recommend never using jinja2 templates
            from untrusted sources.

        Args:
            template: The template to load.
            template_format: The format of the template. Use `jinja2` for jinja2,
                             and `f-string` or None for f-strings.
                             Defaults to `f-string`.
            partial_variables: A dictionary of variables that can be used to partially
                               fill in the template. For example, if the template is
                              `"{variable1} {variable2}"`, and `partial_variables` is
                              `{"variable1": "foo"}`, then the final prompt will be
                              `"foo {variable2}"`. Defaults to None.
            kwargs: Any other arguments to pass to the prompt template.

        Returns:
            The prompt template loaded from the template.
        """

        input_variables = get_template_variables(template, template_format)
        _partial_variables = partial_variables or {}

        if _partial_variables:
            input_variables = [
                var for var in input_variables if var not in _partial_variables
            ]

        return cls(
            input_variables=input_variables,
            template=template,
            template_format=template_format,  # type: ignore[arg-type]
            partial_variables=_partial_variables,
            **kwargs,
        )

PromptTemplate 인스턴스의 invoke 호출하기 

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

# ChatOpenAI 모델을 인스턴스화합니다.
model = ChatOpenAI()
# 주어진 토픽에 대한 농담을 요청하는 프롬프트 템플릿을 생성합니다.
prompt = PromptTemplate.from_template("{topic} 에 대하여 3문장으로 설명해줘.")

# invoke 호출시에 topic 입력값을 준다. 
chain = prompt | model | StrOutputParser()
chain.invoke({"topic": "ChatGPT"})


출처: LangChain KR

LCEL에 적용할 수 있는 것은 전부 Runnable을 상속받아야 하고, Runnable의 abstractmethod인 invoke를 구현해야 한다. "chain.invoke" 호출은 최초 Runnable의 invoke를 구현하고 있는   BasePromptTemplate -->  StringPromptTemplate --> PromptTemplate 상속 관계에서 Runnable을 상속받아 invoke 메서드를 구현하고 있는 BasePromptTemplate의 invoke 보자. 

  - 입력값: Dictionary 예)  {"topic": "ChatGPT"}

  - 반환값: PromptValue 인스턴스이다. 

class BasePromptTemplate(
    RunnableSerializable[Dict, PromptValue], Generic[FormatOutputType], ABC
):
    ... 중략 ...
    
    def invoke(
        self, input: Dict, config: Optional[RunnableConfig] = None
    ) -> PromptValue:
        """Invoke the prompt.

        Args:
            input: Dict, input to the prompt.
            config: RunnableConfig, configuration for the prompt.

        Returns:
            PromptValue: The output of the prompt.
        """
        config = ensure_config(config)
        if self.metadata:
            config["metadata"] = {**config["metadata"], **self.metadata}
        if self.tags:
            config["tags"] = config["tags"] + self.tags
        return self._call_with_config(
            self._format_prompt_with_error_handling,
            input,
            config,
            run_type="prompt",
        )

LangChain KR - 프롬프트 템플릿: https://wikidocs.net/231233

ChatPromptTemplate 은 채팅 형식에서 사용한다. PromptTemplate 을 감싸서 System, Human, AI 롤(role)로 감싸면서 각각에 대한 Message 인스턴스와 MessagePromptTemplate 인스턴스를 Wrapping한다. 

  - 채팅형식: ChatPromptTemplate 사용 (system, humna, ai) 

  - 미리 프롬프트 영역 잡아 놓기: MessagePlaceholder 사용

  - from_messages 로 system, human mesage 를 구성 -> format_messages (또는 format) 으로 입력값 key=value 설정

     - LLM 호출전에 format_messages 형태로 

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

chat_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "당신은 요약 전문 AI 어시스턴트입니다. 당신의 임무는 주요 키워드로 대화를 요약하는 것입니다.",
        ),
        MessagesPlaceholder(variable_name="conversation"),
        ("human", "지금까지의 대화를 {word_count} 단어로 요약합니다."),
    ]
)

...

// 호출 case-1)
messages = chat_prompt.format_messages(
    word_count=5,
    conversation=[
        ("human", "안녕하세요! 저는 오늘 새로 입사한 테디 입니다. 만나서 반갑습니다."),
        ("ai", "반가워요! 앞으로 잘 부탁 드립니다."),
    ],
)
llm.invoke(messages)

// 호출 case-2) 또는 LCEL을 사용하여 chain 을 구성 후 입력값 설정하기 
chain = chat_template | llm
chain.invoke({word_count=5,
    conversation=[
        ("human", "안녕하세요! 저는 오늘 새로 입사한 테디 입니다. 만나서 반갑습니다."),
        ("ai", "반가워요! 앞으로 잘 부탁 드립니다."),
    ]})
    
// 출처: LangChain KR

예의 from_messages 호출 반환 chat_prompt 객체 내역

ChatPromptTemplate(input_variables=['conversation', 'word_count'], input_types={'conversation': typing.List[typing.Union[langchain_core.messages.ai.AIMessage, langchain_core.messages.human.HumanMessage, langchain_core.messages.chat.ChatMessage, langchain_core.messages.system.SystemMessage, langchain_core.messages.function.FunctionMessage, langchain_core.messages.tool.ToolMessage]]}, messages=[SystemMessagePromptTemplate(prompt=PromptTemplate(input_variables=[], template='당신은 요약 전문 AI 어시스턴트입니다. 당신의 임무는 주요 키워드로 대화를 요약하는 것입니다.')), MessagesPlaceholder(variable_name='conversation'), HumanMessagePromptTemplate(prompt=PromptTemplate(input_variables=['word_count'], template='지금까지의 대화를 {word_count} 단어로 요약합니다.'))])

예의 format_message 호출 반환 messages 객체 내역

[SystemMessage(content='당신은 요약 전문 AI 어시스턴트입니다. 당신의 임무는 주요 키워드로 대화를 요약하는 것입니다.'),
 HumanMessage(content='안녕하세요! 저는 오늘 새로 입사한 테디 입니다. 만나서 반갑습니다.'),
 AIMessage(content='반가워요! 앞으로 잘 부탁 드립니다.'),
 HumanMessage(content='지금까지의 대화를 5 단어로 요약합니다.')]

반환 타입에서 messages 서브 패키지의 다양한 메세지타입이 보인다. 이는 다음의 상속관계를 갖는다. 

BaseMessage --> SystemMessage, AIMessage, HumanMessage, ChatMessage, FunctionMessage, ToolMessage
            --> BaseMessageChunk --> SystemMessageChunk, AIMessageChunk, HumanMessageChunk, ChatMessageChunk, FunctionMessageChunk, ToolMessageChunk

여기서 BaseMessage 는 chat_models의 호출 형태로 invoke, stream 호출시 List[BaseMessage] 행태로 전달할 수 있다. 

chat models들은 모두 langchain_core.language_models 서브패키지의 BaseChatModel을 상속하고 있고, 여러 호출 메서드를 제공한다.

        +---------------------------+----------------------------------------------------------------+---------------------------------------------------------------------+--------------------------------------------------------------------------------------------------+
        | Method                    | Input                                                          | Output                                                              | Description                                                                                      |
        +===========================+================================================================+=====================================================================+==================================================================================================+
        | `invoke`                  | str | List[dict | tuple | BaseMessage] | PromptValue           | BaseMessage                                                         | A single chat model call.                                                                        |
        +---------------------------+----------------------------------------------------------------+---------------------------------------------------------------------+--------------------------------------------------------------------------------------------------+
        | `ainvoke`                 | '''                                                            | BaseMessage                                                         | Defaults to running invoke in an async executor.                                                 |
        +---------------------------+----------------------------------------------------------------+---------------------------------------------------------------------+--------------------------------------------------------------------------------------------------+
        | `stream`                  | '''                                                            | Iterator[BaseMessageChunk]                                          | Defaults to yielding output of invoke.                                                           |
        +---------------------------+----------------------------------------------------------------+---------------------------------------------------------------------+--------------------------------------------------------------------------------------------------+
        | `astream`                 | '''                                                            | AsyncIterator[BaseMessageChunk]                                     | Defaults to yielding output of ainvoke.                                                          |
        +---------------------------+----------------------------------------------------------------+---------------------------------------------------------------------+--------------------------------------------------------------------------------------------------+
        | `astream_events`          | '''                                                            | AsyncIterator[StreamEvent]                                          | Event types: 'on_chat_model_start', 'on_chat_model_stream', 'on_chat_model_end'.                 |
        +---------------------------+----------------------------------------------------------------+---------------------------------------------------------------------+--------------------------------------------------------------------------------------------------+
        | `batch`                   | List[''']                                                      | List[BaseMessage]                                                   | Defaults to running invoke in concurrent threads.                                                |
        +---------------------------+----------------------------------------------------------------+---------------------------------------------------------------------+--------------------------------------------------------------------------------------------------+
        | `abatch`                  | List[''']                                                      | List[BaseMessage]                                                   | Defaults to running ainvoke in concurrent threads.                                               |
        +---------------------------+----------------------------------------------------------------+---------------------------------------------------------------------+--------------------------------------------------------------------------------------------------+
        | `batch_as_completed`      | List[''']                                                      | Iterator[Tuple[int, Union[BaseMessage, Exception]]]                 | Defaults to running invoke in concurrent threads.                                                |
        +---------------------------+----------------------------------------------------------------+---------------------------------------------------------------------+--------------------------------------------------------------------------------------------------+
        | `abatch_as_completed`     | List[''']                                                      | AsyncIterator[Tuple[int, Union[BaseMessage, Exception]]]            | Defaults to running ainvoke in concurrent threads.                                               |
        +---------------------------+----------------------------------------------------------------+---------------------------------------------------------------------+--------------------------------------------------------------------------------------------------+

LangChain KR - 프롬프트 템플릿: https://wikidocs.net/231328

패키지

langchain_core 패키지의  prompts 와 messages 서브패키지를 사용합니다. 아래 상속관계를 보면 BaseMessagePromptTemplate 는 개별 메시지 단위로 프롬프트를 구성하는 것이고, BasePromptTemplate를  상속받은 ChatPromptTemplate은 BaseMessagePromptTemplate 상속받은 클래스들과 str, tuple등의 메세지를 담을 수 있다.

    BasePromptTemplate --> PipelinePromptTemplate
                           StringPromptTemplate --> PromptTemplate
                                                    FewShotPromptTemplate
                                                    FewShotPromptWithTemplates
                           BaseChatPromptTemplate --> AutoGPTPrompt
                                                      ChatPromptTemplate --> AgentScratchPadChatPromptTemplate



    BaseMessagePromptTemplate --> MessagesPlaceholder
                                  BaseStringMessagePromptTemplate --> ChatMessagePromptTemplate
                                                                      HumanMessagePromptTemplate
                                                                      AIMessagePromptTemplate
                                                                      SystemMessagePromptTemplate

ChatPromptTemplate의 __init__ 코드 내에서 _convert_to_message를 호출하면서 다양한 타입을 프롬프트 템플릿으로 만들 수 있다. 

class ChatPromptTemplate(BaseChatPromptTemplate):
    """Prompt template for chat models.

    Use to create flexible templated prompts for chat models.

    Examples:

        .. versionchanged:: 0.2.24

            You can pass any Message-like formats supported by
            ``ChatPromptTemplate.from_messages()`` directly to ``ChatPromptTemplate()``
            init.

        .. code-block:: python

            from langchain_core.prompts import ChatPromptTemplate

            template = ChatPromptTemplate([
                ("system", "You are a helpful AI bot. Your name is {name}."),
                ("human", "Hello, how are you doing?"),
                ("ai", "I'm doing well, thanks!"),
                ("human", "{user_input}"),
            ])

            prompt_value = template.invoke(
                {
                    "name": "Bob",
                    "user_input": "What is your name?"
                }
            )
            # Output:
            # ChatPromptValue(
            #    messages=[
            #        SystemMessage(content='You are a helpful AI bot. Your name is Bob.'),
            #        HumanMessage(content='Hello, how are you doing?'),
            #        AIMessage(content="I'm doing well, thanks!"),
            #        HumanMessage(content='What is your name?')
            #    ]
            #)

    Messages Placeholder:

        .. code-block:: python

            # In addition to Human/AI/Tool/Function messages,
            # you can initialize the template with a MessagesPlaceholder
            # either using the class directly or with the shorthand tuple syntax:

            template = ChatPromptTemplate([
                ("system", "You are a helpful AI bot."),
                # Means the template will receive an optional list of messages under
                # the "conversation" key
                ("placeholder", "{conversation}")
                # Equivalently:
                # MessagesPlaceholder(variable_name="conversation", optional=True)
            ])

            prompt_value = template.invoke(
                {
                    "conversation": [
                        ("human", "Hi!"),
                        ("ai", "How can I assist you today?"),
                        ("human", "Can you make me an ice cream sundae?"),
                        ("ai", "No.")
                    ]
                }
            )

            # Output:
            # ChatPromptValue(
            #    messages=[
            #        SystemMessage(content='You are a helpful AI bot.'),
            #        HumanMessage(content='Hi!'),
            #        AIMessage(content='How can I assist you today?'),
            #        HumanMessage(content='Can you make me an ice cream sundae?'),
            #        AIMessage(content='No.'),
            #    ]
            #)

    Single-variable template:

        If your prompt has only a single input variable (i.e., 1 instance of "{variable_nams}"),
        and you invoke the template with a non-dict object, the prompt template will
        inject the provided argument into that variable location.


        .. code-block:: python

            from langchain_core.prompts import ChatPromptTemplate

            template = ChatPromptTemplate([
                ("system", "You are a helpful AI bot. Your name is Carl."),
                ("human", "{user_input}"),
            ])

            prompt_value = template.invoke("Hello, there!")
            # Equivalent to
            # prompt_value = template.invoke({"user_input": "Hello, there!"})

            # Output:
            #  ChatPromptValue(
            #     messages=[
            #         SystemMessage(content='You are a helpful AI bot. Your name is Carl.'),
            #         HumanMessage(content='Hello, there!'),
            #     ]
            # )

    """  # noqa: E501

    messages: List[MessageLike]
    """List of messages consisting of either message prompt templates or messages."""
    validate_template: bool = False
    """Whether or not to try validating the template."""

    def __init__(
        self,
        messages: Sequence[MessageLikeRepresentation],
        *,
        template_format: Literal["f-string", "mustache", "jinja2"] = "f-string",
        **kwargs: Any,
    ) -> None:
        """Create a chat prompt template from a variety of message formats.

        Args:
            messages: sequence of message representations.
                  A message can be represented using the following formats:
                  (1) BaseMessagePromptTemplate, (2) BaseMessage, (3) 2-tuple of
                  (message type, template); e.g., ("human", "{user_input}"),
                  (4) 2-tuple of (message class, template), (4) a string which is
                  shorthand for ("human", template); e.g., "{user_input}".
            template_format: format of the template. Defaults to "f-string".
            input_variables: A list of the names of the variables whose values are
                required as inputs to the prompt.
            optional_variables: A list of the names of the variables that are optional
                in the prompt.
            partial_variables: A dictionary of the partial variables the prompt
                template carries. Partial variables populate the template so that you
                don't need to pass them in every time you call the prompt.
            validate_template: Whether to validate the template.
            input_types: A dictionary of the types of the variables the prompt template
                expects. If not provided, all variables are assumed to be strings.

        Returns:
            A chat prompt template.

        Examples:

            Instantiation from a list of message templates:

            .. code-block:: python

                template = ChatPromptTemplate([
                    ("human", "Hello, how are you?"),
                    ("ai", "I'm doing well, thanks!"),
                    ("human", "That's good to hear."),
                ])

            Instantiation from mixed message formats:

            .. code-block:: python

                template = ChatPromptTemplate([
                    SystemMessage(content="hello"),
                    ("human", "Hello, how are you?"),
                ])

        """
        _messages = [
            _convert_to_message(message, template_format) for message in messages
        ]

        # Automatically infer input variables from messages
        input_vars: Set[str] = set()
        optional_variables: Set[str] = set()
        partial_vars: Dict[str, Any] = {}
        for _message in _messages:
            if isinstance(_message, MessagesPlaceholder) and _message.optional:
                partial_vars[_message.variable_name] = []
                optional_variables.add(_message.variable_name)
            elif isinstance(
                _message, (BaseChatPromptTemplate, BaseMessagePromptTemplate)
            ):
                input_vars.update(_message.input_variables)

        kwargs = {
            **dict(
                input_variables=sorted(input_vars),
                optional_variables=sorted(optional_variables),
                partial_variables=partial_vars,
            ),
            **kwargs,
        }
        cast(Type[ChatPromptTemplate], super()).__init__(messages=_messages, **kwargs)

_convert_to_message 펑션 코드 

def _convert_to_message(
    message: MessageLikeRepresentation,
    template_format: Literal["f-string", "mustache", "jinja2"] = "f-string",
) -> Union[BaseMessage, BaseMessagePromptTemplate, BaseChatPromptTemplate]:
    """Instantiate a message from a variety of message formats.

    The message format can be one of the following:

    - BaseMessagePromptTemplate
    - BaseMessage
    - 2-tuple of (role string, template); e.g., ("human", "{user_input}")
    - 2-tuple of (message class, template)
    - string: shorthand for ("human", template); e.g., "{user_input}"

    Args:
        message: a representation of a message in one of the supported formats.
        template_format: format of the template. Defaults to "f-string".

    Returns:
        an instance of a message or a message template.

    Raises:
        ValueError: If unexpected message type.
        ValueError: If 2-tuple does not have 2 elements.
    """
    if isinstance(message, (BaseMessagePromptTemplate, BaseChatPromptTemplate)):
        _message: Union[
            BaseMessage, BaseMessagePromptTemplate, BaseChatPromptTemplate
        ] = message
    elif isinstance(message, BaseMessage):
        _message = message
    elif isinstance(message, str):
        _message = _create_template_from_message_type(
            "human", message, template_format=template_format
        )
    elif isinstance(message, tuple):
        if len(message) != 2:
            raise ValueError(f"Expected 2-tuple of (role, template), got {message}")
        message_type_str, template = message
        if isinstance(message_type_str, str):
            _message = _create_template_from_message_type(
                message_type_str, template, template_format=template_format
            )
        else:
            _message = message_type_str(
                prompt=PromptTemplate.from_template(
                    cast(str, template), template_format=template_format
                )
            )
    else:
        raise NotImplementedError(f"Unsupported message type: {type(message)}")

    return _message

API

클래스 메서드인 from_messages() 를 호출하면 cls를 반환하므로 ChatPromptTemplate 인스턴스를 반환한다.  

  - from_messages() -> ChatPromptTemplate

    @classmethod
    def from_messages(
        cls,
        messages: Sequence[MessageLikeRepresentation],
        template_format: Literal["f-string", "mustache", "jinja2"] = "f-string",
    ) -> ChatPromptTemplate:
        """Create a chat prompt template from a variety of message formats.

        Examples:

            Instantiation from a list of message templates:

            .. code-block:: python

                template = ChatPromptTemplate.from_messages([
                    ("human", "Hello, how are you?"),
                    ("ai", "I'm doing well, thanks!"),
                    ("human", "That's good to hear."),
                ])

            Instantiation from mixed message formats:

            .. code-block:: python

                template = ChatPromptTemplate.from_messages([
                    SystemMessage(content="hello"),
                    ("human", "Hello, how are you?"),
                ])

        Args:
            messages: sequence of message representations.
                  A message can be represented using the following formats:
                  (1) BaseMessagePromptTemplate, (2) BaseMessage, (3) 2-tuple of
                  (message type, template); e.g., ("human", "{user_input}"),
                  (4) 2-tuple of (message class, template), (4) a string which is
                  shorthand for ("human", template); e.g., "{user_input}".
            template_format: format of the template. Defaults to "f-string".

        Returns:
            a chat prompt template.
        """
        return cls(messages, template_format=template_format)

입력값 맵핑은 format_messages() 또는 format() 을 사용한다. 

  - format_messages() -> List[BaseMessage]

    def format_messages(self, **kwargs: Any) -> List[BaseMessage]:
        """Format the chat template into a list of finalized messages.

        Args:
            **kwargs: keyword arguments to use for filling in template variables
                      in all the template messages in this chat template.

        Returns:
            list of formatted messages.
        """
        kwargs = self._merge_partial_and_user_variables(**kwargs)
        result = []
        for message_template in self.messages:
            if isinstance(message_template, BaseMessage):
                result.extend([message_template])
            elif isinstance(
                message_template, (BaseMessagePromptTemplate, BaseChatPromptTemplate)
            ):
                message = message_template.format_messages(**kwargs)
                result.extend(message)
            else:
                raise ValueError(f"Unexpected input: {message_template}")
        return result

PromptTemplate의 from_template 인스턴스

from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate.from_template("tell me a joke about {topic}")
prompt

// 결과
PromptTemplate(input_variables=['topic'], template='tell me a joke about {topic}')

ChatPromptTemplate의 from_template 인스턴스

  - HumanMessagePromptTemplate이 PromptTemplate을 한번 더 감싼 객체

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_template("tell me a joke about {topic}")
prompt

// 결과
ChatPromptTemplate(input_variables=['topic'], messages=[HumanMessagePromptTemplate(prompt=PromptTemplate(input_variables=['topic'], template='tell me a joke about {topic}'))])

<참조>

- api: https://api.python.langchain.com/en/latest/core_api_reference.html#module-langchain_core.prompts

- LangCain Core 소스: https://github.com/langchain-ai/langchain/tree/master/libs/core/langchain_core/prompts

- LangChain KR - 프롬프트 템플릿: https://wikidocs.net/231233

- LangChain KR - 챗 프롬프트 템플릿: https://wikidocs.net/231328

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

임베딩을 통해 실수(float) 벡터값은 Vector Store에 저장을 한다. 그리고 사용자의 Query에 대해서도 벡터로 변환하여 Vector Store에 Sematic Search를 한다.

유사도가 높은 문서뭉치들을 프롬프트에 담아 LLM에 요청한다. 

패키지

vectorstores 패키지는 langchain.vectorstores 패키지에서 langchain_community.vectorstores 패키지의 모듈을 re-export 하고 있다. 

- 70 개가량의 벡터 저장소 구현체를 제공한다. https://python.langchain.com/v0.2/docs/integrations/vectorstores/

- 가장 많이 사용하는 것은 FAISS, Chroma 등이다. 

Chroma 패키지는 내부적으로 chromadb 패키지를 사용하여 LangChain과 연동시켜주는 인터페이스이다.

- VectoreStore 생성시 "embedding_function", "collection_name"을 지정할 수 있다. 

from chromadb import Client
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

# 1. OpenAI Embeddings 초기화
openai_embedding = OpenAIEmbeddings()

# 2. Chroma 벡터스토어 클라이언트 생성
client = Client()

# 3. Chroma 벡터스토어 생성
vectorstore = Chroma(
    embedding_function=openai_embedding,  # OpenAI Embedding 사용
    collection_name="my_openai_collection",  # 컬렉션 이름
    client=client                           # Chroma 클라이언트
)

# 4. 데이터 추가
documents = ["This is a test document.", "Another document for testing."]
vectorstore.add_texts(texts=documents)

# 5. 데이터 검색
query = "test"
results = vectorstore.similarity_search(query, k=2)

# 6. 검색 결과 출력
for result in results:
    print(f"Document: {result.page_content}, Score: {result.score}")

또는 from_documents 클래스메서드를 이용하여 호출한다. 

- SentenceTransformerEmbeddings 는 내부적으로 HuggingFaceEmbeddings 을 사용하고 디폴디 임베딩 모델로 "BAAI/bge-large-en"을 사용한다. 

- load() --> split_documents() --> from_documents() 를 통해서 

# import
from langchain_community.document_loaders import TextLoader
from langchain_community.embeddings.sentence_transformer import (
    SentenceTransformerEmbeddings,
)
from langchain_community.vectorstores import Chroma
from langchain_text_splitters import CharacterTextSplitter

# 문서를 로드하고 청크로 분할합니다.
loader = TextLoader("./data/appendix-keywords.txt")
documents = loader.load()

# 문서를 청크로 분할합니다.
text_splitter = CharacterTextSplitter(chunk_size=300, chunk_overlap=0)
docs = text_splitter.split_documents(documents)

# 오픈 소스 임베딩 함수를 생성합니다.
stf_embeddings = SentenceTransformerEmbeddings(model_name="all-MiniLM-L6-v2")

# Chroma에 로드합니다.
db = Chroma.from_documents(docs, stf_embeddings)

# 질의합니다.
query = "What is Word2Vec?"
docs = db.similarity_search(query)

# 결과를 출력합니다.
print(docs[0].page_content)

API

VectorStore 드라이버 구현체는 langchain_core의 VectoreStore를 상속받아 구현한다. 

- VectorStore 를 상속받아 __init__ 구현 

 class Chroma(VectorStore):  
   def __init__(
        self,
        collection_name: str = _LANGCHAIN_DEFAULT_COLLECTION_NAME,
        embedding_function: Optional[Embeddings] = None,
        persist_directory: Optional[str] = None,
        client_settings: Optional[chromadb.config.Settings] = None,
        collection_metadata: Optional[Dict] = None,
        client: Optional[chromadb.Client] = None,
        relevance_score_fn: Optional[Callable[[float], float]] = None,
    ) -> None:
        """Initialize with a Chroma client."""
        try:
            import chromadb
            import chromadb.config
        except ImportError:
            raise ImportError(
                "Could not import chromadb python package. "
                "Please install it with `pip install chromadb`."
            )

        if client is not None:
            self._client_settings = client_settings
            self._client = client
            self._persist_directory = persist_directory
        else:
            if client_settings:
                # If client_settings is provided with persist_directory specified,
                # then it is "in-memory and persisting to disk" mode.
                client_settings.persist_directory = (
                    persist_directory or client_settings.persist_directory
                )
                if client_settings.persist_directory is not None:
                    # Maintain backwards compatibility with chromadb < 0.4.0
                    major, minor, _ = chromadb.__version__.split(".")
                    if int(major) == 0 and int(minor) < 4:
                        client_settings.chroma_db_impl = "duckdb+parquet"

                _client_settings = client_settings
            elif persist_directory:
                # Maintain backwards compatibility with chromadb < 0.4.0
                major, minor, _ = chromadb.__version__.split(".")
                if int(major) == 0 and int(minor) < 4:
                    _client_settings = chromadb.config.Settings(
                        chroma_db_impl="duckdb+parquet",
                    )
                else:
                    _client_settings = chromadb.config.Settings(is_persistent=True)
                _client_settings.persist_directory = persist_directory
            else:
                _client_settings = chromadb.config.Settings()
            self._client_settings = _client_settings
            self._client = chromadb.Client(_client_settings)
            self._persist_directory = (
                _client_settings.persist_directory or persist_directory
            )

        self._embedding_function = embedding_function
        self._collection = self._client.get_or_create_collection(
            name=collection_name,
            embedding_function=None,
            metadata=collection_metadata,
        )
        self.override_relevance_score_fn = relevance_score_fn

- from_documents 는 classmethod 이면서 벡터객체를 반환한다.

   - documents: splitted 된 document 리스트

   - embedding: embedding model 인스턴스

    @classmethod
    def from_documents(
        cls: Type[VST],
        documents: List[Document],
        embedding: Embeddings,
        **kwargs: Any,
    ) -> VST:
        """Return VectorStore initialized from documents and embeddings.

        Args:
            documents: List of Documents to add to the vectorstore.
            embedding: Embedding function to use.
            **kwargs: Additional keyword arguments.

        Returns:
            VectorStore: VectorStore initialized from documents and embeddings.
        """
        texts = [d.page_content for d in documents]
        metadatas = [d.metadata for d in documents]
        return cls.from_texts(texts, embedding, metadatas=metadatas, **kwargs)

내부에서 클래스메서드인 from_texts를 호출한다. from_texts는 abstractmethod로 각 vectorstore에서 구현한다. 

    @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 는 abstractmethod로 각 vectorstore에서 구현을 해야 한다. 

    @abstractmethod
    def similarity_search(
        self, query: str, k: int = 4, **kwargs: Any
    ) -> List[Document]:
        """Return docs most similar to query.

        Args:
            query: Input text.
            k: Number of Documents to return. Defaults to 4.
            **kwargs: Arguments to pass to the search method.

        Returns:
            List of Documents most similar to the query.
        """

- search 는 검색 타입을 선택한다. "similarity", "similarity_score_threshold", "mmr" 등이 있다. 

  - 반환값은 List[Document] 이다. 

    def search(self, query: str, search_type: str, **kwargs: Any) -> List[Document]:
        """Return docs most similar to query using a specified search type.

        Args:
            query: Input text
            search_type: Type of search to perform. Can be "similarity",
                "mmr", or "similarity_score_threshold".
            **kwargs: Arguments to pass to the search method.

        Returns:
            List of Documents most similar to the query.

        Raises:
            ValueError: If search_type is not one of "similarity",
                "mmr", or "similarity_score_threshold".
        """
        if search_type == "similarity":
            return self.similarity_search(query, **kwargs)
        elif search_type == "similarity_score_threshold":
            docs_and_similarities = self.similarity_search_with_relevance_scores(
                query, **kwargs
            )
            return [doc for doc, _ in docs_and_similarities]
        elif search_type == "mmr":
            return self.max_marginal_relevance_search(query, **kwargs)
        else:
            raise ValueError(
                f"search_type of {search_type} not allowed. Expected "
                "search_type to be 'similarity', 'similarity_score_threshold'"
                " or 'mmr'."
            )

- as_retriever 는 VectorStore 에는 VectorStoreRetriever 를 생성하여 반환한다. 

def as_retriever(self, **kwargs: Any) -> VectorStoreRetriever:

Retriever 에 대해 다음 글에서 살펴보자. 

<참조>

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

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

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

개념

텍스트 임베딩은 텍스트 조각을 벡터 수치로 생성한다. 텍스트 임베딩 구현체는 LLM 관련 업체를 통해 제공된다. 예로 들면 J2EE에서 JDBC 스펙을 정의하면, Oracle에서 Oracle JDBC 구현 드라이브 라이브러리를 제공하는 방식이다. LangChain이 임베팅 스펙 인터페이스를 만들고, 파트너사가 임베딩 구현체를 제공한다. (구현체를 LangChain 에서 직접 만들었을 수도 있다.)

텍스트 임베딩 모델을 통해서 백터 공간을 만드는 목적은 "Semantic Search"시에 말뭉치 쿼리에 대해 벡터 공간에서 유사성 검색이 효율적이기 때문이다.

먼저 임베딩 모델 과 의미적 검색에 대한 개념을 살펴보자.

임베딩과 임베딩 모델 이해

"임베딩" 과 "임베딩 모델"은 다른 의미를 내포하고 있다. LangChain의 Embedding models 원문 참조

Embedding models create a vector representation of a piece of text. You can think of a vector as an array of numbers that captures the semantic meaning of the text. By representing the text in this way, you can perform mathematical operations that allow you to do things like search for other pieces of text that are most similar in meaning. These natural language search capabilities underpin many types of context retrieval, where we provide an LLM with the relevant data it needs to effectively respond to a query.

1) 임베딩 (Embedding)

임베딩은 단어, 문장 또는 문서와 같은 텍스트 데이터를 고정된 크기의 벡터로 변환하는 과정 또는 결과를 의미합니다. 이 벡터는 원본 텍스트의 의미를 숫자 형태로 표현하여 기계 학습 모델이 이해하고 처리할 수 있도록 합니다. 임베딩은 다양한 NLP 작업에서 사용되며, 유사성 계산, 분류, 군집화, 검색 등 여러 응용 분야에서 중요한 역할을 합니다.

특징:

• 고정 크기 벡터: 텍스트 데이터를 고정된 크기의 숫자 벡터로 변환합니다.

• 의미 보존: 텍스트의 의미와 문맥 정보를 벡터에 포함시킵니다.

• 유사성 계산: 벡터 간의 유사성을 계산하여 텍스트 간의 관계를 파악할 수 있습니다.

예시:

embedding = [0.1, 0.2, 0.3, 0.4, 0.5]  # "example"이라는 단어의 임베딩 벡터

2) 임베딩 모델 (Embedding Model)

임베딩 모델은 텍스트 데이터를 임베딩 벡터로 변환하는 알고리즘이나 기계 학습 모델을 의미합니다. 이 모델은 대량의 텍스트 데이터를 학습하여 각 단어, 문장 또는 문서에 대해 적절한 임베딩을 생성하는 방법을 배웁니다. 임베딩 모델은 일반적으로 기계 학습 또는 딥러닝 기술을 사용하여 구현됩니다.

특징:

• 학습된 모델: 대규모 텍스트 데이터셋을 학습하여 임베딩을 생성하는 방법을 배웁니다.

• 다양한 아키텍처: Word2Vec, GloVe, BERT, GPT 등 다양한 아키텍처가 존재합니다.

• 응용 분야: 자연어 처리(NLP), 정보 검색, 텍스트 분류 등 다양한 분야에서 사용됩니다.

예시:

from transformers import BertModel, BertTokenizer

# BERT 임베딩 모델 로드
tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')
model = BertModel.from_pretrained('bert-base-uncased')

# 텍스트를 임베딩 벡터로 변환
text = "example sentence"
inputs = tokenizer(text, return_tensors='pt')
outputs = model(**inputs)
embedding = outputs.last_hidden_state

요약

• 임베딩 (Embedding): 텍스트 데이터를 고정된 크기의 벡터로 변환한 결과. 예를 들어, 단어 “example”의 벡터 표현 [0.1, 0.2, 0.3, 0.4, 0.5].

• 임베딩 모델 (Embedding Model): 텍스트 데이터를 임베딩 벡터로 변환하는 알고리즘이나 기계 학습 모델. 예를 들어, BERT, Word2Vec 등.

임베딩은 데이터를 벡터로 표현하는 결과를 말하고, 임베딩 모델은 이러한 벡터를 생성하는 알고리즘이나 모델을 의미한다.

written by GPT

다음글에서 임베딩 모델에 대한 상세 설명을 참조하자. 

https://brunch.co.kr/@harryban0917/209

https://brunch.co.kr/@harryban0917/210

의미적 검색(Semantic Search)란?

의미적 검색(Semantic Search)은 단순히 키워드 일치에 기반한 전통적인 검색 방식과 달리, 검색 쿼리와 문서의 의미를 이해하고 이를 기반으로 관련 결과를 반환하는 검색 방식을 말합니다. 이는 자연어 처리와 기계 학습 기술을 활용하여 문장의 맥락과 의미를 파악하고, 사용자 의도를 더 잘 이해하여 보다 정확한 검색 결과를 제공합니다.

의미적 검색의 주요 특징

1. 의도 이해: 단순한 키워드가 아닌 사용자의 검색 의도를 이해합니다. 예를 들어, “서울에서 맛있는 음식점”을 검색하면 “서울의 맛집 추천”과 같은 유사한 의미의 결과를 반환합니다.

2. 문맥 파악: 검색 쿼리와 문서의 전체 문맥을 분석하여 관련성 높은 결과를 도출합니다. 단어의 위치나 사용 방식에 따라 다른 의미를 가질 수 있는 경우에도 이를 이해합니다.

3. 동의어 및 관련어 처리: 동의어나 유사어를 인식하여 더 포괄적인 검색 결과를 제공합니다. 예를 들어, “자동차”를 검색했을 때 “차량”이라는 단어가 포함된 문서도 검색 결과에 포함될 수 있습니다.

의미적 검색의 예시

기존의 키워드 검색에서는 “애플 제품”이라는 쿼리에 대해 “애플”과 “제품”이라는 단어가 정확히 포함된 문서만 반환할 가능성이 높습니다. 반면 의미적 검색에서는 “애플이 만든 기기”, “애플이 출시한 최신 제품” 등과 같이 단어는 다르지만 의미가 유사한 문서도 결과에 포함될 수 있습니다.

의미적 검색의 구현

의미적 검색을 구현하기 위해 일반적으로 다음과 같은 기술이 사용됩니다:

1. 임베딩(Embedding): 단어, 문장 또는 문서를 벡터 형태로 변환하여 의미적 유사성을 계산할 수 있도록 합니다.

2. 벡터 검색(Vector Search): 쿼리와 문서의 벡터를 비교하여 유사한 벡터를 가진 문서를 검색합니다.

3. 자연어 처리(NLP) 모델: BERT, GPT 등과 같은 최신 언어 모델을 사용하여 문장의 의미를 이해하고 임베딩을 생성합니다.

LangChaind을 사용하여 의미적 검색을 구현한 예시를 보자.

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
from langchain.document_loaders import TextLoader

# 문서 로드
documents = TextLoader("path_to_documents").load()

# 임베딩 모델 초기화
embeddings = OpenAIEmbeddings()

# 벡터 스토어에 임베딩 저장
# from_documents를 호출하면 내부에서 embed_documents을 호출하고 있다. 
# 참조) https://github.com/langchain-ai/langchain/blob/master/libs/community/langchain_community/vectorstores/faiss.py#L1039
# document_embeddings = embeddings.embed_documents(documents)
vector_store = FAISS.from_documents(documents, embeddings)

# 벡터 스토어 쿼리
query = "서울에서 맛있는 음식점 추천"
query_embedding = embeddings.embed_query(query)
results = vector_store.similarity_search(query_embedding)

print(results)

이 예시에서는 텍스트 문서를 임베딩하여 벡터로 변환하고, 검색 쿼리도 임베딩하여 벡터 스토어에서 유사한 문서를 검색합니다. 이를 통해 쿼리와 의미적으로 관련된 문서를 효율적으로 찾을 수 있다. written by GPT

패키지

임베딩 모델 구현체들 https://python.langchain.com/v0.2/docs/integrations/text_embedding/

임베딩 구현체는 3가지 패키지 경로를 통해 import 할 수 있다. 

1)  langchain.embedding 패키지를 통한 방법: 내부적으로 langchain_community.embeddings 패키지를 re-export 한다.

from typing import TYPE_CHECKING, Any

from langchain._api import create_importer

if TYPE_CHECKING:
    from langchain_community.embeddings import OllamaEmbeddings

# Create a way to dynamically look up deprecated imports.
# Used to consolidate logic for raising deprecation warnings and
# handling optional imports.
DEPRECATED_LOOKUP = {"OllamaEmbeddings": "langchain_community.embeddings"}

_import_attribute = create_importer(__package__, deprecated_lookups=DEPRECATED_LOOKUP)


def __getattr__(name: str) -> Any:
    """Look up attributes dynamically."""
    return _import_attribute(name)


__all__ = [
    "OllamaEmbeddings",
]

2) langchain_community.embeddings 패키지를 통한 방법: 커뮤니티를 통해 구현된 구현체

3) partners 에서 제공하는 임베딩으로 예로 langchain_ollama.embeddings 모듈을 통한 방법: ollama 업체에서 구현한 구현체

참조) OllamaEmbeddings 의 2) 와 3) 에서 community와 partners에서 제공하는 구현체의 내용이 다르다. 

Langchain의 OllamaEmbeddings 모듈은 두 가지 다른 패키지에서 제공되며, 각기 다른 기능과 구현 방식을 가질 수 있습니다. 이 두 패키지는 langchain_community와 langchain_ollama입니다. 여기에서는 이 두 가지 모듈의 차이점을 설명하겠습니다.

1. langchain_community의 OllamaEmbeddings

langchain_community 패키지의 OllamaEmbeddings는 커뮤니티에서 개발된 통합을 포함합니다. 이 패키지는 다양한 써드파티 통합을 제공하며, 사용자가 필요에 맞게 쉽게 활용할 수 있도록 설계되었습니다. 이 패키지의 OllamaEmbeddings는 다음과 같은 특징을 가질 수 있습니다:

• 다양한 통합: 여러 벡터 스토어 및 임베딩 모델과의 통합을 제공합니다.

• 커뮤니티 지원: 오픈 소스 커뮤니티에 의해 유지 관리되며, 다양한 사용자 요구사항을 반영합니다.

2. langchain_ollama의 OllamaEmbeddings

langchain_ollama 패키지의 OllamaEmbeddings는 특정 벤더나 API와의 통합을 염두에 두고 설계된 경우가 많습니다. 이 패키지는 보다 특정한 기능이나 최적화를 제공할 수 있습니다. 예를 들어, 특정한 임베딩 모델이나 벡터 스토어에 최적화된 구현이 포함될 수 있습니다.

• 벤더 특화: 특정 벤더의 기술 스택에 최적화된 기능을 제공합니다.

• 고성능 최적화: 특정 사용 사례에 맞춘 최적화가 포함될 수 있습니다.

주요 차이점 요약

1. 목적과 통합 대상:

• langchain_community는 다양한 통합을 지원하며, 커뮤니티에서 널리 사용되는 기능을 제공합니다.

• langchain_ollama는 특정 벤더나 API와의 통합을 염두에 두고 설계되어, 특정 사용 사례에 최적화된 기능을 제공합니다.

2. 사용자 기반:

• langchain_community는 오픈 소스 커뮤니티에 의해 유지 관리되며, 다양한 사용자 요구사항을 반영합니다.

• langchain_ollama는 특정 벤더나 기술 스택을 사용하는 사용자에게 더 적합할 수 있습니다.

이와 같은 차이점 때문에, 사용자는 자신의 요구사항에 맞는 패키지를 선택하여 사용할 수 있습니다. 예를 들어, 다양한 임베딩 모델과의 통합이 필요하다면 langchain_community를, 특정 벤더의 최적화된 기능이 필요하다면 langchain_ollama를 선택하는 것이 좋습니다.

written by GPT

주의할 점)

langchain_community 의 OllamaEmbeddings 는 아직 deprecated 명시가 없지만, langchain_community에 있는 OpenAIEmbeddings는 deprecated 명시가 있기 때문에 langchain_openai 패키지를 설치하여 사용한다. 

// langchain_community
@deprecated(
    since="0.0.9",
    removal="0.3.0",
    alternative_import="langchain_openai.OpenAIEmbeddings",
)
class OpenAIEmbeddings(BaseModel, Embeddings):
  ...
  
// langchain_openai 패키지를 사용한다. 
class OpenAIEmbeddings(BaseModel, Embeddings):
    """OpenAI embedding models.

    To use, you should have the
    environment variable ``OPENAI_API_KEY`` set with your API key or pass it
    as a named parameter to the constructor.

    In order to use the library with Microsoft Azure endpoints, use
    AzureOpenAIEmbeddings.

    Example:
        .. code-block:: python

            from langchain_openai import OpenAIEmbeddings

            model = OpenAIEmbeddings(model="text-embedding-3-large")
    """

API

구현체의 이름은 <name>Embeddings 복수로 한다.

LangChain 에서는 langchain_core 패키지에 있는 Embedding 클래스에서 두가지 추상메소드를 구현토록 한다.

  - embed_documents : List[str] -> List[List[float]]

  - embed_query : str -> List[float]

class Embeddings(ABC):
    @abstractmethod
    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        """Embed search docs.

        Args:
            texts: List of text to embed.

        Returns:
            List of embeddings.
        """

    @abstractmethod
    def embed_query(self, text: str) -> List[float]:
        """Embed query text.

        Args:
            text: Text to embed.

        Returns:
            Embedding.
        """
    ... 중략 ...

다음 글에서는 임베딩 모델을 거쳐 벡터화된 값을 저장하는 VectorStore에 대해 살펴본다.

<참조>

- API: https://api.python.langchain.com/en/latest/langchain_api_reference.html#module-langchain.embeddings

- langchain_core의 Embedding 클래스: https://github.com/langchain-ai/langchain/blob/master/libs/core/langchain_core/embeddings/embeddings.py

- LangChain embedding 설명글: https://python.langchain.com/v0.2/docs/concepts/#embedding-models

- text embedding 사용법: https://python.langchain.com/v0.2/docs/how_to/embed_text/

- LangChain Tutorial 코스: https://python.langchain.com/v0.2/docs/additional_resources/tutorials/

- 임베딩의 이해: https://www.syncly.kr/blog/what-is-embedding-and-how-to-use

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

- 한글 임베딩 모델: https://github.com/Atipico1/Kor-IR

- 데이터셋: https://github.com/HeegyuKim/open-korean-instructions

개념

문서를 로딩했으면 의미조각(Chunk)으로 쪼개는 splitting 작업을 한다. 이는 LLM에 인풋으로 들어가는 프롬프트 길이의 제약이 있기 때문에 문서가 길 경우 의미조각으로 나눠서 저장 후 사용자 쿼리에 가장 유사한 의미조각을 찾기 위한 용도이다. 따라서 문서의 특성에 따른 Splitter를 잘 선택해야 한다.

패키지

langchain_text_splitters 패키지가 별도로 분리되어 있다. 두가지 방법으로 import 할 수 있다. 

from langchain.document_loaders import PyPDFLoader
from langchain_text_splitter import RecursiveCharacterTextSplitter
// 또는
// from langchain_text_splitters import RecursiveCharacterTextSplitter

// 예1) 문서 splitting
loader = PyPDFLoader(FILE_PATH)
docs = loader.load()

text_splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=50)
splitted_docs = text_splitter.split_documents(docs)

for i, document in enumerate(splitted_docs):
    print(f"Document {i+1}:\n{document}\n")


// 예2) 텍스트 splitting
text = """
LangChain은 대규모 언어 모델(LLM)을 활용하여 복잡한 응용 프로그램을 구축하기 위한 도구입니다. 
이 도구는 프롬프트 템플릿, 모델, 출력 파서 등을 포함한 다양한 컴포넌트를 제공하여, 
사용자가 쉽게 체인을 구성할 수 있도록 도와줍니다.
"""

chunks = text_splitter.split_text(text)
for i, chunk in enumerate(chunks):
    print(f"Chunk {i+1}:\n{chunk}\n")

Splitter의 종류

1. RecursiveCharacterTextSplitter:

• 분할 기준: 사용자 정의 문자를 기준으로 재귀적으로 분할합니다.

• 특징: 관련된 텍스트 조각을 최대한 근접하게 유지하려고 합니다. 텍스트를 나누는 기본 방법으로 권장됩니다.

2. CharacterTextSplitter:

• 분할 기준: 단순히 특정 문자(예: 공백, 줄 바꿈 등)를 기준으로 분할합니다.

• 특징: 간단한 분할 작업에 적합합니다.

3. TokenTextSplitter:

• 분할 기준: 토큰 단위로 텍스트를 분할합니다.

• 특징: 언어 모델의 토큰화 방식에 따라 텍스트를 나눌 수 있어, 모델의 최대 토큰 길이에 맞춰 텍스트를 분할할 때 유용합니다.

4. SentenceTextSplitter:

• 분할 기준: 문장 단위로 텍스트를 분할합니다.

• 특징: 문장을 기준으로 텍스트를 나누기 때문에 자연스러운 분할이 가능합니다.

5. MarkdownTextSplitter:

• 분할 기준: 마크다운 문서의 구조(헤더, 리스트, 코드 블록 등)를 기준으로 분할합니다.

• 특징: 마크다운 문서를 처리할 때 유용하며, 문서의 논리적 구조를 유지하면서 분할할 수 있습니다.

6. HTMLTextSplitter:

• 분할 기준: HTML 태그를 기준으로 텍스트를 분할합니다.

• 특징: HTML 문서를 처리할 때 유용하며, 문서의 시각적 구조를 유지하면서 분할할 수 있습니다.

7. LanguageSpecificTextSplitter:

• 분할 기준: 특정 언어의 구문을 기준으로 텍스트를 분할합니다.

• 특징: 언어별로 특화된 분할 방식을 사용할 수 있어, 코드나 특정 언어로 작성된 텍스트를 효과적으로 분할할 수 있습니다.

API

주로 다루는 것이 문자일 경우 TextSplitter를 상속받은 RecursiveCharacterTextSplitter를 많이 사용한다. 

  - split_text 메소드는 주로 문자 스트링을 List[str] 로 반환할 때 사용한다. abstractmethod 이다. 

  - create_documents 는 List[str] 을 List[Document] 로 반환할 때 사용한다.
  - split_documents 주로 문서 파일을 load한 후 다시 작은 단위조각의 List[Document] 로 반환할 때 사용한다.

      - Document의 page_content를 List[str] 로 만들어 create_documents를 호출하고 있다.

      - create_documents는 chunk_size, chunk_overlap에 따라 텍스트 조각을 만드는 split_text를 호출하고 있다.

// TextSplitter
class TextSplitter(BaseDocumentTransformer, ABC):
    """Interface for splitting text into chunks."""

    def __init__(
        self,
        chunk_size: int = 4000,
        chunk_overlap: int = 200,
        length_function: Callable[[str], int] = len,
        keep_separator: Union[bool, Literal["start", "end"]] = False,
        add_start_index: bool = False,
        strip_whitespace: bool = True,
    ) -> None:
        """Create a new TextSplitter.

        Args:
            chunk_size: Maximum size of chunks to return
            chunk_overlap: Overlap in characters between chunks
            length_function: Function that measures the length of given chunks
            keep_separator: Whether to keep the separator and where to place it
                            in each corresponding chunk (True='start')
            add_start_index: If `True`, includes chunk's start index in metadata
            strip_whitespace: If `True`, strips whitespace from the start and end of
                              every document
        """
        if chunk_overlap > chunk_size:
            raise ValueError(
                f"Got a larger chunk overlap ({chunk_overlap}) than chunk size "
                f"({chunk_size}), should be smaller."
            )
        self._chunk_size = chunk_size
        self._chunk_overlap = chunk_overlap
        self._length_function = length_function
        self._keep_separator = keep_separator
        self._add_start_index = add_start_index
        self._strip_whitespace = strip_whitespace

    @abstractmethod
    def split_text(self, text: str) -> List[str]:
        """Split text into multiple components."""

    def create_documents(
        self, texts: List[str], metadatas: Optional[List[dict]] = None
    ) -> List[Document]:
        """Create documents from a list of texts."""
        _metadatas = metadatas or [{}] * len(texts)
        documents = []
        for i, text in enumerate(texts):
            index = 0
            previous_chunk_len = 0
            for chunk in self.split_text(text):
                metadata = copy.deepcopy(_metadatas[i])
                if self._add_start_index:
                    offset = index + previous_chunk_len - self._chunk_overlap
                    index = text.find(chunk, max(0, offset))
                    metadata["start_index"] = index
                    previous_chunk_len = len(chunk)
                new_doc = Document(page_content=chunk, metadata=metadata)
                documents.append(new_doc)
        return documents

    def split_documents(self, documents: Iterable[Document]) -> List[Document]:
        """Split documents."""
        texts, metadatas = [], []
        for doc in documents:
            texts.append(doc.page_content)
            metadatas.append(doc.metadata)
        return self.create_documents(texts, metadatas=metadatas)
    ...

RecursiveCharaterTextSplitter 의 split_text 구현 내역

  - abstract method 인 split_text를 구현함.

class RecursiveCharacterTextSplitter(TextSplitter):
    """Splitting text by recursively look at characters.

    Recursively tries to split by different characters to find one
    that works.
    """

    def __init__(
        self,
        separators: Optional[List[str]] = None,
        keep_separator: bool = True,
        is_separator_regex: bool = False,
        **kwargs: Any,
    ) -> None:
        """Create a new TextSplitter."""
        super().__init__(keep_separator=keep_separator, **kwargs)
        self._separators = separators or ["\n\n", "\n", " ", ""]
        self._is_separator_regex = is_separator_regex

    def _split_text(self, text: str, separators: List[str]) -> List[str]:
        """Split incoming text and return chunks."""
        final_chunks = []
        # Get appropriate separator to use
        separator = separators[-1]
        new_separators = []
        for i, _s in enumerate(separators):
            _separator = _s if self._is_separator_regex else re.escape(_s)
            if _s == "":
                separator = _s
                break
            if re.search(_separator, text):
                separator = _s
                new_separators = separators[i + 1 :]
                break

        _separator = separator if self._is_separator_regex else re.escape(separator)
        splits = _split_text_with_regex(text, _separator, self._keep_separator)

        # Now go merging things, recursively splitting longer texts.
        _good_splits = []
        _separator = "" if self._keep_separator else separator
        for s in splits:
            if self._length_function(s) < self._chunk_size:
                _good_splits.append(s)
            else:
                if _good_splits:
                    merged_text = self._merge_splits(_good_splits, _separator)
                    final_chunks.extend(merged_text)
                    _good_splits = []
                if not new_separators:
                    final_chunks.append(s)
                else:
                    other_info = self._split_text(s, new_separators)
                    final_chunks.extend(other_info)
        if _good_splits:
            merged_text = self._merge_splits(_good_splits, _separator)
            final_chunks.extend(merged_text)
        return final_chunks

    def split_text(self, text: str) -> List[str]:
        return self._split_text(text, self._separators)

Custom Text Splitter 만들기 

Custom text splitter를 만들려면 TextSplitter 클래스를 상속받아 자신만의 분할 로직을 구현할 수 있습니다. 다음은 간단한 예제로, 특정 구분자(예: 줄 바꿈 문자)를 기준으로 텍스트를 분할하는 custom text splitter를 만드는 방법을 보여드립니다.

1. CustomTextSplitter 클래스 정의:

TextSplitter 클래스를 상속받아 CustomTextSplitter 클래스를 정의합니다. 생성자에서 구분자, 청크 크기, 겹침 크기를 설정할 수 있습니다.

2. split_text 메서드 구현:

split_text 메서드는 주어진 텍스트를 구분자를 기준으로 분할한 다음, 지정된 청크 크기와 겹침 크기를 기준으로 청크를 생성합니다.

3. 청크 겹침 처리:

청크 간에 겹침을 추가하여 텍스트가 자연스럽게 이어지도록 합니다.

4. CustomTextSplitter 사용:

CustomTextSplitter 인스턴스를 생성하고, split_text 메서드를 호출하여 텍스트를 분할한 후 결과를 출력합니다.

이 예제는 간단한 구분자를 사용한 텍스트 분할을 보여주지만, 실제로는 더 복잡한 분할 로직을 구현할 수 있습니다. 예를 들어, 문단 단위로 분할하거나, 특정 패턴을 기준으로 분할하는 등 다양한 요구에 맞춰 커스터마이징할 수 있습니다.

from langchain.text_splitter import TextSplitter
from typing import List

class CustomTextSplitter(TextSplitter):
    def __init__(self, separator: str = "\n", chunk_size: int = 100, chunk_overlap: int = 20):
        self.separator = separator
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap

    def split_text(self, text: str) -> List[str]:
        # 구분자를 기준으로 텍스트를 분할
        splits = text.split(self.separator)
        
        chunks = []
        current_chunk = ""
        
        for split in splits:
            if len(current_chunk) + len(split) + len(self.separator) > self.chunk_size:
                chunks.append(current_chunk)
                current_chunk = split
            else:
                if current_chunk:
                    current_chunk += self.separator
                current_chunk += split
        
        # 마지막 남은 텍스트 조각 추가
        if current_chunk:
            chunks.append(current_chunk)
        
        # 겹침 처리
        if self.chunk_overlap > 0 and len(chunks) > 1:
            overlapped_chunks = []
            for i in range(len(chunks)):
                if i > 0:
                    overlap = chunks[i-1][-self.chunk_overlap:] + self.separator
                    overlapped_chunks.append(overlap + chunks[i])
                else:
                    overlapped_chunks.append(chunks[i])
            chunks = overlapped_chunks
        
        return chunks

# CustomTextSplitter 사용 예제
text = """
LangChain은 대규모 언어 모델(LLM)을 활용하여 복잡한 응용 프로그램을 구축하기 위한 도구입니다.
이 도구는 프롬프트 템플릿, 모델, 출력 파서 등을 포함한 다양한 컴포넌트를 제공하여,
사용자가 쉽게 체인을 구성할 수 있도록 도와줍니다.
"""

splitter = CustomTextSplitter(separator="\n", chunk_size=100, chunk_overlap=20)
chunks = splitter.split_text(text)

for i, chunk in enumerate(chunks):
    print(f"Chunk {i+1}:\n{chunk}\n")

written by GPT for custom splitter

<참조>

- API: https://api.python.langchain.com/en/latest/text_splitters_api_reference.html

- 소스: https://github.com/langchain-ai/langchain/tree/master/libs/text-splitters/langchain_text_splitters

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

LangChain 의 Document Loaders 에 대한 개념 및 패키지, API를 들여다 보자. 

개념

LLM은 대규모 언어모델로 언어가 텍스트 또는 음성으로 인풋을 받다. 이때 텍스트는 사용자가 직접 입력한 내용일 수도 있고, 첨부한 문서일 수도 있고, 텍스트가 있는 소스에 접근할 수도 있다.

- 다양한 문서 포멧 로딩

- 내용을 축출하기

- 내용을 원하는 단위로 잘 쪼개기

문서 종류 와 위치 (170개 가량)

- 파일: text, csv, pdf, markdown, docx, excel, json, ...

- 위치: folder, database, url, ...

- 서비스: youtube, slack, gitbook, github, git, discord, docusaurus, figma, ...

- LangChain의 document loaders 개념 설명

- LangChain의 document loaders 가이드

패키지

document_loaders 구현체들 https://python.langchain.com/v0.2/docs/integrations/document_loaders/

langchain 패키지 소스를 보면 document_loaders 패키지가 존재하고, 대부분 langchain_community 패키지 소스를 re-exporting 하고 있다. 

패키지 및 로더 호출 방법

- 기본 langchain.document_loaders 에서 xxxLoader 를 임포트하여 사용한다. 

//----------
// 예-1)
from langchain.document_loaders import PyPDFLoader
// 또는 from langchain_community.document_loaders import PyPDFLoader

loader = PyPDFLoader(FILE_PATH)
pages = loader.load()

//----------
// 예-2)
from langchain.document_loaders import WebBaseLoader

loader = WebBaseLoader("https://github.com/langchain-ai/langchain/blob/master/README.md")
docs = loader.load()
print(docs[0].page_content[:500])

API 

API 설명을 살펴보자.

- langchain_community 패키지밑으로 구현체가 존재하기에 필요한 경우 다양한 Custom Loader를 만들 수 있다. 

- 추상 클래스 (Abstract Class)은 langchain_core 추상클래스인 BaseLoader를 상속받아 구현한다. 

class BaseLoader(ABC):
  def load(self) -> List[Document]:
      """Load data into Document objects."""
      return list(self.lazy_load())
  def lazy_load(self) -> Iterator[Document]:
      """Custom Loader에서 구현해야 함."""

- BaseLoader <- BasePDFLoader <- PyPDFLoader 를 구현한 클래스를 보면 lazy_load 메소드를 구현하고 있다. 

class PyPDFLoader(BasePDFLoader):
    def lazy_load(
        self,
    ) -> Iterator[Document]:
        """Lazy load given path as pages."""
        if self.web_path:
            blob = Blob.from_data(open(self.file_path, "rb").read(), path=self.web_path)  # type: ignore[attr-defined]
        else:
            blob = Blob.from_path(self.file_path)

 - 반환값은 langchain_core의 Document 리스트이고, BaseMedia <- Document 를 통해 항시 기본적으로 접근 가능한 애트리뷰트가 있다. 

    - metadata : 문서 속성 정보

    - page_content : 문서의 내용

class BaseMedia(Serializable):
  id: Optional[str] = None
  metadata: dict = Field(default_factory=dict)
  
class Document(BaseMedia):
  page_content: str
  type: Literal["Document"] = "Document"

- GenericLoader 는 다양한 문서를 로드할 수 있다. API 설명의 예제를 참조한다.

Custom Document Loader 만들기

- LangChain custom loader 가이드를 우선 참조하자.

- 구현 순서

   - BaseLoader를 상속받는다.

   - lazy_load 메서드를 구현한다.

   - alazy_load는 옵션으로 lazy_load로 위임해도 된다. 

- HWPLoader를 구현한 TeddyNote 소스를 참조하자.

<참조>

API: https://api.python.langchain.com/en/latest/langchain_api_reference.html

Source: https://github.com/langchain-ai/langchain/tree/master

Guide: https://python.langchain.com/v0.2/docs/concepts/

HWPLoader: https://github.com/teddylee777/langchain-teddynote/blob/main/langchain_teddynote/document_loaders/hwp.py

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

LangChain은 LLM 애플리케이션을 구축하기 위한 개발 프레임워크로써 다양한 컴포넌트를 제공한다. 프러덕션 애플리케이션 개발시 RAG(Retrieval Augmented Generation)를 기반으로 할 때 LangChain 컴포넌트를 통해 일관된 코드 인터페이스를 유지할 수 있다.

LangChain Components

Prompts

• Prompt Templates
• Output Parsers: 5+ implementations
  • Retry/fixing logic
• Example Selectors: 5+ implementations

Models

• LLM's: 20+ integrations
• Chat Models
• Text Embedding Models: 10+ integrations

Indexes

• Document Loaders: 50+ implementations
• Text Splitters: 10+ implementations
• Vector stores: 10+ integrations
• Retrievers: 5+ integrations/implementations

Chains

• Can be used as building blocks for other chains
• More application specific cahins: 20+ different types

Agents

• Agent Types: 5+ types
  • Algorithms for getting LLMs to use tools
• Agent Tookkits: 10+ implementations
  • Agents armed with specific tools for a specific application

LangChain API

각 컴포넌트는 구현된 패키지와 API랑 함께 볼 필요가 있다.

• API: LangChain API
• 저장소: LangChain Repository

langchain

• document_loaders
• text_splitter
• embeddings
• retrievers
• prompts
• chat_models
• output_parsers
• chains
• agents

등의 패키지에서 langchain_community, langchain_core 패키지의 모듈을 re-export 하고 있다. 중요 패키지는 langchain.* 으로 import 할 수 있다.

• langchain 소스

• text_splitter.py : text-splitters를 re-export하고 있다.
• text_splitter 별도 패키지

langchain-core

• core 패키지 소스
  

  

langchain-community

• community 패키지 소스
  

  

RAG Step

RAG는 크게 2단계로 볼 수 있다. 사전에 프라이빗 정보를 로딩->쪼개기->임베딩벡터->저장하기 단계를 거쳐서 준비를 한다. 이때 LangChain의 Indexes 영역의 컴포넌트를 사용한다. 다음으로 사용자가 질의를 하게되면 프라이빗 정보를 기반으로 증강검색->프롬프트생성->LLM응답->응답처리 등의 과정을 거쳐, 유의미한 응답을 생성한다. 이때 LangChain의 Prompts, Models, Chains, Agents 영역의 컴포넌트를 사용한다.

1 단계

• Document Loader
  • Structured/Unstructured 미디어를 로딩한다.
    • 문서, 이미지, 동영상, 음성
• Splitting
• Embedding
• Vector Storing

2단계

• Retrieval
• Prompting
• LLM
• Output Parsing

"LangChain 은 LLM기반  AI 애플리케이션 개발에 있어서의  Spring Framework이다."

Java로 웹개발을 한다면 기본 Framework으로 개발자들이 Spring Framework을 사용하듯이 LLM 기반 AI 애플리케이션 개발은 LangChain 프레임워크로 시작하게 될 것이다. 따라서 LLM 기반 AI 애플리케이션을 개발하고 싶다면 기본 LangChain Framework을 사용하게되는 시즌이 시작되었다는 뜻이다. 

Python vs Javascript(Typescripit)

LangChain, LangGraph, LangSmith등 서비스를 사용하기 위하여 Python 또는 Javascript 중 하나를 선택해서 시작할 수 있다. 

  - LangGraph Cloud는 아직 Python만 지원한다.

  - LangSmith는 상업적 이용시 Commercial 이다.  

Javascript를 해보았으니, Python으로 시작해 본다. 

ChatGPT 설명

LangChain Package Structure

LangChain은 언어 모델을 사용한 애플리케이션 개발을 위한 프레임워크로, 다양한 패키지와 모듈로 구성되어 있습니다. 아래는 LangChain의 일반적인 패키지 구조입니다:

• LangChain: 최상위 패키지

• Chains: 체인 및 워크플로우 관리

• LLMChain: 대형 언어 모델 체인 클래스

• VectorDBQAChain: 벡터 데이터베이스 기반 QA 체인 클래스

• SQLDBQAChain: SQL 데이터베이스 기반 QA 체인 클래스

• Prompts: 프롬프트 관리 및 생성

• PromptTemplate: 프롬프트 템플릿 클래스

• FewShotPromptTemplate: 몇 샷 학습 프롬프트 템플릿 클래스

• Agents: 에이전트 및 도구 통합

• AgentExecutor: 에이전트 실행기 클래스

• Tool: 도구 클래스

• Memory: 메모리 관리

• BufferMemory: 버퍼 메모리 클래스

• VectorStoreRetrieverMemory: 벡터 저장소 리트리버 메모리 클래스

• LLMs: 대형 언어 모델 통합

• OpenAI: OpenAI 모델 통합 클래스

• Cohere: Cohere 모델 통합 클래스

• Utilities: 유틸리티 도구

• Wikipedia: Wikipedia 통합 유틸리티

• PythonREPL: Python REPL 유틸리티

이 구조는 LangChain의 주요 구성 요소와 각 구성 요소가 제공하는 기능을 잘 나타내고 있습니다. LangChain은 체인 관리, 프롬프트 생성, 에이전트 통합, 메모리 관리, 대형 언어 모델 통합, 그리고 다양한 유틸리티 도구를 통해 언어 모델 기반 애플리케이션 개발을 지원합니다.

LangGraph Package Structure

LangGraph는 언어 모델을 활용한 그래프 기반 애플리케이션을 구축하기 위한 프레임워크입니다. 아래는 LangGraph의 일반적인 패키지 구조입니다

• LangGraph: 최상위 패키지

• Graphs: 그래프 생성 및 관리

• GraphBuilder: 그래프 빌더 클래스

• Node: 그래프의 노드 클래스

• Edge: 그래프의 엣지 클래스

• Algorithms: 그래프 알고리즘

• PathFinding: 경로 찾기 알고리즘 (예: Dijkstra, A*)

• Centrality: 중심성 측정 알고리즘 (예: Betweenness, Closeness)

• Clustering: 그래프 클러스터링 알고리즘

• Visualization: 그래프 시각화 도구

• GraphVisualizer: 그래프 시각화 클래스

• PlotSettings: 시각화 설정 클래스

• Data: 데이터 처리 및 로딩

• DataLoader: 데이터 로딩 클래스

• DataProcessor: 데이터 처리 클래스

• Models: 언어 모델 통합

• ModelInterface: 언어 모델 인터페이스

• LLMIntegration: 대형 언어 모델 통합 클래스 (예: GPT-4, BERT)

• Utilities: 유틸리티 도구

• Logger: 로깅 유틸리티

• ConfigManager: 설정 관리 유틸리티

이 패키지 구조는 LangGraph의 주요 구성 요소와 각 구성 요소가 제공하는 기능을 잘 나타내고 있습니다. LangGraph는 그래프 기반의 데이터 구조를 생성, 관리, 분석 및 시각화하는 데 필요한 다양한 도구를 제공하며, 언어 모델과의 통합을 통해 더욱 강력한 기능을 지원합니다.

---

LangChain과 Spring Framework는 각각의 도메인에서 비슷한 역할을 수행합니다, 단지 다른 맥락에서 사용된다는 차이가 있습니다:

LangChain:

• 도메인: 언어 모델 및 자연어 처리.

• 목적: 언어 모델(예: GPT-3, GPT-4 등)로 구동되는 애플리케이션을 구축하기 위한 프레임워크를 제공합니다. LangChain은 언어 모델의 통합, 배포 및 오케스트레이션을 간소화하는 것을 목표로 합니다.

• 특징:

• 체인 관리: 여러 언어 모델과 작업을 관리하고 오케스트레이션하는 데 도움을 줍니다.

• 모듈성: 언어 모델 애플리케이션을 구축, 미세 조정 및 배포하는 모듈식 접근 방식을 제공합니다.

• 확장성: 기능을 향상시키기 위해 다양한 API 및 외부 도구와 쉽게 통합할 수 있습니다.

• 유틸리티 도구: 프롬프트 엔지니어링, 대화 흐름 관리 등을 위한 유틸리티를 제공합니다.

Spring Framework:

• 도메인: 엔터프라이즈 자바 애플리케이션.

• 목적: 자바 기반의 엔터프라이즈 애플리케이션을 개발하기 위한 포괄적인 프레임워크입니다. Spring은 인프라 지원을 제공하여 애플리케이션 개발을 단순화합니다.

• 특징:

• 의존성 주입: 객체 생성과 의존성을 유연하고 느슨하게 관리합니다.

• 관점 지향 프로그래밍(AOP): 횡단 관심사(예: 로깅, 보안 등)를 분리할 수 있습니다.

• 데이터 접근: 데이터베이스 상호작용과 트랜잭션 관리를 위한 템플릿을 제공합니다.

• 웹 프레임워크: 웹 애플리케이션, RESTful 서비스 등을 구축하기 위한 모듈을 포함합니다.

• 보안: 애플리케이션을 보호하기 위한 강력한 보안 기능을 제공합니다.

유사점:

1. 프레임워크 목적: 각 도메인에서 애플리케이션 구축에 구조적인 접근 방식을 제공하여 보일러플레이트 코드를 줄이고 생산성을 높이는 것을 목표로 합니다.

2. 모듈성: 모듈성 및 확장성을 강조하여 개발자가 필요에 따라 다양한 구성 요소를 플러그인할 수 있습니다.

3. 통합: 다른 도구 및 기술과의 광범위한 통합을 지원하여 원활한 개발 워크플로우를 가능하게 합니다.

4. 커뮤니티와 생태계: 강력한 커뮤니티 지원과 풍부한 확장 및 플러그인 생태계를 가지고 있습니다.

차이점:

1. 도메인: LangChain은 언어 모델 애플리케이션에 특화되어 있고, Spring은 자바 엔터프라이즈 애플리케이션을 위한 일반 목적의 프레임워크입니다.

2. 언어 및 플랫폼: LangChain은 일반적으로 파이썬과 언어 모델 API를 사용하고, Spring은 자바 및 JVM 기반 애플리케이션과 함께 사용됩니다.

3. 범위: LangChain은 대형 언어 모델의 사용을 간소화하는 데 중점을 두고 있으며, Spring은 데이터 접근, 보안, 웹 개발 등 엔터프라이즈 애플리케이션 개발의 다양한 측면을 위한 도구를 제공합니다.

요약하면, LangChain과 Spring Framework는 다른 기술적 맥락에서 작동하지만, 각각의 도메인에서 애플리케이션 개발을 단순화하고 구조화하는 공통 목표를 공유합니다.

References

https://python.langchain.com/v0.2/docs/introduction/

https://js.langchain.com/v0.2/docs/introduction/

"LangSmith는 LLM 애플리케이션 개발, 모니터링 및 테스트를 위한 플랫폼이다. "

LangSmith 추적 기능

"추적은 LLM 애플리케이션의 동작을 이해하기 위한 강력한 도구이다.". Java, Javascript 의 stackoverflow같은 단계별 call 추적 가능.

- 예상치 못한 최종 결과

- 에이전트가 루팅되는 이유

- 체닝이 예상보다 느린 이유

- 에이전트가 각 단계에서 사용한 토큰 수

LangSmith 추적 사용

- https://smith.langchain.com/ 가입

- API Key를 발급 받는다. 참조

- poetry를 사용하여 "테디노트"의 패키지를 사용한다. 

    - poetry 설정 및 사용 참조

[1] langchain-teddynote 패키지 설치

poetry add langchain-teddynote

[2] 테스트 코드에 항상 입력

from langchain_teddynote import logging

# 프로젝트 이름을 입력합니다.
logging.langsmith("CH01-Basic")

[3] ipynb 에서 테스트 코드 수행하면 LangSmith 사이트에 자동으로 로깅됨

  - 본인 계정으로 로그인

  - 프로젝트명을 선택

  - RunnableSequence는 테디노트의 https://wikidocs.net/233344 LCEL 파트의 "prompt | model | output_parser" 에 대한 Runnable 내역이 각각 기록됨. 

LangSmith는 Application Performacne Monitoring 서비스중 하나인 DataDog같은 개념으로 LLM 의 사용 성능 및 과금내역등 그외 많은 기능을 제공하고 있다. 

- https://www.datadoghq.com/

다양한 LLM Trace 도구들

LLM 요청에 대한 추적, Prompt 관리, DataSet 관리등의 다양한 기능을 제공한다. 물론 LangSmith도 같은 기능을 제공한다. 

- LangFuse: https://langfuse.com/

- Pheonix: https://docs.arize.com/phoenix

References

https://wikidocs.net/250954

https://mobicon.tistory.com/600

https://github.com/teddylee777/langchain-teddynote