Techbrad

langchain에서 제공하는 cache 기능을 사용하여 질의 응답을 캐싱하였으나 내부적으로 어떻게 동작하는지 궁금해서 찾아보았다. 먼저 chat_models.py의 _agenerate_with_cache 메소드가 실행이 된다. 그 안에 아래의 함수가 호출이 되면서 cache에 같은 데이터가 있는지 찾는다.cache_val = await llm_cache.alookup(prompt, llm_string) #chat_models.pyasync def _agenerate_with_cache( self, messages: List[BaseMessage], stop: Optional[List[str]] = None, run_manager: Optional[AsyncCallbackManage..

사용자의 쿼리에 대하여 토큰 비용을 줄일 수 있는 방법 중에 캐싱 기법을 적용해보았다. 물론 토큰 비용이 얼마 되지 않아 그냥 해도 되지만 응답 속도는 확연하게 체감이 될 정도로 빨랐다. 랭체인에서 제공하는 라이브러리를 사용하면 캐싱 구현은 정말 간단하다. from langchain.cache import RedisCache from langchain.globals import set_llm_cache set_llm_cache(RedisCache(redis_=Redis(host='redis', port=6379, db=1))) response = rag_chain_with_source.invoke(text.question) 질문을 할때 마다 Key가 생성이 되었다. Key는 Hash로 이루어 져있는데 어..

FastAPI + Sqlalchemy를 사용하여 개발하던 중 insert 실패 후 다른 데이터를 insert를 할때 아래 와 같은 애러가 발생하였다. sqlalchemy.exc.PendingRollbackError: This Session's transaction has been rolled back due to a previous exception during flush. To begin a new transaction with this Session, first issue Session.rollback(). Original exception was: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint 위의 애..

RAG 어플리케이션을 개발하면서 LLM에서 생성하는 텍스트를 스트리밍으로 한글자씩 나오도록 구현해보았다. 큐를 사용하여 글자를 순차적으로 전달하였고 Langchain의 BaseCallbackHandler 를 사용했다. BaseCallbackHandler에 대해서는 조금 더 공부가 필요한 듯 보인다. API class Question(BaseModel): question: str @router.post( path='/text-stream/', description="Enter the question" ) @inject async def generate_text_streaming( query: Question, rag_service: Rag_Service = Depends(Provide[Container.r..