[Langchain] Langchain v0.3 패치노트

1. 변경 사항

• 내부적으로 모든 패키지가 Pydantic 1에서 Pydantic 2로 업그레이드됨. 이제 Pydantic 2를 완전히 지원
• Pydantic 1은 2024년 6월에 지원 종료
• Python 3.8은 2024년 10월에 지원이 종료

---

2. 새롭게 추가된 기능

2.1. the latest integration packages 패치,

• 기존 langchain-community 의 내용이 integration packages로 많이 넘어감 
  •  langchain-openai
  •  langchain-anthropic
  •  langchain-google-vertexai
  •  langchain-aws
  •  langchain-huggingface
  •  langchain-mistralai

2.2. tool의 정의와 사용이 단순화됨

2.2.1. Tool 정의 간소화

• bind_tools() 함수는 일반적인 Python 함수를 LangChain 모델이 호출할 수 있는 도구(tool)로 변환합니다.
• 예를 들어, 이 예시에서 validate_user 함수는 원래 단순한 Python 함수이지만, bind_tools()를 통해 LangChain의 모델이 해당 함수를 직접 호출할 수 있게 된 것입니다.

from typing import List
from typing_extensions import TypedDict

from langchain_anthropic import ChatAnthropic

# 사용자의 주소 정보를 저장하는 데이터 타입을 정의 (TypedDict 사용)
class Address(TypedDict):
    street: str  # 거리 정보
    city: str    # 도시 정보
    state: str   # 주 정보

# validate_user 함수 정의: 사용자 ID와 이전 주소 목록을 받아 유효성 검사를 수행하는 도구
def validate_user(user_id: int, addresses: List[Address]) -> bool:
    """
    사용자의 이전 주소를 기반으로 유효성을 검사하는 함수.
    
    Args:
        user_id: (int) 유저 ID
        addresses: (List[Address]) 유저가 거주했던 이전 주소 목록
    
    Returns:
        bool: 유효성 검사 결과 (유효하면 True, 유효하지 않으면 False)
    """
    # 간단한 유효성 검사 로직 (여기서는 항상 True 반환)
    return True

# Claude-3 모델을 사용한 ChatAnthropic 객체 생성
llm = ChatAnthropic(
    model="claude-3-sonnet-20240229"
).bind_tools([validate_user])  # validate_user 함수를 도구로 바인딩하여 LLM에서 호출 가능하게 함

# 모델에 입력을 전달하여 validate_user 도구 호출
result = llm.invoke(
    "Could you validate user 123? They previously lived at "
    "123 Fake St in Boston MA and 234 Pretend Boulevard in "
    "Houston TX."
)

# 모델이 호출한 도구의 결과를 확인
result.tool_calls
[{'name': 'validate_user',
  'args': {'user_id': 123,  # user_id는 입력에서 추출됨
   'addresses': [  # addresses는 입력에서 추출된 주소 목록
     {'street': '123 Fake St', 'city': 'Boston', 'state': 'MA'},
     {'street': '234 Pretend Boulevard', 'city': 'Houston', 'state': 'TX'}]},
  'id': 'toolu_011KnPwWqKuyQ3kMy6McdcYJ',  # 도구 호출에 대한 고유 ID
  'type': 'tool_call'}]  # 호출된 도구의 타입은 'tool_call'로 명시됨

 

 

2.2.2. 유연해진 Tool 의 input

• 모델에서 생성한 입력을 도구에 직접 전달 할 수 있음

tool_call = ai_msg.tool_calls[0]
# -> ToolCall(args={...}, id=..., ...)

tool_message = tool.invoke(tool_call)
# -> ToolMessage(
#    content="도구 결과 예시...",
#    tool_call_id=...,
#    name="tool_name"
#)

• Tool에 LangGraph 상태를 전달 (상세 내용)
  • 내용이 길기 때문에 요약하자면, 먼저 tool을 정의한다.
  • 모델에 tool을 입력해준다.
  • 그래프를 정의할 때, node에 tool을 넣어준다.

2.2.3. 풍부한 도구(Tool) 출력

• 도구는 모델에서 호출할 수 있는 유틸리티이며, 그 출력은 모델에 피드백되도록 설계되었습니다.
• 도구는 모델에 전달되지 않을 추가 데이터를 포함하여 결과를 반환할 수 있으며, 일부만 모델에 반환할 수 있습니다.
• 예를 들어, 도구가 사용자 지정 개체, 데이터 프레임 또는 이미지를 반환하는 경우 실제 출력을 모델에 전달하지 않고 이 출력에 대한 일부 메타데이터를 모델에 전달하고 싶을 수 있습니다.
• 이를 위해, ToolMessage는 모델에 전달될 메시지(content)와 모델 외부에서 사용할 수 있는 아티팩트(artifact)를 구분합니다.

Tool 정의 예시

import random
from typing import List, Tuple

from langchain_core.tools import tool


@tool(response_format="content_and_artifact")
def generate_random_ints(min: int, max: int, size: int) -> Tuple[str, List[int]]:
    """Generate size random ints in the range [min, max]."""
    array = [random.randint(min, max) for _ in range(size)]
    content = f"Successfully generated array of {size} random ints in [{min}, {max}]."
    return content, array

 

도구 호출

• 이 코드는 generate_random_ints 도구를 호출하지만, 단순히 함수의 인수만 전달하기 때문에 content만 반환합니다.

generate_random_ints.invoke({"min": 0, "max": 9, "size": 10})

 

ToolCall을 이용한 호출

• ToolCall을 사용하여 도구를 호출할 때, 추가적으로 ToolMessage를 반환받을 수 있습니다.
• 여기서는 name, args, id, type을 포함하는 ToolCall을 전달합니다. 이 호출은 도구 호출 결과로 content와 artifact를 모두 반환합니다.

generate_random_ints.invoke(
    {
        "name": "generate_random_ints",
        "args": {"min": 0, "max": 9, "size": 10},
        "id": "123",  # 도구 호출에 필요한 고유 ID
        "type": "tool_call",  # 호출 유형
    }
)

 

모델과 도구 통합

import getpass
import os

os.environ["OPENAI_API_KEY"] = getpass.getpass()

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools([generate_random_ints])

 

모델 실행

• 반환된 ai_msg.tool_calls에는 generate_random_ints 도구를 호출한 기록이 포함됩니다. 이 호출은 도구의 이름, 인수(min, max, size), 호출 ID 등이 포함된 구조입니다.

ai_msg = llm_with_tools.invoke("generate 6 positive ints less than 25")
ai_msg.tool_calls

 

2.2.4. 도구(Tool)의 오류 처리

1. tool에 try / except 로 처리해준다.

2. Fallbacks을 이용한다.

• 더 나은 체인을 하나 더 만들어서, 사용하도록 한다.
• 이는 chain에 with_fallbacks 함수를 사용해준다.

chain = llm_with_tools | (lambda msg: msg.tool_calls[0]["args"]) | complex_tool

better_model = ChatOpenAI(model="gpt-4-1106-preview", temperature=0).bind_tools(
    [complex_tool], tool_choice="complex_tool"
)

better_chain = better_model | (lambda msg: msg.tool_calls[0]["args"]) | complex_tool

chain_with_fallback = chain.with_fallbacks([better_chain])

chain_with_fallback.invoke(
    "use complex tool. the args are 5, 2.1, empty dictionary. don't forget dict_arg"
)

 

2.3. 채팅 모델과 상호작용할 수 있는 유틸리티 추가

2.3.1. 범용 모델 생성자

최종 사용자가 애플리케이션에 어떤 모델 공급자와 모델을 제공할지 지정할 수 있도록 합니다. init_chat_model() 메서드로 여러 다른 모델을 통합할 수 있습니다.

 

다양한 모델을 테스트하거나 동적으로 선택할 필요가 있는 경우 적합합니다.

from langchain.chat_models import init_chat_model

# 기본적으로 특정 모델을 지정하지 않으면, 설정을 바꾸는 옵션 없이 모델을 사용 가능
configurable_model = init_chat_model(temperature=0)  # 모델의 temperature 설정

# GPT-4o 모델을 사용하여 질문
response_gpt4o = configurable_model.invoke(
    "What's your name?",
    config={"configurable": {"model": "gpt-4o"}}  # 모델을 GPT-4o로 설정
)
print(f"GPT-4o response: {response_gpt4o}")

# Claude 3.5 Sonnet 모델을 사용하여 질문
response_claude = configurable_model.invoke(
    "What's your name?",
    config={"configurable": {"model": "claude-3-5-sonnet-20240620"}}  # Claude 3.5 Sonnet으로 설정
)
print(f"Claude-3.5 Sonnet response: {response_claude}")

 

2.3.2. 속도 제한기

API 제공자의 제한으로 인해 너무 많은 요청을 보내면 속도 제한(rate limit)에 걸릴 수 있습니다. 예를 들어, 테스트 데이터셋에서 채팅 모델의 성능을 벤치마킹하기 위해 여러 병렬 쿼리를 실행할 때 이런 상황이 발생할 수 있습니다. 이러한 상황에서는 속도 제한기를 사용하여 요청 빈도를 API에서 허용하는 속도에 맞출 수 있습니다.

from langchain_core.rate_limiters import InMemoryRateLimiter

rate_limiter = InMemoryRateLimiter(
    requests_per_second=0.1,  # <-- 매우 느림! 10초에 한 번씩만 요청을 보낼 수 있음!!
    check_every_n_seconds=0.1,  # 100ms마다 요청을 보낼 수 있는지 확인하기 위해 깨어남,
    max_bucket_size=10,  # 최대 버스트 크기를 제어.
)

 

모델 선택 및 타임 리미터 적용

from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(model_name="claude-3-opus-20240229", rate_limiter=rate_limiter)

 

 

2.3.3. 메시지 유틸리티

• 메세지(Message)는 chat models의 인풋과 아웃풋이다.
• content, role로 구성되어 있다.

 

2.4. 사용자 지정 이벤트 전송

 

사용자 지정 이벤트는 어디에 사용하는가? 커스텀 이벤트는 주로 특정 작업의 상태 추적, 실시간 모니터링, 워크플로우 관리 등 다양한 상황에서 사용됩니다. 특히, 비동기 환경이나 이벤트 기반 시스템에서 작업의 흐름을 제어하거나 중요한 정보를 수집하는 데 유용합니다. 커스텀 이벤트는 다음과 같은 주요 용도에서 많이 활용됩니다.

 

2.4.1. Astream Events API

adispatch_custom_event 함수는 이벤트를 발행하는 역할을 담당합니다.

 

커스텀 이벤트를 비동기 환경에서 처리하기 위한 API입니다. adispatch_custom_event를 사용해 비동기 방식으로 이벤트를 발행할 수 있습니다.

 

Astream Events API는 모델 실행 중 발생하는 중요한 이벤트를 실시간으로 기록하고 추적하는 데 사용됩니다. 이를 통해 모델의 상태, 성능, 오류, 상호작용 등을 상세하게 모니터링할 수 있으며, 비동기 작업을 효율적으로 관리하는 데 도움을 줍니다.

from langchain_core.callbacks.manager import (
    adispatch_custom_event,
)
from langchain_core.runnables import RunnableLambda
from langchain_core.runnables.config import RunnableConfig


@RunnableLambda
async def foo(x: str) -> str:
    await adispatch_custom_event("event1", {"x": x})
    await adispatch_custom_event("event2", 5)
    return x


async for event in foo.astream_events("hello world", version="v2"):
    print(event) # 이벤트 정보를 출력

 

2.4.2. Async Callback Handler (비동기 콜백 핸들러)

AsyncCallbackHandler는 비동기적으로 커스텀 이벤트를 처리하는 핸들러 클래스입니다.

 

on_custom_event 메서드는 커스텀 이벤트가 발생할 때 호출되며, 이벤트 이름, 데이터, 실행 ID(run_id), 태그(tags), 메타데이터(metadata)를 인수로 받아 처리합니다.

class AsyncCustomCallbackHandler(AsyncCallbackHandler):
    async def on_custom_event(
        self,
        name: str,
        data: Any,
        *,
        run_id: UUID,
        tags: Optional[List[str]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        **kwargs: Any,
    ) -> None:
        print(
            f"Received event {name} with data: {data}, with tags: {tags}, with metadata: {metadata} and run_id: {run_id}"
        )
        
@RunnableLambda
async def bar(x: str, config: RunnableConfig) -> str:
    """An example that shows how to manually propagate config.

    You must do this if you're running python<=3.10.
    """
    await adispatch_custom_event("event1", {"x": x}, config=config)
    await adispatch_custom_event("event2", 5, config=config)
    return x


async_handler = AsyncCustomCallbackHandler()
await foo.ainvoke(1, {"callbacks": [async_handler], "tags": ["foo", "bar"]})

 

2.4.3. Sync Callback Handler (동기 콜백 핸들러)

동기적으로 작업하는 것은 예만 보고 넘어갑시다.

from typing import Any, Dict, List, Optional
from uuid import UUID

from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.callbacks.manager import (
    dispatch_custom_event,
)
from langchain_core.runnables import RunnableLambda
from langchain_core.runnables.config import RunnableConfig


class CustomHandler(BaseCallbackHandler):
    def on_custom_event(
        self,
        name: str,
        data: Any,
        *,
        run_id: UUID,
        tags: Optional[List[str]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        **kwargs: Any,
    ) -> None:
        print(
            f"Received event {name} with data: {data}, with tags: {tags}, with metadata: {metadata} and run_id: {run_id}"
        )


@RunnableLambda
def foo(x: int, config: RunnableConfig) -> int:
    dispatch_custom_event("event1", {"x": x})
    dispatch_custom_event("event2", {"x": x})
    return x


handler = CustomHandler()
foo.invoke(1, {"callbacks": [handler], "tags": ["foo", "bar"]})

 

2.5. 통합 문서 및 API 참조가 개선됨

• 문서를 새로 만듦

2.6. 마이그레이션 가이드

• 생략

---

3. 코드 업데이트 방법

• langchain, langchain-community, langchain-core 버전 0.2에서 0.3으로 업그레이드 권장.
• langgraph는 0.2.20 버전 이상으로 업그레이드해야 함.

---

4. Pydantic 2

더이상 pydancit v1을 사용하지 않습니다. 그리고 pydantic을 직접적으로 사용합니다.

 

예를들어 아래와 같은 코드는

from langchain_core.pydantic_v1 import BaseModel

 

아래와 같이 변형해 사용합니다.

from pydantic import BaseModel