LangChain

Weave를 사용해 LangChain Python 라이브러리를 통해 이루어지는 모든 call을 추적하고 로깅합니다

Weave는 LangChain Python 라이브러리를 통해 이루어지는 모든 call을 손쉽게 추적하고 로깅할 수 있도록 설계되었습니다.

LLM으로 작업할 때 디버깅은 피할 수 없습니다. 모델 호출이 실패하거나, 출력 형식이 잘못되었거나, 중첩된 모델 call로 인해 혼란이 생기는 등 문제의 원인을 정확히 짚어내기 어려울 수 있습니다. LangChain 애플리케이션은 여러 step과 LLM call로 구성되는 경우가 많기 때문에, chain과 agent의 내부 동작을 이해하는 것이 매우 중요합니다.

Weave는 LangChain 애플리케이션의 트레이스를 자동으로 수집하여 이 과정을 단순화합니다. 이를 통해 애플리케이션의 성능을 모니터링하고 분석할 수 있으므로, LLM 워크플로를 더 쉽게 디버깅하고 최적화할 수 있습니다.

​

시작하기

시작하려면 스크립트 맨 앞에서 weave.init()를 호출하면 됩니다. weave.init()의 인수는 프로젝트 이름으로, 트레이스를 정리하는 데 도움이 됩니다.

import weave
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# 프로젝트 이름으로 Weave 초기화
weave.init("langchain_demo")

llm = ChatOpenAI()
prompt = PromptTemplate.from_template("1 + {number} = ")

llm_chain = prompt | llm

output = llm_chain.invoke({"number": 2})

print(output)

​

call 메타데이터 추적

LangChain call의 메타데이터를 추적하려면 weave.attributes 컨텍스트 관리자를 사용할 수 있습니다. 이 컨텍스트 관리자를 사용하면 체인이나 단일 요청처럼 특정 코드 블록에 맞춤형 메타데이터를 설정할 수 있습니다.

import weave
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# 프로젝트 이름으로 Weave 초기화
weave.init("langchain_demo")

llm = ChatOpenAI()
prompt = PromptTemplate.from_template("1 + {number} = ")

llm_chain = prompt | llm

with weave.attributes({"my_awesome_attribute": "value"}):
    output = llm_chain.invoke()

print(output)

Weave는 LangChain call의 트레이스에 메타데이터를 자동으로 기록합니다. 아래와 같이 Weave 웹 인터페이스에서 메타데이터를 볼 수 있습니다.

​

트레이스

개발 단계와 프로덕션 환경 모두에서 LLM 애플리케이션의 트레이스를 중앙 데이터베이스에 저장하는 것은 매우 중요합니다. 이러한 트레이스는 유용한 데이터셋을 제공하므로 애플리케이션을 디버깅하고 개선하는 데 필수적입니다.

Weave는 LangChain 애플리케이션의 트레이스를 자동으로 캡처합니다. 프롬프트 템플릿, chain, LLM calls, 도구, 에이전트 step를 포함해 LangChain 라이브러리를 통해 이루어지는 모든 call을 추적하고 로깅합니다. Weave 웹 인터페이스에서 트레이스를 확인할 수 있습니다.

​

call 수동 트레이싱하기

자동 트레이싱 외에도 WeaveTracer 콜백이나 weave_tracing_enabled 컨텍스트 관리자를 사용해 call을 수동으로 트레이싱할 수 있습니다. 이러한 방법은 LangChain 애플리케이션의 개별 부분에서 요청 콜백을 사용하는 것과 유사합니다.

참고: Weave는 기본적으로 LangChain Runnable을 트레이스하며, 이 기능은 weave.init()를 호출하면 활성화됩니다. weave.init()를 호출하기 전에 환경 변수 WEAVE_TRACE_LANGCHAIN을 "false"로 설정하면 이 동작을 비활성화할 수 있습니다. 이렇게 하면 애플리케이션에서 특정 체인이나 개별 요청의 트레이싱 동작을 제어할 수 있습니다.

​

WeaveTracer 사용

특정 요청을 트레이스하려면 개별 LangChain 컴포넌트에 WeaveTracer 콜백을 전달할 수 있습니다.

import os

os.environ["WEAVE_TRACE_LANGCHAIN"] = "false" # <- 전역 트레이싱을 명시적으로 비활성화합니다.

from weave.integrations.langchain import WeaveTracer
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
import weave

# 프로젝트 이름으로 Weave를 초기화합니다
weave.init("langchain_demo")  # <-- 환경 변수가 `false`로 명시적으로 설정되어 있으므로 여기서는 트레이싱을 활성화하지 않습니다

weave_tracer = WeaveTracer()

config = {"callbacks": [weave_tracer]}

llm = ChatOpenAI()
prompt = PromptTemplate.from_template("1 + {number} = ")

llm_chain = prompt | llm

output = llm_chain.invoke({"number": 2}, config=config) # <-- 이 체인 호출에 대해서만 트레이싱을 활성화합니다.

llm_chain.invoke({"number": 4})  # <-- langchain 호출에 대한 트레이싱은 비활성화되지만 openai 호출은 계속 트레이싱됩니다

​

weave_tracing_enabled 컨텍스트 관리자 사용

또는 weave_tracing_enabled 컨텍스트 관리자를 사용해 특정 코드 블록에서만 Tracing을 활성화할 수 있습니다.

import os

os.environ["WEAVE_TRACE_LANGCHAIN"] = "false" # <- 전역 트레이싱을 명시적으로 비활성화합니다.

from weave.integrations.langchain import weave_tracing_enabled
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
import weave

# 프로젝트 이름으로 Weave를 초기화합니다
weave.init("langchain_demo")  # <-- 환경 변수가 명시적으로 `false`로 설정되어 있으므로 여기서는 트레이싱을 활성화하지 않습니다

llm = ChatOpenAI()
prompt = PromptTemplate.from_template("1 + {number} = ")

llm_chain = prompt | llm

with weave_tracing_enabled():  # <-- 이 체인 호출에 대해서만 트레이싱을 활성화합니다.
    output = llm_chain.invoke({"number": 2})


llm_chain.invoke({"number": 4})  # <-- langchain 호출에 대해서는 트레이싱이 활성화되지 않지만 openai 호출은 계속 트레이싱됩니다

​

설정

weave.init()를 호출하면 환경 변수 WEAVE_TRACE_LANGCHAIN을 "true"로 설정해 트레이싱을 활성화합니다. 이렇게 하면 Weave가 LangChain 애플리케이션의 트레이스를 자동으로 수집할 수 있습니다. 이 동작을 비활성화하려면 환경 변수를 "false"로 설정하세요.

​

LangChain 콜백과의 관계

​

자동 로깅

weave.init()이 제공하는 자동 로깅은 LangChain 애플리케이션의 모든 구성 요소에 생성자 콜백을 전달하는 것과 유사합니다. 즉, 프롬프트 템플릿, 체인, LLM 호출, 도구, 에이전트 step을 포함한 모든 상호작용이 애플리케이션 전체에서 전역적으로 추적됩니다.

​

수동 로깅

수동 로깅 방법(WeaveTracer 및 weave_tracing_enabled)은 LangChain 애플리케이션의 개별 부분에서 요청 콜백을 사용하는 방식과 유사합니다. 이 방법을 사용하면 애플리케이션의 어떤 부분을 트레이스할지 더 세밀하게 제어할 수 있습니다.

• 생성자 콜백: 전체 체인 또는 컴포넌트에 적용되어 모든 상호작용을 일관되게 로깅합니다.
• 요청 콜백: 특정 요청에 적용되어 특정 호출을 자세히 트레이스할 수 있습니다.

Weave를 LangChain과 통합하면 LLM 애플리케이션 전반에서 포괄적인 로깅과 모니터링을 수행할 수 있어 디버깅과 성능 최적화가 더 쉬워집니다.

더 자세한 내용은 LangChain 문서를 참고하세요.

​

Models 및 평가

프롬프트, 모델 설정, Inference 파라미터 등 여러 컴포넌트가 관련되므로 애플리케이션의 다양한 사용 사례에 맞게 LLM을 구성하고 평가하는 일은 까다롭습니다. weave.Model을 사용하면 시스템 프롬프트나 사용 중인 모델과 같은 실험 세부 정보를 수집하고 정리할 수 있어 서로 다른 반복 버전을 더 쉽게 비교할 수 있습니다.

다음 예제는 LangChain 체인을 WeaveModel로 래핑하는 방법을 보여줍니다:

import json
import asyncio

import weave

from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# 프로젝트 이름으로 Weave 초기화
weave.init("langchain_demo")

class ExtractFruitsModel(weave.Model):
    model_name: str
    prompt_template: str

    @weave.op()
    async def predict(self, sentence: str) -> dict:
        llm = ChatOpenAI(model=self.model_name, temperature=0.0)
        prompt = PromptTemplate.from_template(self.prompt_template)

        llm_chain = prompt | llm
        response = llm_chain.invoke({"sentence": sentence})
        result = response.content

        if result is None:
            raise ValueError("No response from model")
        parsed = json.loads(result)
        return parsed

model = ExtractFruitsModel(
    model_name="gpt-3.5-turbo-1106",
    prompt_template='Extract fields ("fruit": <str>, "color": <str>, "flavor": <str>) from the following text, as json: {sentence}',
)
sentence = "There are many fruits that were found on the recently discovered planet Goocrux. There are neoskizzles that grow there, which are purple and taste like candy."

prediction = asyncio.run(model.predict(sentence))

# Jupyter Notebook에서 실행하는 경우:
# prediction = await model.predict(sentence)

print(prediction)

이 코드는 Weave UI에서 시각화할 수 있는 모델을 생성합니다:

또한 Weave Models를 serve 및 평가와 함께 사용할 수도 있습니다.

​

평가

평가는 모델의 성능을 측정하는 데 도움이 됩니다. weave.Evaluation 클래스를 사용하면 특정 작업이나 데이터셋에서 모델이 얼마나 잘 작동하는지 파악할 수 있어, 여러 모델과 애플리케이션의 다양한 반복 버전을 더 쉽게 비교할 수 있습니다. 다음 예제에서는 앞서 만든 모델을 평가하는 방법을 보여줍니다.

from weave.scorers import MultiTaskBinaryClassificationF1

sentences = [
    "There are many fruits that were found on the recently discovered planet Goocrux. There are neoskizzles that grow there, which are purple and taste like candy.",
    "Pounits are a bright green color and are more savory than sweet.",
    "Finally, there are fruits called glowls, which have a very sour and bitter taste which is acidic and caustic, and a pale orange tinge to them.",
]
labels = [
    {"fruit": "neoskizzles", "color": "purple", "flavor": "candy"},
    {"fruit": "pounits", "color": "bright green", "flavor": "savory"},
    {"fruit": "glowls", "color": "pale orange", "flavor": "sour and bitter"},
]
examples = [
    {"id": "0", "sentence": sentences[0], "target": labels[0]},
    {"id": "1", "sentence": sentences[1], "target": labels[1]},
    {"id": "2", "sentence": sentences[2], "target": labels[2]},
]

@weave.op()
def fruit_name_score(target: dict, output: dict) -> dict:
    return {"correct": target["fruit"] == output["fruit"]}


evaluation = weave.Evaluation(
    dataset=examples,
    scorers=[
        MultiTaskBinaryClassificationF1(class_names=["fruit", "color", "flavor"]),
        fruit_name_score,
    ],
)
scores = asyncio.run(evaluation.evaluate(model)))
# Jupyter Notebook에서 실행하는 경우:
# scores = await evaluation.evaluate(model)

print(scores)

이 코드는 Weave UI에서 시각화할 수 있는 Evaluation 트레이스를 생성합니다:

Weave를 Langchain과 통합하면 LLM 애플리케이션 전반에 걸쳐 로깅과 모니터링을 포괄적으로 수행할 수 있어, 디버깅과 성능 최적화가 한층 쉬워집니다.

​

알려진 문제

• 비동기 call 트레이싱 - Langchain의 AsyncCallbackManager 구현상 버그로 인해 비동기 call이 올바른 순서대로 트레이스되지 않습니다. 이 문제를 해결하기 위해 PR을 제출했습니다. 따라서 Langchain Runnable에서 ainvoke, astream, abatch 메서드를 사용할 때 트레이스 내 call 순서가 정확하지 않을 수 있습니다.