문제 상황
여러분은 langchain의 prompt template들을 통해 llm chain을 구성해 활용하신 적이 있으신가요? 일반적인 프롬프트 상황은 아래와 같을 것입니다.
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from icecream import ic
prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"Answer below question for user:\n{question}"
)
]
)
llm = ChatOpenAI()
chain = prompt | llm
ic(chain.invoke(input={"question": "인공지능은 무엇입니까?"})) 위 예시처럼, ChatPromptTemplate.from_messages(...) 인자인 message들은 (chat_type, template)으로 구성됩니다.
이후 chain.invoke(...) 부분에서 input dict에 message 내 template 변수 명을 key로 주어 value를 넣을 수 있게 됩니다.
따라서 기존 chatgpt action 프로젝트에서 사용하던 prompt들을 간단하게 옮겨 사용할 수 있을 것이라 생각했고 싱글벙글 코드를 작성하게 되었습니다.
하.지.만
문제는 생각지도 못한 부분에서 발생하였습니다. 제가 작성한 chatgpt action용 프롬프트에는 복잡한 task가 많았는데요. 특히 그중에는 사용자 입력을 기반으로 elasticsearch query에 사용될 filter dict를 생성하는 task가 있었습니다.
따라서 프롬프트에 원하는 포맷, 추가 지시문을 작성하기 위해 불가피하게 아래 내용이 들어갔습니다.
{
"filter": ...
}이러다보니, ChatPromptTemplate에서는 위 내용 중 "filter"를 template 변수라고 인식을 했고 위 프롬프트를 ChatPromptTemplate 내에서 사용할 수 없게 되었습니다 (이번 경우는 프롬프트를 다른 식으로 변경을 해 수정할 수 없던 이유가 filter dict의 예문이 들어가야 올바른 elasticsearch query filter가 생성되었기 때문입니다..).
해결 법
현재 langchain의 미리 구현된 chain로직(prompt->llm)으로는 제가 원하는 프롬프트 내용을 담을 수 없고 이를 해결하기 위해 저는 openai 라이브러리의 OpenAI(...)를 사용해 답을 생성하는 사용자 함수를 만들고 이를 chain으로 만들어 해결하였습니다. 중요한 부분은 langchain의 @chain과 RunnableLambda입니다.
간단히 설명하겠지만, chain.invoke(...)는 langchain에서 매우 중요한 메서드이며, 이는 langchain_core.runnables의 Runnable 클래스를 상속한 경우에 사용할 수 있게 됩니다. 그렇다면 사용자가 임의로 생성한 코드는 invoke를 무조건 사용할 수 없는 것일까요? 당연히도 langchain은 사용자 함수 또한 invoke를 사용할 수 있도록 조치해 두었습니다.
langchain_core.runnables.RunnableLambda
langchain_core.runnables.RunnableLambda(통칭 RunnableLambda)는 아래 코드와 같습니다(langchain==0.1.16).
class RunnableLambda(Runnable[Input, Output]):
"""RunnableLambda converts a python callable into a Runnable.
Wrapping a callable in a RunnableLambda makes the callable usable
within either a sync or async context.
RunnableLambda can be composed as any other Runnable and provides
seamless integration with LangChain tracing.
Examples:
.. code-block:: python
# This is a 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
"""
def __init__(
...위 내용에서 확인할 수 있듯, RunnableLambda는 Runnable 클래스를 상속받고 example 내용처럼 사용자 함수로 invoke, ainvoke, ... 등을 사용할 수 있게 만듭니다.
langchain_core.runnables.chain
위에서 말씀드린 해결법에서 사용자 함수를 chain으로 만든다고 했습니다. 이와 사용자 함수를 Runnable로 만드는 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)위 코드는 langchain의 @chain 코드입니다(langchain==0.1.16). Callable인 func를 입력받아 RunnableLambda로 감싸 리턴하는 것을 확인할 수 있습니다.
해결 코드
import typing as t
from icecream import ic
from langchain_core.runnables import chain
from openai import OpenAI
@chain
def gen_es_query_filter(user_input:str)->t.Dict:
"""
Generate elasticsearch query filter.
"""
client = OpenAI()
es_filter_gen_result = client.chat.completions.create(
model="gpt-4o",
temperature=0.0,
top_p=0.75,
messages=[{"role": "system", "content": METADATA_FILTER_PROMPTS[0]}, {"role": "user", "content": str(METADATA_FILTER_PROMPTS[1] % user_input)}],
)
es_filters = eval(es_filter_gen_result.choices[0].message.content)["filter"]
ic(es_filters)
return es_filters
if __name__ == "__main__":
gen_es_query_filter.invoke("2023년도 출원 특허 조회")(아, 혹시 제가 정리해둔 내용이 이미 langchain에서 지원하고 있거나 더 좋은 방법이 있으시면 공유 부탁드리겠습니다 ㅠㅠ)
