Chain Of Density

W&B Weave에서 Chain of Density를 사용하는 방법을 알아보세요

이 노트북은 인터랙티브 노트북입니다. 로컬에서 실행하거나 아래 링크를 사용할 수 있습니다:

• Google Colab에서 열기
• GitHub에서 소스 보기

​

Chain of Density를 활용한 요약

중요한 세부 정보를 유지하면서 복잡한 기술 문서를 요약하는 일은 쉽지 않습니다. Chain of Density(CoD) 요약 기법은 요약문을 반복적으로 정제해 더 간결하면서도 정보 밀도가 높은 형태로 만드는 방법을 제공합니다. 이 가이드는 애플리케이션을 추적하고 평가하기 위해 Weave를 사용해 CoD를 구현하는 방법을 설명합니다.

​

Chain of Density 요약이란?

Chain of Density (CoD)는 반복적으로 요약을 정제해, 점점 더 간결하면서도 정보 밀도가 높은 요약을 생성하는 기법입니다. 동작 방식은 다음과 같습니다.

1. 초기 요약을 작성한다.
2. 핵심 정보를 유지한 채, 요약을 반복적으로 다듬어 더 간결하게 만든다.
3. 각 반복마다 Entities와 기술적 세부 정보의 밀도를 높인다.

이 방식은 세부 정보를 보존하는 것이 중요한 과학 논문이나 기술 문서를 요약할 때 특히 유용합니다.

​

왜 Weave를 사용하나요?

이 튜토리얼에서는 Weave를 사용해 ArXiv 논문에 대한 Chain of Density 요약 파이프라인을 구현하고 평가합니다. 이 튜토리얼을 통해 다음을 배우게 됩니다:

1. LLM 파이프라인 추적: Weave를 사용해 요약 과정의 입력, 출력, 중간 단계를 자동으로 로깅합니다.
2. LLM 출력 평가: Weave의 내장 도구를 사용해 요약 결과를 엄밀하고 공정하게 비교·평가합니다.
3. 조합 가능한 오퍼레이션 구성: 요약 파이프라인의 여러 부분에서 Weave 오퍼레이션을 결합하고 재사용합니다.
4. 원활한 통합: 기존 Python 코드에 최소한의 오버헤드로 Weave를 추가합니다.

이 튜토리얼을 마치면, Weave의 모델 서빙, 평가, 결과 추적 기능을 활용하는 CoD 요약 파이프라인을 구축하게 됩니다.

​

환경 설정

먼저 환경을 설정하고 필요한 라이브러리를 불러오겠습니다:

!pip install -qU anthropic weave pydantic requests PyPDF2 set-env-colab-kaggle-dotenv

Anthropic API 키를 발급받으려면:

1. https://www.anthropic.com 에서 계정을 생성합니다.
2. 계정 설정에서 API 섹션으로 이동합니다.
3. 새 API 키를 생성합니다.
4. .env 파일에 API 키를 안전하게 저장합니다.

import io
import os
from datetime import datetime, timezone

import anthropic
import requests
from pydantic import BaseModel
from PyPDF2 import PdfReader
from set_env import set_env

import weave

set_env("WANDB_API_KEY")
set_env("ANTHROPIC_API_KEY")

weave.init("summarization-chain-of-density-cookbook")
anthropic_client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

우리는 실험을 추적하기 위해 Weave를 사용하고, 텍스트 생성을 위해 Anthropic의 Claude 모델을 사용합니다. weave.init(<project name>) 호출은 요약 작업을 위한 새로운 Weave 프로젝트를 초기화합니다.

​

ArxivPaper 모델 정의

데이터를 표현하기 위한 간단한 ArxivPaper 클래스를 정의하겠습니다:

# ArxivPaper 모델 정의
class ArxivPaper(BaseModel):
    entry_id: str
    updated: datetime
    published: datetime
    title: str
    authors: list[str]
    summary: str
    pdf_url: str

# 샘플 ArxivPaper 생성
arxiv_paper = ArxivPaper(
    entry_id="http://arxiv.org/abs/2406.04744v1",
    updated=datetime(2024, 6, 7, 8, 43, 7, tzinfo=timezone.utc),
    published=datetime(2024, 6, 7, 8, 43, 7, tzinfo=timezone.utc),
    title="CRAG -- Comprehensive RAG Benchmark",
    authors=["Xiao Yang", "Kai Sun", "Hao Xin"],  # 간결성을 위해 생략
    summary="Retrieval-Augmented Generation (RAG) has recently emerged as a promising solution...",  # 생략
    pdf_url="https://arxiv.org/pdf/2406.04744",
)

이 클래스는 ArXiv 논문의 메타데이터와 내용을 담고 있으며, 요약 파이프라인의 입력으로 사용됩니다.

​

PDF 콘텐츠 불러오기

전체 논문 내용을 다루기 위해 PDF에서 텍스트를 로드하고 추출하는 함수를 추가하겠습니다:

@weave.op()
def load_pdf(pdf_url: str) -> str:
    # PDF 다운로드
    response = requests.get(pdf_url)
    pdf_file = io.BytesIO(response.content)

    # PDF 읽기
    pdf_reader = PdfReader(pdf_file)

    # 모든 페이지에서 텍스트 추출
    text = ""
    for page in pdf_reader.pages:
        text += page.extract_text()

    return text

​

Chain of Density 요약 구현

이제 Weave 연산을 사용하여 핵심 CoD 요약 로직을 구현해 보겠습니다:

# Chain of Density Summarization
@weave.op()
def summarize_current_summary(
    document: str,
    instruction: str,
    current_summary: str = "",
    iteration: int = 1,
    model: str = "claude-3-sonnet-20240229",
):
    prompt = f"""
    Document: {document}
    Current summary: {current_summary}
    Instruction to focus on: {instruction}
    Iteration: {iteration}

    Generate an increasingly concise, entity-dense, and highly technical summary from the provided document that specifically addresses the given instruction.
    """
    response = anthropic_client.messages.create(
        model=model, max_tokens=4096, messages=[{"role": "user", "content": prompt}]
    )
    return response.content[0].text

@weave.op()
def iterative_density_summarization(
    document: str,
    instruction: str,
    current_summary: str,
    density_iterations: int,
    model: str = "claude-3-sonnet-20240229",
):
    iteration_summaries = []
    for iteration in range(1, density_iterations + 1):
        current_summary = summarize_current_summary(
            document, instruction, current_summary, iteration, model
        )
        iteration_summaries.append(current_summary)
    return current_summary, iteration_summaries

@weave.op()
def final_summary(
    instruction: str, current_summary: str, model: str = "claude-3-sonnet-20240229"
):
    prompt = f"""
    Given this summary: {current_summary}
    And this instruction to focus on: {instruction}
    Create an extremely dense, final summary that captures all key technical information in the most concise form possible, while specifically addressing the given instruction.
    """
    return (
        anthropic_client.messages.create(
            model=model, max_tokens=4096, messages=[{"role": "user", "content": prompt}]
        )
        .content[0]
        .text
    )

@weave.op()
def chain_of_density_summarization(
    document: str,
    instruction: str,
    current_summary: str = "",
    model: str = "claude-3-sonnet-20240229",
    density_iterations: int = 2,
):
    current_summary, iteration_summaries = iterative_density_summarization(
        document, instruction, current_summary, density_iterations, model
    )
    final_summary_text = final_summary(instruction, current_summary, model)
    return {
        "final_summary": final_summary_text,
        "accumulated_summary": current_summary,
        "iteration_summaries": iteration_summaries,
    }

각 함수가 하는 일은 다음과 같습니다:

• summarize_current_summary: 현재 상태를 기반으로 한 번의 요약 단계를 생성합니다.
• iterative_density_summarization: summarize_current_summary를 여러 번 호출하여 CoD 기법을 적용합니다.
• chain_of_density_summarization: 전체 요약 프로세스를 오케스트레이션하고 결과를 반환합니다.

@weave.op() 데코레이터를 사용하면 Weave가 이 함수들의 입력, 출력, 실행을 추적하도록 할 수 있습니다.

​

Weave Model 생성

이제 요약 파이프라인을 Weave Model로 감싸서 만들어 보겠습니다:

# Weave Model
class ArxivChainOfDensityPipeline(weave.Model):
    model: str = "claude-3-sonnet-20240229"
    density_iterations: int = 3

    @weave.op()
    def predict(self, paper: ArxivPaper, instruction: str) -> dict:
        text = load_pdf(paper.pdf_url)
        result = chain_of_density_summarization(
            text,
            instruction,
            model=self.model,
            density_iterations=self.density_iterations,
        )
        return result

이 ArxivChainOfDensityPipeline 클래스는 요약 로직을 Weave Model로 캡슐화하여 다음과 같은 주요 이점을 제공합니다:

1. 자동 실험 추적: Weave가 모델의 각 run마다 입력, 출력, 파라미터를 자동으로 캡처합니다.
2. 버전 관리: 모델의 속성이나 코드에 대한 변경 사항이 자동으로 버전 관리되어, 요약 파이프라인이 시간에 따라 어떻게 발전하는지에 대한 명확한 이력을 생성합니다.
3. 재현성: 버전 관리와 추적을 통해 이전의 어떤 결과나 요약 파이프라인 설정도 쉽게 재현할 수 있습니다.
4. 하이퍼파라미터 관리: model과 density_iterations 같은 모델 속성이 명확하게 정의되고 서로 다른 run 전반에 걸쳐 추적되므로, 실험을 수월하게 진행할 수 있습니다.
5. Weave 생태계와의 인테그레이션: weave.Model을 사용하면 평가 및 서빙 기능과 같은 다른 Weave 도구와 매끄럽게 인테그레이션할 수 있습니다.

​

평가 메트릭 구현

요약의 품질을 평가하기 위해 간단한 평가 메트릭을 구현합니다.

import json

@weave.op()
def evaluate_summary(
    summary: str, instruction: str, model: str = "claude-3-sonnet-20240229"
) -> dict:
    prompt = f"""
    Summary: {summary}
    Instruction: {instruction}

    Evaluate the summary based on the following criteria:
    1. Relevance (1-5): How well does the summary address the given instruction?
    2. Conciseness (1-5): How concise is the summary while retaining key information?
    3. Technical Accuracy (1-5): How accurately does the summary convey technical details?

    Your response MUST be in the following JSON format:
    {{
        "relevance": {{
            "score": <int>,
            "explanation": "<string>"
        }},
        "conciseness": {{
            "score": <int>,
            "explanation": "<string>"
        }},
        "technical_accuracy": {{
            "score": <int>,
            "explanation": "<string>"
        }}
    }}

    Ensure that the scores are integers between 1 and 5, and that the explanations are concise.
    """
    response = anthropic_client.messages.create(
        model=model, max_tokens=1000, messages=[{"role": "user", "content": prompt}]
    )
    print(response.content[0].text)

    eval_dict = json.loads(response.content[0].text)

    return {
        "relevance": eval_dict["relevance"]["score"],
        "conciseness": eval_dict["conciseness"]["score"],
        "technical_accuracy": eval_dict["technical_accuracy"]["score"],
        "average_score": sum(eval_dict[k]["score"] for k in eval_dict) / 3,
        "evaluation_text": response.content[0].text,
    }

이러한 평가 함수는 Claude 모델을 사용해 생성된 요약의 관련성, 간결성, 기술적 정확성을 기준으로 품질을 평가합니다.

​

Weave Dataset 생성 및 평가 실행

파이프라인을 평가하기 위해 Weave Dataset을 생성하고 평가를 실행합니다.

# Weave Dataset 생성
dataset = weave.Dataset(
    name="arxiv_papers",
    rows=[
        {
            "paper": arxiv_paper,
            "instruction": "What was the approach to experimenting with different data mixtures?",
        },
    ],
)

weave.publish(dataset)

이번 평가에서는 LLM-as-a-judge 접근 방식을 사용할 것입니다. 이 기법은 한 모델이나 시스템이 생성한 출력물의 품질을 평가하기 위해 다른 언어 모델을 활용하는 방법입니다. 전통적인 메트릭으로는 한계가 있을 수 있는 작업에서 LLM의 이해 및 추론 능력을 활용해 보다 정교한 평가를 제공할 수 있습니다.

# 스코어러 함수 정의
@weave.op()
def quality_scorer(instruction: str, output: dict) -> dict:
    result = evaluate_summary(output["final_summary"], instruction)
    return result
python
# 평가 실행
evaluation = weave.Evaluation(dataset=dataset, scorers=[quality_scorer])
arxiv_chain_of_density_pipeline = ArxivChainOfDensityPipeline()
results = await evaluation.evaluate(arxiv_chain_of_density_pipeline)

이 코드는 예시 ArXiv 논문으로 데이터셋을 생성하고, 품질 스코어러를 정의한 뒤, 요약 파이프라인에 대한 평가를 수행합니다.

​

결론

이 예제에서는 Weave를 사용해 ArXiv 논문을 위한 Chain of Density 요약 파이프라인을 구현하는 방법을 보여주었습니다. 다음과 같은 내용을 다루었습니다:

1. 요약 과정의 각 단계에 대해 Weave operation을 생성하는 방법
2. 파이프라인을 Weave Model로 래핑하여 손쉽게 추적 및 평가하는 방법
3. Weave operation을 사용해 커스텀 평가 메트릭을 구현하는 방법
4. 데이터셋을 생성하고 파이프라인을 평가하는 방법

Weave의 매끄러운 인테그레이션을 통해 요약 과정 전반에 걸쳐 입력, 출력, 중간 단계를 모두 추적할 수 있으므로, LLM 애플리케이션을 디버깅하고 최적화하며 평가하기가 더 쉬워집니다. 이 예제를 확장해 더 큰 데이터셋을 처리하거나, 더 정교한 평가 메트릭을 구현하거나, 다른 LLM 워크플로우와 인테그레이션할 수 있습니다.

W&B에서 전체 리포트 보기