Langchain 깊게 파보기
Langchain 의 내부 동작 깊게 파보기
https://github.com/langchain-ai/langchain
GitHub - langchain-ai/langchain: 🦜🔗 Build context-aware reasoning applications
🦜🔗 Build context-aware reasoning applications. Contribute to langchain-ai/langchain development by creating an account on GitHub.
github.com
Langchain 내부는 어떻게 구현되어있을까?
본 궁금증은 Langchain 에 Gemini 를 이용하여 RAG 를 테스트하던 중,
RAG 를 사용하거나 안 하거나 받은 LLM의 답변이 거의 비슷한 것을 보고 생겨났다.
(OpenAI 의 Chat GPT 3.5 는 RAG 를 활용하여 정상적으로 응답했다.)
=> 아래는 내가 다시 보기 위해 탐색한 과정을 기록한 것이며,
결론을 정리하면 Langchain 에서 아래와 같은 형태로 프롬프트를 전달하는데,
System: Use the following pieces of context to answer the user's question.
If you don't know the answer, just say that you don't know, don't try to make up an answer.
----------------
{context - 문서에서 검색한 RAG 결과}
Human: {question - 입력한 질문}
Gemini 는 시스템 메세지를 지원하지 않는다. 동일한 내용을 휴먼 메세지로 보내면 정상적으로 동작한다.
https://python.langchain.com/docs/integrations/chat/google_generative_ai/#gemini-prompting-faqs
+ Agent 또한 동일하다. 시스템 메세지를 지원하지 않는다. 휴먼 메세지로 들어가도록 코드를 작성해야한다.
0. 파이썬의 상속
코드를 분석하기 전에 알아야할 것은 파이썬에서 Class 상속 방법이다.
class 자식클래스(부모클래스) 이렇게 괄호 안에 상속할 부모 클래스 이름을 적으면 부모의 속성과 메소드(함수)를 자식 클래스에서 그대로 상속받아 사용할 수 있다.
class 부모클래스:
...내용...
class 자식클래스(부모클래스):
...내용...43. class 정리 - 상속(inheritance)
## 1. 상속(inheritance) 이란? - 클래스에서 상속이란, 물려주는 클래스(Parent Class, Super class)의 내용(속성과 메소드)을 물려받는 클래스…
wikidocs.net
1. RAG 동작 방법
랭체인 완벽 입문 (위키북스) 에서 공개해준 소스코드를 보면 RAG 를 사용하는 코드는 다음과 같다.
https://github.com/wikibook/langchain
참고) langchain/03_retrieval/query_3.py
from langchain.chains import RetrievalQA #← RetrievalQA를 가져오기
from langchain.chat_models import ChatOpenAI
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
chat = ChatOpenAI(model="gpt-3.5-turbo")
embeddings = OpenAIEmbeddings(
model="text-embedding-ada-002"
)
database = Chroma(
persist_directory="./.data",
embedding_function=embeddings
)
retriever = database.as_retriever() #← 데이터베이스를 Retriever로 변환
qa = RetrievalQA.from_llm( #← RetrievalQA를 초기화
llm=chat, #← Chat models를 지정
retriever=retriever, #← Retriever를 지정
return_source_documents=True #← 응답에 원본 문서를 포함할지를 지정
)
result = qa("비행 자동차의 최고 속도를 알려주세요")
print(result["result"]) #← 응답을 표시
print(result["source_documents"]) #← 원본 문서를 표시
RetrievalQA.from_llm() 를 호출하여 나온 결과를 qa 에 저장하고,
qa("원하는 질문") 형태로 LLM 에 요청하는 것을 볼 수 있다.
Langchain 소스코드의 RetrievalQA 를 따라가보면 다음과 같다. (한 줄씩만 남기고 코드를 생략했다.) BaseRetrievalQA 를 상속받았고, 별도로 정의한 메소드는 _get_docs, _aget_docs, _chain_type 밖에 없다.
참고) libs/langchain/langchain/chains/retrieval_qa/base.py
@deprecated(since="0.1.17", alternative="create_retrieval_chain", removal="0.3.0")
class RetrievalQA(BaseRetrievalQA):
retriever: BaseRetriever = Field(exclude=True)
def _get_docs(
async def _aget_docs(
@property
def _chain_type(self) -> str:
부모 클래스인 BaseRetrievalQA 를 따라가보면 다음과 같다. 처음에 봤던 RAG 예제코드에서 호출했던 .from_llm 메소드를 가지고 있다!
참고) libs/langchain/langchain/chains/retrieval_qa/base.py (위와 같은 파일)
class BaseRetrievalQA(Chain):
combine_documents_chain: BaseCombineDocumentsChain
class Config:
@property
def input_keys(self) -> List[str]:
@property
def output_keys(self) -> List[str]:
@classmethod
def from_llm(
cls,
llm: BaseLanguageModel,
prompt: Optional[PromptTemplate] = None,
callbacks: Callbacks = None,
llm_chain_kwargs: Optional[dict] = None,
**kwargs: Any,
) -> BaseRetrievalQA:
@classmethod
def from_chain_type(
@abstractmethod
def _get_docs(
def _call(
@abstractmethod
async def _aget_docs(
async def _acall(
from_llm 만 자세히 보면 다음과 같다. from_llm(cls, ~) 형태인데, 그 cls 를 return 하고 있다. cls 란 일반 메소드가 self 를 써서 자기 자신을 가리키듯이 class 를 가리키는 것이다. https://firework-ham.tistory.com/101 이를 이용하면 인스턴스를 만들지 않아도 클래스의 멤버 변수에 접근할 수 있다. def from_llm() -> BaseRetrievalQA 로 친절하게 표시되어있으므로 return cls() 했을 때 클래스는 BaseRetrievalQA 임을 알 수 있다.
@classmethod
def from_llm(
cls,
llm: BaseLanguageModel,
prompt: Optional[PromptTemplate] = None,
callbacks: Callbacks = None,
llm_chain_kwargs: Optional[dict] = None,
**kwargs: Any,
) -> BaseRetrievalQA:
"""Initialize from LLM."""
_prompt = prompt or PROMPT_SELECTOR.get_prompt(llm)
llm_chain = LLMChain(
llm=llm, prompt=_prompt, callbacks=callbacks, **(llm_chain_kwargs or {})
)
document_prompt = PromptTemplate(
input_variables=["page_content"], template="Context:\n{page_content}"
)
combine_documents_chain = StuffDocumentsChain(
llm_chain=llm_chain,
document_variable_name="context",
document_prompt=document_prompt,
callbacks=callbacks,
)
return cls(
combine_documents_chain=combine_documents_chain,
callbacks=callbacks,
**kwargs,
)
예제 코드에서는 이 cls() 를 qa 로 저장했고, qa("원하는 질문") 형태로 사용했으므로, BaseRetrievalQA 에서 이 부분을 찾아보면 된다. 호출 형태로 사용할 때는 내부에 __call__ 이라는 메소드가 호출되어야한다. https://wjunsea.tistory.com/61 그런데 BaseRetrievalQA 에는 __call__ 이 없기 때문에 부모 클래스인 Chain 을 찾아보았다.
class BaseRetrievalQA(Chain):
"""Base class for question-answering chains."""
빙고. Chain 에는 __call__ 이 있었다. (참고로 앞으로는 직접 호출대신 .invoke 를 사용할 것을 장려하고 있었다.)
참고) langchain/libs/langchain/langchain/chains/base.py
class Chain(RunnableSerializable[Dict[str, Any], Dict[str, Any]], ABC):
memory: Optional[BaseMemory] = None
callbacks: Callbacks = Field(default=None, exclude=True)
verbose: bool = Field(default_factory=_get_verbosity)
tags: Optional[List[str]] = None
metadata: Optional[Dict[str, Any]] = None
callback_manager: Optional[BaseCallbackManager] = Field(default=None, exclude=True)
class Config:
def get_input_schema(
def get_output_schema(
def invoke(
async def ainvoke(
@property
def _chain_type(self) -> str:
@root_validator()
def raise_callback_manager_deprecation(cls, values: Dict) -> Dict:
@validator("verbose", pre=True, always=True)
def set_verbose(cls, verbose: Optional[bool]) -> bool:
@property
@abstractmethod
def input_keys(self) -> List[str]:
@property
@abstractmethod
def output_keys(self) -> List[str]:
def _validate_inputs(self, inputs: Dict[str, Any]) -> None:
def _validate_outputs(self, outputs: Dict[str, Any]) -> None:
@abstractmethod
def _call(
async def _acall(
@deprecated("0.1.0", alternative="invoke", removal="0.3.0")
def __call__(
@deprecated("0.1.0", alternative="ainvoke", removal="0.3.0")
async def acall(
def prep_outputs(
### skip
__call__ 내부에서는 invoke 를 호출하고 있었다. (return self.invoke())
@deprecated("0.1.0", alternative="invoke", removal="0.3.0")
def __call__(
self,
inputs: Union[Dict[str, Any], Any],
return_only_outputs: bool = False,
callbacks: Callbacks = None,
*,
tags: Optional[List[str]] = None,
metadata: Optional[Dict[str, Any]] = None,
run_name: Optional[str] = None,
include_run_info: bool = False,
) -> Dict[str, Any]:
config = {
"callbacks": callbacks,
"tags": tags,
"metadata": metadata,
"run_name": run_name,
}
return self.invoke(
inputs,
cast(RunnableConfig, {k: v for k, v in config.items() if v is not None}),
return_only_outputs=return_only_outputs,
include_run_info=include_run_info,
)
invoke 메소드는 다음과 같다.
def invoke(
self,
input: Dict[str, Any],
config: Optional[RunnableConfig] = None,
**kwargs: Any,
) -> Dict[str, Any]:
config = ensure_config(config)
callbacks = config.get("callbacks")
tags = config.get("tags")
metadata = config.get("metadata")
run_name = config.get("run_name") or self.get_name()
include_run_info = kwargs.get("include_run_info", False)
return_only_outputs = kwargs.get("return_only_outputs", False)
inputs = self.prep_inputs(input)
callback_manager = CallbackManager.configure(
callbacks,
self.callbacks,
self.verbose,
tags,
self.tags,
metadata,
self.metadata,
)
new_arg_supported = inspect.signature(self._call).parameters.get("run_manager")
run_manager = callback_manager.on_chain_start(
dumpd(self),
inputs,
name=run_name,
)
try:
self._validate_inputs(inputs)
outputs = (
self._call(inputs, run_manager=run_manager)
if new_arg_supported
else self._call(inputs)
)
final_outputs: Dict[str, Any] = self.prep_outputs(
inputs, outputs, return_only_outputs
)
except BaseException as e:
run_manager.on_chain_error(e)
raise e
run_manager.on_chain_end(outputs)
if include_run_info:
final_outputs[RUN_KEY] = RunInfo(run_id=run_manager.run_id)
return final_outputstry: 부분을 보면 _validate_inputs 로 인풋을 확인하고, self._call 로 호출을 한 후에 self.prep_outputs 로 리턴할 결과를 가져온다. 그렇다면 중요한 호출 부분은 self._call 일 것이다. _call은 이제 Chain 의 자식 클래스인 BaseRetrievalQA 의 _call 메소드를 보면 된다. (__call__ 이나 invoke 는 자식에게 없었기 때문에 부모 클래스인 Chain 의 메소드를 본 것이었고, _call 은 Chain 에게는 이름만 있고 내부가 빈 껍데기이다. 자식 클래스인 BaseRetrievalQA 에 가야 실제 내부가 구현되어있다.)
BaseRetrievalQA 의 _call 메소드
def _call(
self,
inputs: Dict[str, Any],
run_manager: Optional[CallbackManagerForChainRun] = None,
) -> Dict[str, Any]:
"""Run get_relevant_text and llm on input query.
If chain has 'return_source_documents' as 'True', returns
the retrieved documents as well under the key 'source_documents'.
Example:
.. code-block:: python
res = indexqa({'query': 'This is my query'})
answer, docs = res['result'], res['source_documents']
"""
_run_manager = run_manager or CallbackManagerForChainRun.get_noop_manager()
question = inputs[self.input_key]
accepts_run_manager = (
"run_manager" in inspect.signature(self._get_docs).parameters
)
if accepts_run_manager:
docs = self._get_docs(question, run_manager=_run_manager)
else:
docs = self._get_docs(question) # type: ignore[call-arg]
answer = self.combine_documents_chain.run(
input_documents=docs, question=question, callbacks=_run_manager.get_child()
)
if self.return_source_documents:
return {self.output_key: answer, "source_documents": docs}
else:
return {self.output_key: answer}answer = self.combine_documents_chain.run() 부분에서 처리하고 있는 모습이 보인다. combine_documents_chain 은 BaseRetrievalQA 가 처음에는 멤버변수로 BaseCombineDocumentsChain 를 가지고 있었다.
class BaseRetrievalQA(Chain):
"""Base class for question-answering chains."""
combine_documents_chain: BaseCombineDocumentsChain
그러나, from_llm 을 호출할 때 StuffDocumentsChain 으로 바뀐다.
@classmethod
def from_llm(
### skip
) -> BaseRetrievalQA:
_prompt = prompt or PROMPT_SELECTOR.get_prompt(llm)
llm_chain = LLMChain(
llm=llm, prompt=_prompt, callbacks=callbacks, **(llm_chain_kwargs or {})
)
document_prompt = PromptTemplate(
input_variables=["page_content"], template="Context:\n{page_content}"
)
combine_documents_chain = StuffDocumentsChain(
llm_chain=llm_chain,
document_variable_name="context",
document_prompt=document_prompt,
callbacks=callbacks,
)
return cls(
combine_documents_chain=combine_documents_chain,
callbacks=callbacks,
**kwargs,
)
그러면 이제 StuffDocumentsChain 의 run 메소드를 찾으면 될 것이다. run 메소드는 없다... 몇 번 따라가다보니 단련되었다. 부모 클래스인 BaseCombineDocumentsChain 에 run 메소드를 찾으러 가본다.
참고) langchain/libs/langchain/langchain/chains/combine_documents/stuff.py
class StuffDocumentsChain(BaseCombineDocumentsChain):
llm_chain: LLMChain
document_prompt: BasePromptTemplate = Field(
default_factory=lambda: DEFAULT_DOCUMENT_PROMPT
)
document_variable_name: str
document_separator: str = "\n\n"
class Config:
@root_validator(pre=True)
def get_default_document_variable_name(cls, values: Dict) -> Dict:
@property
def input_keys(self) -> List[str]:
def _get_inputs(self, docs: List[Document], **kwargs: Any) -> dict:
def prompt_length(self, docs: List[Document], **kwargs: Any) -> Optional[int]:
def combine_docs(
async def acombine_docs(
@property
def _chain_type(self) -> str:
BaseCombineDocumentsChain 에도 run 이 없다.
참고) libs/langchain/langchain/chains/combine_documents/base.py
class BaseCombineDocumentsChain(Chain, ABC):
input_key: str = "input_documents" #: :meta private:
output_key: str = "output_text" #: :meta private:
def get_input_schema(
def get_output_schema(
@property
def input_keys(self) -> List[str]:
@property
def output_keys(self) -> List[str]:
def prompt_length(self, docs: List[Document], **kwargs: Any) -> Optional[int]:
@abstractmethod
def combine_docs(self, docs: List[Document], **kwargs: Any) -> Tuple[str, dict]:
@abstractmethod
async def acombine_docs(
def _call(
async def _acall(
부모로 Chain 과 ABC 를 상속받았으니, 또 부모를 찾으러 가야할 것 같다. 참고로 ABC 는 랭체인 소스코드에 만든 것이 아니라 파이썬에서 내장으로 가지고 있는 추상 베이스 클래스이다. https://docs.python.org/ko/3/library/abc.html Chain 의 run 은 다음과 같다. (이것 또한 invoke 사용을 장려하고 있다.)
@deprecated("0.1.0", alternative="invoke", removal="0.3.0")
def run(
self,
*args: Any,
callbacks: Callbacks = None,
tags: Optional[List[str]] = None,
metadata: Optional[Dict[str, Any]] = None,
**kwargs: Any,
) -> Any:
# Run at start to make sure this is possible/defined
_output_key = self._run_output_key
if args and not kwargs:
if len(args) != 1:
raise ValueError("`run` supports only one positional argument.")
return self(args[0], callbacks=callbacks, tags=tags, metadata=metadata)[
_output_key
]
if kwargs and not args:
return self(kwargs, callbacks=callbacks, tags=tags, metadata=metadata)[
_output_key
]
if not kwargs and not args:
raise ValueError(
"`run` supported with either positional arguments or keyword arguments,"
" but none were provided."
)
else:
raise ValueError(
f"`run` supported with either positional arguments or keyword arguments"
f" but not both. Got args: {args} and kwargs: {kwargs}."
)return self() 로 구현되어있는데, 그러면 위에서 따라간 것처럼 __call__ 이 불리고 결국 invoke 가 불릴 것이다.
@deprecated("0.1.0", alternative="invoke", removal="0.3.0")
def __call__(
### skip
) -> Dict[str, Any]:
### skip
return self.invoke(
inputs,
cast(RunnableConfig, {k: v for k, v in config.items() if v is not None}),
return_only_outputs=return_only_outputs,
include_run_info=include_run_info,
) def invoke(
### skip
) -> Dict[str, Any]:
config = ensure_config(config)
callbacks = config.get("callbacks")
tags = config.get("tags")
metadata = config.get("metadata")
run_name = config.get("run_name") or self.get_name()
include_run_info = kwargs.get("include_run_info", False)
return_only_outputs = kwargs.get("return_only_outputs", False)
inputs = self.prep_inputs(input)
callback_manager = CallbackManager.configure(
callbacks,
self.callbacks,
self.verbose,
tags,
self.tags,
metadata,
self.metadata,
)
new_arg_supported = inspect.signature(self._call).parameters.get("run_manager")
run_manager = callback_manager.on_chain_start(
dumpd(self),
inputs,
name=run_name,
)
try:
self._validate_inputs(inputs)
outputs = (
self._call(inputs, run_manager=run_manager)
if new_arg_supported
else self._call(inputs)
)
final_outputs: Dict[str, Any] = self.prep_outputs(
inputs, outputs, return_only_outputs
)
except BaseException as e:
run_manager.on_chain_error(e)
raise e
run_manager.on_chain_end(outputs)
if include_run_info:
final_outputs[RUN_KEY] = RunInfo(run_id=run_manager.run_id)
return final_outputs
invoke 내부는 길어보이지만 결국 self._call 이 불린다. BaseCombineDocumentsChain 의 메소드로 돌아왔다.
def _call(
self,
inputs: Dict[str, List[Document]],
run_manager: Optional[CallbackManagerForChainRun] = None,
) -> Dict[str, str]:
"""Prepare inputs, call combine docs, prepare outputs."""
_run_manager = run_manager or CallbackManagerForChainRun.get_noop_manager()
docs = inputs[self.input_key]
# Other keys are assumed to be needed for LLM prediction
other_keys = {k: v for k, v in inputs.items() if k != self.input_key}
output, extra_return_dict = self.combine_docs(
docs, callbacks=_run_manager.get_child(), **other_keys
)
extra_return_dict[self.output_key] = output
return extra_return_dictself.combine_docs 가 불린다. 이건 원래 보고 있던 StuffDocumentsChain 에게 있다.
def combine_docs(
self, docs: List[Document], callbacks: Callbacks = None, **kwargs: Any
) -> Tuple[str, dict]:
inputs = self._get_inputs(docs, **kwargs)
# Call predict on the LLM.
return self.llm_chain.predict(callbacks=callbacks, **inputs), {}
input 을 인자로 넣고, llm_chain.predict 가 호출된다.
class StuffDocumentsChain(BaseCombineDocumentsChain):
llm_chain: LLMChain
기본이 LLMChain 이기도 하지만, BaseRetrievalQA의 이 함수가 호출될 때 llm_chain 을 인자로 넣어주었었다. 참고로 교재의 예제코드에서 llm 은 ChatOpenAI() 로 넣어줬었다.
@classmethod
def from_llm(
cls,
llm: BaseLanguageModel,
prompt: Optional[PromptTemplate] = None,
callbacks: Callbacks = None,
llm_chain_kwargs: Optional[dict] = None,
**kwargs: Any,
) -> BaseRetrievalQA:
### skip
llm_chain = LLMChain(
llm=llm, prompt=_prompt, callbacks=callbacks, **(llm_chain_kwargs or {})
)
### skip
combine_documents_chain = StuffDocumentsChain(
llm_chain=llm_chain,
document_variable_name="context",
document_prompt=document_prompt,
callbacks=callbacks,
)
### skip
그러면 이제 llm_chain = LLMChain 이었으니,
LLMChain 이 가진 predict 를 찾으러 가본다.
참고) langchain/libs/langchain/langchain/chains/llm.py
class LLMChain(Chain):
@classmethod
def is_lc_serializable(self) -> bool:
prompt: BasePromptTemplate
llm: Union[
Runnable[LanguageModelInput, str], Runnable[LanguageModelInput, BaseMessage]
]
output_key: str = "text" #: :meta private:
output_parser: BaseLLMOutputParser = Field(default_factory=StrOutputParser)
return_final_only: bool = True
llm_kwargs: dict = Field(default_factory=dict)
class Config:
@property
def input_keys(self) -> List[str]:
@property
def output_keys(self) -> List[str]:
def _call(
def generate(
async def agenerate(
def prep_prompts(
async def aprep_prompts(
def apply(
async def aapply(
@property
def _run_output_key(self) -> str:
def create_outputs(self, llm_result: LLMResult) -> List[Dict[str, Any]]:
async def _acall(
def predict(self, callbacks: Callbacks = None, **kwargs: Any) -> str:
async def apredict(self, callbacks: Callbacks = None, **kwargs: Any) -> str:
def predict_and_parse(
async def apredict_and_parse(
def apply_and_parse(
def _parse_generation(
async def aapply_and_parse(
@property
def _chain_type(self) -> str:
@classmethod
def from_string(cls, llm: BaseLanguageModel, template: str) -> LLMChain:
def _get_num_tokens(self, text: str) -> int:
predict 를 가지고 있다.
def predict(self, callbacks: Callbacks = None, **kwargs: Any) -> str:
return self(kwargs, callbacks=callbacks)[self.output_key]
predict 는 self() 이므로 __call__ 을 호출할 것이고, 부모 클래스인 Chain 에서 __call__ 을 호출하면, 결국 invoke 가 불리고 _call 이 불릴 것이다. 그러므로, LLMChain 이 가진 _call 을 본다.
def _call(
self,
inputs: Dict[str, Any],
run_manager: Optional[CallbackManagerForChainRun] = None,
) -> Dict[str, str]:
response = self.generate([inputs], run_manager=run_manager)
return self.create_outputs(response)[0]generate 가 불린다. 이번에는 generate 를 살펴본다.
def generate(
self,
input_list: List[Dict[str, Any]],
run_manager: Optional[CallbackManagerForChainRun] = None,
) -> LLMResult:
prompts, stop = self.prep_prompts(input_list, run_manager=run_manager)
callbacks = run_manager.get_child() if run_manager else None
if isinstance(self.llm, BaseLanguageModel):
return self.llm.generate_prompt(
prompts,
stop,
callbacks=callbacks,
**self.llm_kwargs,
)
else:
results = self.llm.bind(stop=stop, **self.llm_kwargs).batch(
cast(List, prompts), {"callbacks": callbacks}
)
generations: List[List[Generation]] = []
for res in results:
if isinstance(res, BaseMessage):
generations.append([ChatGeneration(message=res)])
else:
generations.append([Generation(text=res)])
return LLMResult(generations=generations)self.llm 이 인스턴스인지 확인하고 있는데, self.llm 은 ChatOpenAI 클래스로 만든 인스턴스이므로 if 문 안에 있는 self.llm.generate_prompt 가 호출될 것이다. ChatOpenAI 의 generate_prompt 대신 원래 궁금했던 Gemini LLM을 살펴보려했다. https://api.python.langchain.com/en/latest/chat_models/langchain_google_genai.chat_models.ChatGoogleGenerativeAI.html 그러나 api 로만 제공되고 내부는 볼 수 없었다.
llm = ChatGoogleGenerativeAI(model = "gemini-pro", temperature=0.0)
어쩔 수 없이 ChatOpenAI 의 generate_prompt 를 쫓아가봤다. 아래 그림과 같이 ChatOpenAI 클래스를 2곳에서 정의하고 있는데, community 용이 맞는 것으로 보인다.
DEPRECATED_LOOKUP = {"ChatOpenAI": "langchain_community.chat_models.openai"}
ChatOpenAI 에는 generate_prompt 메소드가 없다. 부모인 BaseChatModel 을 찾으러 간다.
Breadcrumbslangchain/libs/community/langchain_community/chat_models/openai.py
class ChatOpenAI(BaseChatModel):
@property
def lc_secrets(self) -> Dict[str, str]:
@classmethod
def get_lc_namespace(cls) -> List[str]:
@property
def lc_attributes(self) -> Dict[str, Any]:
@classmethod
def is_lc_serializable(cls) -> bool:
client: Any = Field(default=None, exclude=True) #: :meta private:
async_client: Any = Field(default=None, exclude=True) #: :meta private:
model_name: str = Field(default="gpt-3.5-turbo", alias="model")
temperature: float = 0.7
model_kwargs: Dict[str, Any] = Field(default_factory=dict)
openai_api_key: Optional[str] = Field(default=None, alias="api_key")
openai_api_base: Optional[str] = Field(default=None, alias="base_url")
openai_organization: Optional[str] = Field(default=None, alias="organization")
openai_proxy: Optional[str] = None
request_timeout: Union[float, Tuple[float, float], Any, None] = Field(
default=None, alias="timeout"
)
max_retries: int = 2
streaming: bool = False
n: int = 1
max_tokens: Optional[int] = None
tiktoken_model_name: Optional[str] = None
default_headers: Union[Mapping[str, str], None] = None
default_query: Union[Mapping[str, object], None] = None
http_client: Union[Any, None] = None
class Config:
@root_validator(pre=True)
def build_extra(cls, values: Dict[str, Any]) -> Dict[str, Any]:
@root_validator()
def validate_environment(cls, values: Dict) -> Dict:
@property
def _default_params(self) -> Dict[str, Any]:
def completion_with_retry(
def _combine_llm_outputs(self, llm_outputs: List[Optional[dict]]) -> dict:
def _stream(
def _generate(
def _create_message_dicts(
def _create_chat_result(self, response: Union[dict, BaseModel]) -> ChatResult:
async def _astream(
async def _agenerate(
@property
def _identifying_params(self) -> Dict[str, Any]:
@property
def _client_params(self) -> Dict[str, Any]:
def _get_invocation_params(
@property
def _llm_type(self) -> str:
def _get_encoding_model(self) -> Tuple[str, tiktoken.Encoding]:
def get_token_ids(self, text: str) -> List[int]:
def get_num_tokens_from_messages(self, messages: List[BaseMessage]) -> int:
def bind_functions(BaseChatModel 에는 generate_prompt 가 있다.
Breadcrumbslangchain/libs/core/langchain_core/language_models/chat_models.py
class BaseChatModel(BaseLanguageModel[BaseMessage], ABC):
callback_manager: Optional[BaseCallbackManager] = Field(default=None, exclude=True)
@root_validator()
def raise_deprecation(cls, values: Dict) -> Dict:
class Config:
@property
def OutputType(self) -> Any:
def _convert_input(self, input: LanguageModelInput) -> PromptValue:
def invoke(
async def ainvoke(
def stream(
async def astream(
def _combine_llm_outputs(self, llm_outputs: List[Optional[dict]]) -> dict:
def _get_invocation_params(
def _get_llm_string(self, stop: Optional[List[str]] = None, **kwargs: Any) -> str:
def generate(
async def agenerate(
def generate_prompt(
async def agenerate_prompt(
def _generate_with_cache(
async def _agenerate_with_cache(
@abstractmethod
def _generate(
async def _agenerate(
def _stream(
async def _astream(
@deprecated("0.1.7", alternative="invoke", removal="0.3.0")
def __call__(
async def _call_async(
@deprecated("0.1.7", alternative="invoke", removal="0.3.0")
def call_as_llm(
@deprecated("0.1.7", alternative="invoke", removal="0.3.0")
def predict(
@deprecated("0.1.7", alternative="invoke", removal="0.3.0")
def predict_messages(
@deprecated("0.1.7", alternative="ainvoke", removal="0.3.0")
async def apredict(
@deprecated("0.1.7", alternative="ainvoke", removal="0.3.0")
async def apredict_messages(
@property
@abstractmethod
def _llm_type(self) -> str:
def dict(self, **kwargs: Any) -> Dict:
def bind_tools(
generate_prompt 를 보면 prompts 들을 to_message 로 변형해서 generate 메소드에 넘겨준다.
def generate_prompt(
self,
prompts: List[PromptValue],
stop: Optional[List[str]] = None,
callbacks: Callbacks = None,
**kwargs: Any,
) -> LLMResult:
prompt_messages = [p.to_messages() for p in prompts]
return self.generate(prompt_messages, stop=stop, callbacks=callbacks, **kwargs)
BaseChatModel의 generate 메소드는 callback_manager.on_chat_model_start() 를 호출해서 run manager 를 만들고 _generate_with_cache 에서 호출하는 것으로 보인다.
def generate(
self,
messages: List[List[BaseMessage]],
stop: Optional[List[str]] = None,
callbacks: Callbacks = None,
*,
tags: Optional[List[str]] = None,
metadata: Optional[Dict[str, Any]] = None,
run_name: Optional[str] = None,
run_id: Optional[uuid.UUID] = None,
**kwargs: Any,
) -> LLMResult:
params = self._get_invocation_params(stop=stop, **kwargs)
options = {"stop": stop}
callback_manager = CallbackManager.configure(
callbacks,
self.callbacks,
self.verbose,
tags,
self.tags,
metadata,
self.metadata,
)
run_managers = callback_manager.on_chat_model_start(
dumpd(self),
messages,
invocation_params=params,
options=options,
name=run_name,
run_id=run_id,
batch_size=len(messages),
)
results = []
for i, m in enumerate(messages):
try:
results.append(
self._generate_with_cache(
m,
stop=stop,
run_manager=run_managers[i] if run_managers else None,
**kwargs,
)
)
except BaseException as e:
if run_managers:
run_managers[i].on_llm_error(e, response=LLMResult(generations=[]))
raise e
flattened_outputs = [
LLMResult(generations=[res.generations], llm_output=res.llm_output) # type: ignore[list-item]
for res in results
]
llm_output = self._combine_llm_outputs([res.llm_output for res in results])
generations = [res.generations for res in results]
output = LLMResult(generations=generations, llm_output=llm_output) # type: ignore[arg-type]
if run_managers:
run_infos = []
for manager, flattened_output in zip(run_managers, flattened_outputs):
manager.on_llm_end(flattened_output)
run_infos.append(RunInfo(run_id=manager.run_id))
output.run = run_infos
return output_generate_with_cache 에서는 self._generate가 호출된다.
def _generate_with_cache(
### skip
if type(self)._stream != BaseChatModel._stream and kwargs.pop(
### skip
result = generate_from_stream(iter(chunks))
else:
if inspect.signature(self._generate).parameters.get("run_manager"):
result = self._generate(
messages, stop=stop, run_manager=run_manager, **kwargs
)
else:
result = self._generate(messages, stop=stop, **kwargs)ChatOpenAI 가 _generate 를 가지고 있다. return self.completion_with_retry() 가 호출된다.
def _generate(
self,
messages: List[BaseMessage],
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
stream: Optional[bool] = None,
**kwargs: Any,
) -> ChatResult:
should_stream = stream if stream is not None else self.streaming
if should_stream:
stream_iter = self._stream(
messages, stop=stop, run_manager=run_manager, **kwargs
)
return generate_from_stream(stream_iter)
message_dicts, params = self._create_message_dicts(messages, stop)
params = {
**params,
**({"stream": stream} if stream is not None else {}),
**kwargs,
}
response = self.completion_with_retry(
messages=message_dicts, run_manager=run_manager, **params
)
return self._create_chat_result(response)completion_with_retry 는 self.client.create 를 호출한다.
def completion_with_retry(
self, run_manager: Optional[CallbackManagerForLLMRun] = None, **kwargs: Any
) -> Any:
"""Use tenacity to retry the completion call."""
if is_openai_v1():
return self.client.create(**kwargs)
retry_decorator = _create_retry_decorator(self, run_manager=run_manager)
@retry_decorator
def _completion_with_retry(**kwargs: Any) -> Any:
return self.client.create(**kwargs)
return _completion_with_retry(**kwargs)ChatOpenAI 가 client 를 갖고 있긴 한데, create 가 어떻게 활용되는 건지는 못 찾았다. 아마 확인하지 못한 이전 함수 중에 model_name 을 이용해서 client 를 만드는 부분이 있었을 것 같다.
class ChatOpenAI(BaseChatModel):
### skip
client: Any = Field(default=None, exclude=True) #: :meta private:
async_client: Any = Field(default=None, exclude=True) #: :meta private:
model_name: str = Field(default="gpt-3.5-turbo", alias="model")
"""Model name to use."""
temperature: float = 0.7
"""What sampling temperature to use."""
model_kwargs: Dict[str, Any] = Field(default_factory=dict)
내가 궁금한 것은 prompt 의 마법이므로, 호출하는 함수는 여기까지 따라가보고 다시 prompt 를 찾으러 가봐야겠다.
BaseChatModel이 가진 generate_prompt 부터 살펴본다. p.to_messages 로 프롬프트를 만든다는 건, 이전에 RAG 로 document 를 넘겨줄 때 이미 프롬프트를 위한 인풋들이 만들어졌을 것이다.
def generate_prompt(
self,
prompts: List[PromptValue],
stop: Optional[List[str]] = None,
callbacks: Callbacks = None,
**kwargs: Any,
) -> LLMResult:
prompt_messages = [p.to_messages() for p in prompts]
return self.generate(prompt_messages, stop=stop, callbacks=callbacks, **kwargs)교재의 예시에 따르면 RAG를 위한 문서는 Chroma 를 활용해서 저장한다. VectorStore 를 상속받았고,
class Chroma(VectorStore):
### skip
_LANGCHAIN_DEFAULT_COLLECTION_NAME = "langchain"
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."""
### skip
if client is not None:
self._client_settings = client_settings
self._client = client
self._persist_directory = persist_directoryVectorStore 는 as_retriever 을 가지고 있다. (교재 예시에서 Chroma 의 인스턴스를 만든 다음에 .as_retriever 를 호출)
def as_retriever(self, **kwargs: Any) -> VectorStoreRetriever:
tags = kwargs.pop("tags", None) or []
tags.extend(self._get_retriever_tags())
return VectorStoreRetriever(vectorstore=self, **kwargs, tags=tags)
이렇게 retriever 을 만들고 RetrievalQA.from_llm 에 인자로 넣어주었었다. 그 인자는 return cls() 에 kwargs 로 들어가고, BaseRetrievalQA 의 kwargs 로 들어간다.
@classmethod
def from_llm(
cls,
llm: BaseLanguageModel,
prompt: Optional[PromptTemplate] = None,
callbacks: Callbacks = None,
llm_chain_kwargs: Optional[dict] = None,
**kwargs: Any,
) -> BaseRetrievalQA:
"""Initialize from LLM."""
_prompt = prompt or PROMPT_SELECTOR.get_prompt(llm)
llm_chain = LLMChain(
llm=llm, prompt=_prompt, callbacks=callbacks, **(llm_chain_kwargs or {})
)
document_prompt = PromptTemplate(
input_variables=["page_content"], template="Context:\n{page_content}"
)
combine_documents_chain = StuffDocumentsChain(
llm_chain=llm_chain,
document_variable_name="context",
document_prompt=document_prompt,
callbacks=callbacks,
)
return cls(
combine_documents_chain=combine_documents_chain,
callbacks=callbacks,
**kwargs,
)아마 내부에서 retriever = 멤버 변수에 채워주는 게 있을 것 같은데 못 찾았다.
아까 보던 흐름 중에 중간에 BaseCombineDocumentsChain 의 _call 을 보면, self._get_docs 가 있다.
def _call(
self,
inputs: Dict[str, Any],
run_manager: Optional[CallbackManagerForChainRun] = None,
) -> Dict[str, Any]:
_run_manager = run_manager or CallbackManagerForChainRun.get_noop_manager()
question = inputs[self.input_key]
accepts_run_manager = (
"run_manager" in inspect.signature(self._get_docs).parameters
)
if accepts_run_manager:
docs = self._get_docs(question, run_manager=_run_manager)
else:
docs = self._get_docs(question) # type: ignore[call-arg]
answer = self.combine_documents_chain.run(
input_documents=docs, question=question, callbacks=_run_manager.get_child()
)
if self.return_source_documents:
return {self.output_key: answer, "source_documents": docs}
else:
return {self.output_key: answer}
RetrieverQA가 _get_docs 를 가지고 있다. retriever 의 invoke 를 호출하고 있다. retriever 는 Chroma 를 vectorstore 로 가지고 있는 VectorStoreRetriever 였다.
@deprecated(since="0.1.17", alternative="create_retrieval_chain", removal="0.3.0")
class RetrievalQA(BaseRetrievalQA):
retriever: BaseRetriever = Field(exclude=True)
def _get_docs(
self,
question: str,
*,
run_manager: CallbackManagerForChainRun,
) -> List[Document]:
return self.retriever.invoke(
question, config={"callbacks": run_manager.get_child()}
)
VectorStoreRetriever 는 invoke 가 없어서 부모 클래스인 BaseRetriever 이 가진 invoke 를 찾았다.
class BaseRetriever(RunnableSerializable[RetrieverInput, RetrieverOutput], ABC):
class Config:
_new_arg_supported: bool = False
_expects_other_args: bool = False
tags: Optional[List[str]] = None
metadata: Optional[Dict[str, Any]] = None
### skip
def invoke(
self, input: str, config: Optional[RunnableConfig] = None, **kwargs: Any
) -> List[Document]:
config = ensure_config(config)
return self.get_relevant_documents(
input,
callbacks=config.get("callbacks"),
tags=config.get("tags"),
metadata=config.get("metadata"),
run_name=config.get("run_name"),
**kwargs,
)
내부에서는 get_relevant_documents 가 호출되었다. get_relevant_documents는 내부적으로 _get_relevant_documents 를 호출한다.
@deprecated(since="0.1.46", alternative="invoke", removal="0.3.0")
def get_relevant_documents(
### skip
) -> List[Document]:
from langchain_core.callbacks.manager import CallbackManager
callback_manager = CallbackManager.configure(
callbacks,
None,
verbose=kwargs.get("verbose", False),
inheritable_tags=tags,
local_tags=self.tags,
inheritable_metadata=metadata,
local_metadata=self.metadata,
)
run_manager = callback_manager.on_retriever_start(
dumpd(self),
query,
name=run_name,
run_id=kwargs.pop("run_id", None),
)
try:
_kwargs = kwargs if self._expects_other_args else {}
if self._new_arg_supported:
result = self._get_relevant_documents(
query, run_manager=run_manager, **_kwargs
)
else:
result = self._get_relevant_documents(query, **_kwargs)
except Exception as e:
run_manager.on_retriever_error(e)
raise e
else:
run_manager.on_retriever_end(
result,
)
return result
_get_relevant_documents 는 VectorStoreRetriver 가 가지고 있다.
def _get_relevant_documents(
self, query: str, *, run_manager: CallbackManagerForRetrieverRun
) -> List[Document]:
if self.search_type == "similarity":
docs = self.vectorstore.similarity_search(query, **self.search_kwargs)
elif self.search_type == "similarity_score_threshold":
docs_and_similarities = (
self.vectorstore.similarity_search_with_relevance_scores(
query, **self.search_kwargs
)
)
docs = [doc for doc, _ in docs_and_similarities]
elif self.search_type == "mmr":
docs = self.vectorstore.max_marginal_relevance_search(
query, **self.search_kwargs
)
else:
raise ValueError(f"search_type of {self.search_type} not allowed.")
return docs
vectorstore 인 Chroma 의 similarity_search 를 보면 similarity_search_with_score 가 호출된다.
def similarity_search(
self,
query: str,
k: int = DEFAULT_K,
filter: Optional[Dict[str, str]] = None,
**kwargs: Any,
) -> List[Document]:
docs_and_scores = self.similarity_search_with_score(
query, k, filter=filter, **kwargs
)
return [doc for doc, _ in docs_and_scores]
similarity_search_with_score 에는 __query_collection 을 호출하고 _results_to_docs_and_scores 로 결과를 반환한다.
def similarity_search_with_score(
self,
query: str,
k: int = DEFAULT_K,
filter: Optional[Dict[str, str]] = None,
where_document: Optional[Dict[str, str]] = None,
**kwargs: Any,
) -> List[Tuple[Document, float]]:
if self._embedding_function is None:
results = self.__query_collection(
query_texts=[query],
n_results=k,
where=filter,
where_document=where_document,
**kwargs,
)
else:
query_embedding = self._embedding_function.embed_query(query)
results = self.__query_collection(
query_embeddings=[query_embedding],
n_results=k,
where=filter,
where_document=where_document,
**kwargs,
)
return _results_to_docs_and_scores(results)
__query_collections를 보면 _collection.query 가 호출된다.
@xor_args(("query_texts", "query_embeddings"))
def __query_collection(
self,
query_texts: Optional[List[str]] = None,
query_embeddings: Optional[List[List[float]]] = None,
n_results: int = 4,
where: Optional[Dict[str, str]] = None,
where_document: Optional[Dict[str, str]] = None,
**kwargs: Any,
) -> List[Document]:
try:
import chromadb # noqa: F401
except ImportError:
raise ImportError(
"Could not import chromadb python package. "
"Please install it with `pip install chromadb`."
)
return self._collection.query(
query_texts=query_texts,
query_embeddings=query_embeddings,
n_results=n_results,
where=where,
where_document=where_document,
**kwargs,
)_collection 은 Chroma 가 __init__ 될 때 생긴다.
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
chroma db 에 query 를 해서 가장 유사도가 높은 문서 일부를 추출하는 것 같은데, 문서를 어떻게 넘기고 token 이 어떻게 사용되는지 알아보면 좋을 것 같다. 어쨌든 이 쿼리 결과를 반환받아 사용하게 된다.
docs를 input_documents 에 넣었었으므로, combine_documents_chain.run 을 다시 찾아보면 될 것 같다.
docs = self._get_docs(question) # type: ignore[call-arg]
answer = self.combine_documents_chain.run(
input_documents=docs, question=question, callbacks=_run_manager.get_child()
)
run, _call 등을 거쳐 combine_docs 로 갔었다. StuffDocumentsChain 가 가진 combine_docs인데, 아까 찾아온 docs 를 가지고 self._get_inputs 를 호출해서 inputs 를 만든다.
def combine_docs(
self, docs: List[Document], callbacks: Callbacks = None, **kwargs: Any
) -> Tuple[str, dict]:
inputs = self._get_inputs(docs, **kwargs)
# Call predict on the LLM.
return self.llm_chain.predict(callbacks=callbacks, **inputs), {}_get_inputs 는 document_prompt 를 이용해서 format_document 를 한다.
def _get_inputs(self, docs: List[Document], **kwargs: Any) -> dict:
# Format each document according to the prompt
doc_strings = [format_document(doc, self.document_prompt) for doc in docs]
# Join the documents together to put them in the prompt.
inputs = {
k: v
for k, v in kwargs.items()
if k in self.llm_chain.prompt.input_variables
}
inputs[self.document_variable_name] = self.document_separator.join(doc_strings)
return inputs
StuffDocumentsChain 의 document_prompt 는 아래와 같이 만들어져있다가, from_llm 을 호출할 때 만들어진다.
"""LLM chain which is called with the formatted document string,
along with any other inputs."""
document_prompt: BasePromptTemplate = Field(
default_factory=lambda: DEFAULT_DOCUMENT_PROMPT
)
### 참고
# DEFAULT_DOCUMENT_PROMPT = PromptTemplate.from_template("{page_content}")from_llm의 document_prompt = PromptTramplate 참고
@classmethod
def from_llm(
cls,
llm: BaseLanguageModel,
prompt: Optional[PromptTemplate] = None,
callbacks: Callbacks = None,
llm_chain_kwargs: Optional[dict] = None,
**kwargs: Any,
) -> BaseRetrievalQA:
"""Initialize from LLM."""
_prompt = prompt or PROMPT_SELECTOR.get_prompt(llm)
llm_chain = LLMChain(
llm=llm, prompt=_prompt, callbacks=callbacks, **(llm_chain_kwargs or {})
)
document_prompt = PromptTemplate(
input_variables=["page_content"], template="Context:\n{page_content}"
)
combine_documents_chain = StuffDocumentsChain(
llm_chain=llm_chain,
document_variable_name="context",
document_prompt=document_prompt,
callbacks=callbacks,
)
return cls(
combine_documents_chain=combine_documents_chain,
callbacks=callbacks,
**kwargs,
)
format_document 는 별도의 함수이다.
def format_document(doc: Document, prompt: BasePromptTemplate[str]) -> str:
return prompt.format(**_get_document_info(doc, prompt))
위에서 만들어진 prompt 에서 format 메소드를 호출한다.
class PromptTemplate(StringPromptTemplate):
### skip
@property
def _prompt_type(self) -> str:
"""Return the prompt type key."""
return "prompt"
def format(self, **kwargs: Any) -> str:
kwargs = self._merge_partial_and_user_variables(**kwargs)
return DEFAULT_FORMATTER_MAPPING[self.template_format](self.template, **kwargs)DEFAULT_FORMATTER_MAPPING 에 self.template_format 을 인자로 넣는데, 기본이 f-string 인 듯하다.
template_format: Literal["f-string", "mustache", "jinja2"] = "f-string"
DEFAULT_FORMATTER_MAPPING: Dict[str, Callable] = {
"f-string": formatter.format,
"mustache": mustache_formatter,
"jinja2": jinja2_formatter,
}self.template 이 Context: 형태였다.
document_prompt = PromptTemplate(
input_variables=["page_content"], template="Context:\n{page_content}"
)
Context: "문서에서 비슷하게 찾은 부분" 의 형태로 들어가게 되고, 이 결과를 이용해서 llm_chain.predict 를 호출하는 것이다.
def combine_docs(
self, docs: List[Document], callbacks: Callbacks = None, **kwargs: Any
) -> Tuple[str, dict]:
inputs = self._get_inputs(docs, **kwargs)
# Call predict on the LLM.
return self.llm_chain.predict(callbacks=callbacks, **inputs), {}
LLMChain의 generate가 호출될 때, input 을 받아서 self.prep_prompts 를 호출하는 부분이 보인다. LLMChain 의 prep_prompts 이다.
def generate(
self,
input_list: List[Dict[str, Any]],
run_manager: Optional[CallbackManagerForChainRun] = None,
) -> LLMResult:
prompts, stop = self.prep_prompts(input_list, run_manager=run_manager)
callbacks = run_manager.get_child() if run_manager else None
if isinstance(self.llm, BaseLanguageModel):
return self.llm.generate_prompt(
prompts,
stop,
callbacks=callbacks,
**self.llm_kwargs,
)
else:
results = self.llm.bind(stop=stop, **self.llm_kwargs).batch(
cast(List, prompts), {"callbacks": callbacks}
)
generations: List[List[Generation]] = []
for res in results:
if isinstance(res, BaseMessage):
generations.append([ChatGeneration(message=res)])
else:
generations.append([Generation(text=res)])
return LLMResult(generations=generations)
단순히 input 값들을 하나씩 꺼내서 붙여주는 것으로 보인다.
def prep_prompts(
self,
input_list: List[Dict[str, Any]],
run_manager: Optional[CallbackManagerForChainRun] = None,
) -> Tuple[List[PromptValue], Optional[List[str]]]:
"""Prepare prompts from inputs."""
stop = None
if len(input_list) == 0:
return [], stop
if "stop" in input_list[0]:
stop = input_list[0]["stop"]
prompts = []
for inputs in input_list:
selected_inputs = {k: inputs[k] for k in self.prompt.input_variables}
prompt = self.prompt.format_prompt(**selected_inputs)
_colored_text = get_colored_text(prompt.to_string(), "green")
_text = "Prompt after formatting:\n" + _colored_text
if run_manager:
run_manager.on_text(_text, end="\n", verbose=self.verbose)
if "stop" in inputs and inputs["stop"] != stop:
raise ValueError(
"If `stop` is present in any inputs, should be present in all."
)
prompts.append(prompt)
return prompts, stop
여기까지 보면 프롬프트가 만들어져서 들어간다는 건 이해했는데, 정확히 어떤 형태로 프롬프트가 들어갔는지 알아보기로 했다. verbose 옵션을 함수에 주거나 global 로 주는 방법이 있었다.
아래와 같이 chain_type_kwargs 에 "verbose": True 를 주면,
qa = RetrievalQA.from_chain_type(llm, chain_type="stuff",
retriever=db.as_retriever(
search_type="mmr",
search_kwargs={"k": 3, "fetch_k" : 10}),
return_source_documents=True,
chain_type_kwargs={
"verbose": True,
}
)
다음과 같은 형태로 어떤 프롬프트가 들어갔는지 출력된다.
(참고로 교재는 from_llm() 을 썼는데, 그 대신 from_chain_type() 을 사용한 것이다.)
초록색 글자의 시작을 보면 System: 으로 context 를 활용해서 대답하라는 부분이 있고, 맨 아래 초록색 글자에는 Human: 하고 질문을 넣은 부분이 보인다.
해당 프롬프트를 코드에서 검색해보니 이런 형태였다. (messages = [] 으로 시스템 메세지, 휴먼 메세지 넣어주는 부분)
prompt_template = """Use the following pieces of context to answer the question at the end. If you don't know the answer, just say that you don't know, don't try to make up an answer.
{context}
Question: {question}
Helpful Answer:"""
PROMPT = PromptTemplate(
template=prompt_template, input_variables=["context", "question"]
)
system_template = """Use the following pieces of context to answer the user's question.
If you don't know the answer, just say that you don't know, don't try to make up an answer.
----------------
{context}"""
messages = [
SystemMessagePromptTemplate.from_template(system_template),
HumanMessagePromptTemplate.from_template("{question}"),
]
CHAT_PROMPT = ChatPromptTemplate.from_messages(messages)
PROMPT_SELECTOR = ConditionalPromptSelector(
default_prompt=PROMPT, conditionals=[(is_chat_model, CHAT_PROMPT)]
)
이 프롬프트를 그대로 복사해서 넣으면 잘 동작한다. (이것 저것 테스트해보느라 다른 document 로 테스트한 내용이지만, llm 의 invoke 를 호출할 때 전체 내용을 복사해서 넣으면 잘 동작한다.)
복사하지 않으면 이렇게 엉뚱한 답을 한다. 내가 전달해준 문서에 없는 내용을 답하며, 정상적으로 답할 때는 위와 같이 정보가 부족하다는 답변이 나와야하는 듯하다.
참고로 OpenAI 의 ChatGPT 로도 테스트해봤는데, RAG 를 사용했을 때 답변과 그렇지 않을 때의 답변이 크게 차이가 났다. 아래는 내가 넣은 문서를 토대로 논문에 맞게 attention 개념을 응답한 예시
RAG 를 연결하지 않았을 때 기본적인 attention 개념에 대해 응답한 예시
그렇다면, Gemini 는 System Message 를 무시하는 것이 아닐까? (쿼리로 보낼 때는 Human: System: 이런 형태로 들어가는 것일 테니 결국 Human 메세지로 인식되었을 것이다.) 아래는 대답을 잘 할 때 global verbose 옵션을 켜고 본 예시
완전히 동일한 질문은 아니지만 chat history 를 전달할 때 system 이 있으면 Gemini 에서는 에러가 발생한다는 질문글이 있다.
https://github.com/langchain-ai/langchain/issues/14700
답변에 따르면 "2. Gemini 는 시스템 메세지를 지원하지 않는다." https://python.langchain.com/docs/integrations/chat/google_generative_ai/#gemini-prompting-faqs
또 다른 질문글. GPT 에서 시스템 역할을 Gemini Pro 에서는 어떻게 구현하냐는 질문 https://www.googlecloudcommunity.com/gc/AI-ML/Implementing-System-Prompts-in-Gemini-Pro-for-Chatbot-Creation/m-p/715501
그렇다... Gemini 는 시스템 메세지를 지원하지 않고, Human 의 입력만 받는다. 아마 내부에서 System: 으로 들어오는 부분은 무시하는 것으로 추정된다.
Gemini 와 RAG 를 사용한 공식 예제를 보면, prompt 를 직접 넣어주는 것을 볼 수 있다.
# Prompt template to query Gemini
llm_prompt_template = """You are an assistant for question-answering tasks.
Use the following context to answer the question.
If you don't know the answer, just say that you don't know.
Use five sentences maximum and keep the answer concise.\n
Question: {question} \nContext: {context} \nAnswer:"""
llm_prompt = PromptTemplate.from_template(llm_prompt_template)
print(llm_prompt)물론 이렇게 넣으면 정상적으로 동작했다.
아주 혹시의 혹시나 프롬프트의 문제인가 싶어서 system 메세지로 저 템플릿 "You are an assistant ~" 을 넣어보기도 했는데, 어림도 없지 시스템 메세지는 또 무시당했다. attention 이 무엇이냐는 질문에 일반적인 cognitive process 라는 답변이 돌아왔다.
from langchain.prompts import SystemMessagePromptTemplate, HumanMessagePromptTemplate, ChatPromptTemplate
system_message_template = SystemMessagePromptTemplate.from_template(
"""You are an assistant for question-answering tasks.
Use the following context to answer the question.
If you don't know the answer, just say that you don't know.
Use five sentences maximum and keep the answer concise.
----------------
Context: {context}
"""
)
# Add this system message to a ChatPromptTemplate
chat_template = ChatPromptTemplate.from_messages([
system_message_template,
HumanMessagePromptTemplate.from_template("{question}"),
# Add other messages here...
])system message 를 넣는 방법 참고
https://github.com/langchain-ai/langchain/discussions/12256
또 비슷한 질문을 찾던 중, langchain 에 Gemini 코드가 포함되어있을 때 user, model 이 아니면 에러를 뱉었다는 질문 답변을 찾았다.
https://github.com/langchain-ai/langchain/issues/14710
def _get_rol(message: BaseMessage) -> str:
if isinstance(message, ChatMessage):
if message.role not in ("user", "model"):
raise ChatGoogleGenerativeAIError(
"Gemini only supports user and model roles when"
" providing it with Chat messages."
)
return message.role
elif isinstance(message, HumanMessage):
return "user"
elif isinstance(message, AIMessage):
return "model"
else:
# TODO: Gemini doesn't seem to have a concept of system messages yet.
raise ChatGoogleGenerativeAIError(
f"Message of '{message.type}' type not supported by Gemini."
" Please only provide it with Human or AI (user/assistant) messages."
)그래서 system 대신 model 을 넣어보려했으나, 기존 템플릿 (Human, System, AI) 말고는 넣을 수 없었다. assistant (AI) 메세지로 시스템 프롬프트를 넣어봤는데 에러가 났다.
ChatGoogleGenerativAI 에 프롬프트 예시를 보면 system 과 human 을 넣고 있는데 이 부분은 잘못된 것으로 보인다.
template = ChatPromptTemplate.from_messages(
[("system", "You are Cat Agent 007"), ("human", "{question}")]
).with_config({"run_name": "my_template", "tags": ["my_template"]})
참고) Gemma 도 system message 지원 안 함
https://huggingface.co/google/gemma-7b-it/discussions/25
방법을 찾은 듯하다. 지원하지 않지만 convert_system_message_to_human 을 설정하라고 한다.
https://python.langchain.com/docs/integrations/chat/google_generative_ai/
Gemini doesn’t support SystemMessage at the moment, but it can be added to the first human message in the row. If you want such behavior, just set the convert_system_message_to_human to True:
from langchain_core.messages import HumanMessage, SystemMessage
model = ChatGoogleGenerativeAI(model="gemini-pro", convert_system_message_to_human=True)
model(
[
SystemMessage(content="Answer only yes or no."),
HumanMessage(content="Is apple a fruit?"),
]
)아니다... 이 예제를 그대로 넣어도 동작하지 않는다..... 시스템 메세지가 들어가고 아니고의 답변 차이가 없다...
/usr/local/lib/python3.10/dist-packages/langchain_core/_api/deprecation.py:119: LangChainDeprecationWarning: The method `BaseChatModel.__call__` was deprecated in langchain-core 0.1.7 and will be removed in 0.2.0. Use invoke instead.
warn_deprecated(
/usr/local/lib/python3.10/dist-packages/langchain_google_genai/chat_models.py:326: UserWarning: Convert_system_message_to_human will be deprecated!
warnings.warn("Convert_system_message_to_human will be deprecated!")
[llm/start] [llm:ChatGoogleGenerativeAI] Entering LLM run with input:
{
"prompts": [
"System: Answer only yes or no.\nHuman: Is apple a fruit?"
]
}
[llm/end] [llm:ChatGoogleGenerativeAI] [961ms] Exiting LLM run with output:
{
"generations": [
[
{
"text": "Botanically speaking, no. Apples are a type of pome, which is a fleshy fruit that contains multiple seeds.",
"generation_info": {
"finish_reason": "STOP",deprecated 라고 warning 나오던 부분 찾았다!!!
def _parse_chat_history(
input_messages: Sequence[BaseMessage], convert_system_message_to_human: bool = False
) -> Tuple[Optional[genai.types.ContentDict], List[genai.types.ContentDict]]:
messages: List[genai.types.MessageDict] = []
if convert_system_message_to_human:
warnings.warn("Convert_system_message_to_human will be deprecated!")
system_instruction: Optional[genai.types.ContentDict] = None
for i, message in enumerate(input_messages):
if i == 0 and isinstance(message, SystemMessage):
system_instruction = _convert_to_parts(message.content)
continue
Vertex AI 예시에는 system_instruction 을 넣어줄 수 있는 부분이 보이는데 이게 나중에 langchain에서 지원하게 되려나?
https://cloud.google.com/vertex-ai/generative-ai/docs/multimodal/send-chat-prompts-gemini?hl=ko
하여튼 현재까지 내용을 종합해보면, system message 를 사용할 수 있는 방법은 없고, human message 에 원하는 지시를 합쳐서 넣어줘야 동작했다.
2. Agent 동작 방법
https://github.com/wikibook/langchain
참고) 06_agent/agent_3.py
import random #←임의의 숫자를 생성하기 위해 필요한 모듈을 가져오기
from langchain.agents import AgentType, Tool, initialize_agent #←Tool을 가져오기
from langchain.chat_models import ChatOpenAI
from langchain.tools import WriteFileTool
chat = ChatOpenAI(
temperature=0,
model="gpt-3.5-turbo"
)
tools = [] #← 다른 도구는 필요 없으므로 일단 삭제
tools.append(WriteFileTool(
root_dir="./"
))
def min_limit_random_number(min_number): #←최솟값을 지정할 수 있는 임의의 숫자를 생성하는 함수
return random.randint(int(min_number), 100000)
tools.append( #←도구를 추가
Tool(
name="Random", #←도구 이름
description="특정 최솟값 이상의 임의의 숫자를 생성할 수 있습니다.", #←도구 설명
func=min_limit_random_number #←도구가 실행될 때 호출되는 함수
)
)
agent = initialize_agent(
tools,
chat,
agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
verbose=True
)
result = agent.run("10 이상의 난수를 생성해 random.txt 파일에 저장하세요.")
print(f"실행 결과: {result}")
위의 소스코드를 실행할 때 global verbose 를 켜고 보니까 system prompt 가 아래와 같은 것이 보였다.
(교재 코드가 바로 돌아가진 않았고 !pip install openai 를 하고 import openai 랑 openai key 추가가 필요했다)
system prompt
"prompts": [
"System: Respond to the human as helpfully and accurately as possible.
You have access to the following tools:\n\n
write_file: Write file to disk,
args: {'file_path': {'title': 'File Path', 'description': 'name of file', 'type': 'string'},
'text': {'title': 'Text', 'description': 'text to write to file', 'type': 'string'},
'append': {'title': 'Append', 'description': 'Whether to append to an existing file.', 'default': False, 'type': 'boolean'}}\n
Random: 특정 최솟값 이상의 임의의 숫자를 생성할 수 있습니다.,
args: {'tool_input': {'type': 'string'}}\n\n
Use a json blob to specify a tool by providing an action key (tool name) and an action_input key (tool input).\n\n
Valid \"action\" values: \"Final Answer\" or write_file, Random\n\n
Provide only ONE action per $JSON_BLOB, as shown:\n\n
```\n
{\n
\"action\": $TOOL_NAME,\n
\"action_input\": $INPUT\n
}\n
```\n\n
Follow this format:\n\n
Question: input question to answer\n
Thought: consider previous and subsequent steps\n
Action:\n
```\n
$JSON_BLOB\n
```\n
Observation: action result\n
... (repeat Thought/Action/Observation N times)\n
Thought: I know what to respond\n
Action:\n
```\n
{\n
\"action\": \"Final Answer\",\n
\"action_input\": \"Final response to human\"\n
}\n
```\n\n
Begin! Reminder to ALWAYS respond with a valid json blob of a single action.
Use tools if necessary. Respond directly if appropriate.
Format is Action:```$JSON_BLOB```then Observation:.\n
Thought:\n
Human: 10 이상의 난수를 생성해 random.txt 파일에 저장하세요." ]
Generations text
"generations":
[ [ { "text": "Thought: I need to generate a random number greater than or equal to 10 and save it to a file.\n\n
Action:\n
```\n
{\n
\"action\": \"Random\",\n
\"action_input\": \"10\"\n
}\n
```",
10 이상의 난수를 생성하고 파일에 저장하겠다는 생각을 만들고, action 으로 Random 을 호출하는 것을 볼 수 있다.
그 다음으로 전달된 프롬프트는 다음과 같다.... 헉 이전과 똑같이 들어가고 그냥 마지막에 This was your previous work 해서 생각한 것을 그대로 복붙해준다. Observation 에 요청 결과도 추가해주긴 했다.
[llm/start]
[chain:AgentExecutor > chain:LLMChain > llm:ChatOpenAI] Entering LLM run with input:
{ "prompts": [ "System: Respond to the human as helpfully and accurately as possible. You have access to the following tools:\n\n
write_file: Write file to disk, args: {'file_path': {'title': 'File Path', 'description': 'name of file', 'type': 'string'}, 'text': {'title': 'Text', 'description': 'text to write to file', 'type': 'string'}, 'append': {'title': 'Append', 'description': 'Whether to append to an existing file.', 'default': False, 'type': 'boolean'}}\n
Random: 특정 최솟값 이상의 임의의 숫자를 생성할 수 있습니다., args: {'tool_input': {'type': 'string'}}\n\n
Use a json blob to specify a tool by providing an action key (tool name) and an action_input key (tool input).\n\nValid \"action\" values: \"Final Answer\" or write_file, Random\n\nProvide only ONE action per $JSON_BLOB, as shown:\n\n```\n{\n \"action\": $TOOL_NAME,\n \"action_input\": $INPUT\n}\n```\n\nFollow this format:\n\nQuestion: input question to answer\nThought: consider previous and subsequent steps\nAction:\n```\n$JSON_BLOB\n```\nObservation: action result\n... (repeat Thought/Action/Observation N times)\nThought: I know what to respond\nAction:\n```\n{\n \"action\": \"Final Answer\",\n \"action_input\": \"Final response to human\"\n}\n```\n\nBegin! Reminder to ALWAYS respond with a valid json blob of a single action. Use tools if necessary. Respond directly if appropriate.
Format is Action:```$JSON_BLOB```then Observation:.\n
Thought:\n
Human: 10 이상의 난수를 생성해 random.txt 파일에 저장하세요.\n\n
This was your previous work (but I haven't seen any of it! I only see what you return as final answer):\n
Thought: I need to generate a random number greater than or equal to 10 and save it to a file.\n\n
Action:\n
```\n
{\n
\"action\": \"Random\",\n
\"action_input\": \"10\"\n
}\n
```\n
Observation: 96445\n
Thought:" ] }
이 결과를 받은 llm 의 답변은 다음과 같다. 생성했으니 이제 file 에 저장하겠다고 write_file 을 호출해서 숫자를 썼다.
[llm/end]
[chain:AgentExecutor > chain:LLMChain > llm:ChatOpenAI] [1.58s] Exiting LLM run with output:
{ "generations": [ [
{ "text": "I have generated a random number greater than or equal to 10.
Now, I will save it to a file.\n\n
Action:\n
```\n
{\n
\"action\": \"write_file\",\n
\"action_input\": {\n
\"file_path\": \"random.txt\",\n
\"text\": \"96445\",\n
\"append\": false\n
}\n
}\n
```", "generation_info": { "finish_reason": "stop", "logprobs": null },
파일이 정상적으로 적혔다는 메세지를 llm 에 다시 프롬프트로 전달한다.
[llm/start]
[chain:AgentExecutor > chain:LLMChain > llm:ChatOpenAI] Entering LLM run with input:
{ "prompts": [
"System: Respond to the human as helpfully and accurately as possible. You have access to the following tools:\n\n
write_file: Write file to disk, args: {'file_path': {'title': 'File Path', 'description': 'name of file', 'type': 'string'}, 'text': {'title': 'Text', 'description': 'text to write to file', 'type': 'string'}, 'append': {'title': 'Append', 'description': 'Whether to append to an existing file.', 'default': False, 'type': 'boolean'}}\n
Random: 특정 최솟값 이상의 임의의 숫자를 생성할 수 있습니다., args: {'tool_input': {'type': 'string'}}\n\nUse a json blob to specify a tool by providing an action key (tool name) and an action_input key (tool input).\n\nValid \"action\" values: \"Final Answer\" or write_file, Random\n\n
Provide only ONE action per $JSON_BLOB, as shown:\n\n```\n{\n \"action\": $TOOL_NAME,\n \"action_input\": $INPUT\n}\n```\n\nFollow this format:\n\nQuestion: input question to answer\nThought: consider previous and subsequent steps\nAction:\n```\n$JSON_BLOB\n```\nObservation: action result\n... (repeat Thought/Action/Observation N times)\nThought: I know what to respond\nAction:\n```\n{\n \"action\": \"Final Answer\",\n \"action_input\": \"Final response to human\"\n}\n```\n\nBegin! Reminder to ALWAYS respond with a valid json blob of a single action. Use tools if necessary. Respond directly if appropriate. Format is Action:```$JSON_BLOB```then Observation:.\nThought:\nHuman: 10 이상의 난수를 생성해 random.txt 파일에 저장하세요.\n\n
This was your previous work (but I haven't seen any of it! I only see what you return as final answer):\n
Thought: I need to generate a random number greater than or equal to 10 and save it to a file.\n\n
Action:\n```\n{\n \"action\": \"Random\",\n \"action_input\": \"10\"\n}\n```\n
Observation: 96445\n
Thought:I have generated a random number greater than or equal to 10. Now, I will save it to a file.\n\n
Action:\n```\n{\n \"action\": \"write_file\",\n \"action_input\": {\n \"file_path\": \"random.txt\",\n \"text\": \"96445\",\n \"append\": false\n }\n}\n```\n
Observation: File written successfully to random.txt.\n
Thought:" ] }
그러면 이제 llm 은 Final Answer 를 반환한다.
[llm/end]
[chain:AgentExecutor > chain:LLMChain > llm:ChatOpenAI] [1.39s] Exiting LLM run with output:
{ "generations": [ [ {
"text": "I have generated a random number greater than or equal to 10 and saved it to a file named random.txt.\n
Action:\n```\n{\n \"action\": \"Final Answer\",\n \"action_input\": \"Random number greater than or equal to 10 has been saved to random.txt.\"\n}\n
```", "generation_info": {
이 action input 의 결과가 나에게 실행 결과로 리턴된다.
[chain/end]
[chain:AgentExecutor] [4.27s] Exiting Chain run with output:
{ "output": "Random number greater than or equal to 10 has been saved to random.txt." }
실행 결과: Random number greater than or equal to 10 has been saved to random.txt.
agent 의 비밀은 사용 가능한 tool 목록을 넘겨주고 json 파싱을 해서 사용하는 순차적인 프롬프트 엔지니어링이었다.
역시나 Gemini 는 동작하지 않았다.
동일한 Prompt 가 입력되었으나, 파이썬으로 난수 생성하고 저장하는 코드를 짜줬다...
[llm/end]
[chain:AgentExecutor > chain:LLMChain > llm:ChatGoogleGenerativeAI] [2.91s] Exiting LLM run with output:
{ "generations": [ [ {
"text": "```python\n
import random\n\n
# 난수 생성\n
numbers = [random.randint(10, 100) for _ in range(10)]\n\n
# 파일 열기\n
with open(\"random.txt\", \"w\") as f:\n
# 난수 파일 저장\n
for number in numbers:\n
f.write(str(number) + \"\\n
\")\n
```", "generation_info": {
Gemini 에서 Agent 를 사용하려면 이 system message 가 human message 로 들어가게 해야한다.
좋은 예시 코드를 찾았다.
https://github.com/MikeChan-HK/Gemini-agent-example/blob/main/Gemini_agents.ipynb
langchain hub 에 이렇게 사용할 프롬프트를 업로드해두는데, 위 예시 코드에서 활용한 프롬프트를 직접 보려면 아래 경로에 가면 볼 수 있다.
https://smith.langchain.com/hub/mikechan/gemini
이렇게 하면 System Message 가 아닌 Human Message 로 들어가기 때문에 Thought: 이런 식으로 원하는 형태로 동작하는 걸 볼 수 있었다.
"text": "Thought: Do I need to use a tool? No\n
Final Answer: Let the cost price of the chair be Rs. x. Then, Marked price = Rs. (x + 20% of x) = Rs. (1.2x)\n\n
Selling price = Marked price - Discount = Rs. (1.2x - 90)\n\n
Loss = Rs. (x - Selling price) = Rs. (x - 1.2x + 90) = Rs. (0.2x - 90)\n\n
Percentage loss = (Loss / Cost price) * 100 = (0.2x - 90 / x) * 100 = 16\n\n
=> 0.2x - 90 = 0.16x\n\n
=> 0.04x = 90\n\n
=> x = 90 / 0.04 = Rs. 2250\n\n
Therefore, the cost price of the chair is Rs. 2250.",
"generation_info": {
유사한 예시 (유튜브)
https://www.youtube.com/watch?v=6tmGPhDhQto
Note)
RetrievalQA 에서 BaseRetrievalQA 로 올라가도록 한 이유는 무엇일까? 다른 상속받은 자식 클래스 있는지 확인해보기
흐름을 그림으로 그려둘 필요 있을지
끝