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. 반복할 때마다 엔티티와 기술적 세부 사항의 밀도를 높입니다.

이 접근 방식은 상세 정보 보존이 중요한 과학 논문이나 기술 문서를 요약하는 데 특히 유용합니다.

​

왜 Weave를 사용하나요?

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

1. LLM 파이프라인 추적: Weave 를 사용하여 요약 프로세스의 입력, 출력 및 중간 단계를 자동으로 로깅합니다.
2. LLM 출력 평가: Weave 의 내장 도구를 사용하여 요약본에 대한 엄격하고 객관적인 평가를 생성합니다.
3. 구성 가능한 작업(Operations) 구축: 요약 파이프라인의 여러 부분에서 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. API 키를 .env 파일에 안전하게 저장합니다.

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 요약
@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 는 모델의 각 실행에 대한 입력, 출력 및 파라미터를 캡처합니다.
2. 버전 관리: 모델의 속성이나 코드 변경 사항이 자동으로 버전 관리되어, 요약 파이프라인이 시간이 지남에 따라 어떻게 발전하는지 명확한 이력을 생성합니다.
3. 재현성: 버전 관리 및 추적을 통해 요약 파이프라인의 이전 결과나 설정을 쉽게 재현할 수 있습니다.
4. 하이퍼파라미터 관리: 모델 속성(model 및 density_iterations 등)이 명확하게 정의되고 여러 실행에 걸쳐 추적되어 실험을 용이하게 합니다.
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

# 평가 실행
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 작업을 생성합니다.
2. 쉬운 추적 및 평가를 위해 파이프라인을 Weave Model로 감쌉니다.
3. Weave 작업을 사용하여 사용자 정의 평가 메트릭을 구현합니다.
4. 데이터셋을 생성하고 파이프라인의 평가를 실행합니다.

Weave 의 원활한 통합을 통해 요약 프로세스 전반에 걸쳐 입력, 출력 및 중간 단계를 추적할 수 있으며, 이를 통해 LLM 애플리케이션을 더 쉽게 디버깅, 최적화 및 평가할 수 있습니다. 이 예제를 확장하여 더 큰 데이터셋을 처리하거나, 더 정교한 평가 메트릭을 구현하거나, 다른 LLM 워크플로우와 통합할 수 있습니다.

W&B에서 전체 Report 보기