Weave로 컴퓨터 비전 파이프라인 트레이스 및 평가하기

# 의존성 설치
!pip install openai weave -q
python
import json
import os

from google.colab import userdata
from openai import OpenAI

import weave
python
# API 키 조회
os.environ["OPENAI_API_KEY"] = userdata.get(
    "OPENAI_API_KEY"
)  # 왼쪽 메뉴에서 Colab 환경 시크릿으로 키를 설정하세요
os.environ["WANDB_API_KEY"] = userdata.get("WANDB_API_KEY")

# 프로젝트 이름 설정
# PROJECT 값을 프로젝트 이름으로 변경하세요
PROJECT = "vlm-handwritten-ner"

# Weave 프로젝트 초기화
weave.init(PROJECT)

# Weave로 프롬프트 객체 생성
prompt = """
Extract all readable text from this image. Format the extracted entities as a valid JSON.
Do not return any extra text, just the JSON. Do not include ```json```
Use the following format:
{"Patient Name": "James James","Date": "4/22/2025","Patient ID": "ZZZZZZZ123","Group Number": "3452542525"}
"""
system_prompt = weave.StringPrompt(prompt)
# Weave에 프롬프트 게시
weave.publish(system_prompt, name="NER-prompt")

better_prompt = """
You are a precision OCR assistant. Given an image of patient information, extract exactly these fields into a single JSON object—and nothing else:

- Patient Name
- Date (MM/DD/YYYY)
- Patient ID
- Group Number

Validation rules:
1. Date must match MM/DD/YY; if not, set Date to "".
2. Patient ID must be alphanumeric; if unreadable, set to "".
3. Always zero-pad months and days (e.g. "04/07/25").
4. Omit any markup, commentary, or code fences.
5. Return strictly valid JSON with only those four keys.

Do not return any extra text, just the JSON. Do not include ```json```
Example output:
{"Patient Name":"James James","Date":"04/22/25","Patient ID":"ZZZZZZZ123","Group Number":"3452542525"}
"""
# 프롬프트 수정
system_prompt = weave.StringPrompt(better_prompt)
# 수정된 프롬프트를 Weave에 게시
weave.publish(system_prompt, name="NER-prompt")

# 다음 Weave 프로젝트에서 데이터셋을 조회합니다
dataset = weave.ref(
    "weave://wandb-smle/vlm-handwritten-ner/object/NER-eval-dataset:G8MEkqWBtvIxPYAY23sXLvqp8JKZ37Cj0PgcG19dGjw"
).get()

# 데이터셋에서 특정 예시를 가져옵니다
example_image = dataset.rows[3]["image_base64"]

# example_image를 표시합니다
from IPython.display import HTML, display

html = f'<img src="{example_image}" style="max-width: 100%; height: auto;">'
display(HTML(html))

# GPT-4-Vision을 사용하는 추적 가능한 함수
def extract_named_entities_from_image(image_base64) -> dict:
    # LLM 클라이언트 초기화
    client = OpenAI()

    # 지시 프롬프트 설정
    # Weave에 저장된 프롬프트를 사용할 수도 있습니다: weave.ref("weave://wandb-smle/vlm-handwritten-ner/object/NER-prompt:FmCv4xS3RFU21wmNHsIYUFal3cxjtAkegz2ylM25iB8").get().content.strip()
    prompt = better_prompt

    response = client.responses.create(
        model="gpt-4.1",
        input=[
            {
                "role": "user",
                "content": [
                    {"type": "input_text", "text": prompt},
                    {
                        "type": "input_image",
                        "image_url": image_base64,
                    },
                ],
            }
        ],
    )

    return response.output_text

# evaluations를 위한 NER 함수
@weave.op()
def named_entity_recognation(image_base64, id):
    result = {}
    try:
        # 1) vision op 호출 후 JSON 문자열 반환
        output_text = extract_named_entities_from_image(image_base64)

        # 2) JSON을 정확히 한 번만 파싱
        result = json.loads(output_text)

        print(f"Processed: {str(id)}")
    except Exception as e:
        print(f"Failed to process {str(id)}: {e}")
    return result

# 출력 결과
results = []

# 데이터셋의 모든 이미지를 순회
for row in dataset.rows:
    result = named_entity_recognation(row["image_base64"], str(row["id"]))
    result["image_id"] = str(row["id"])
    results.append(result)

# 모든 결과를 JSON 파일로 저장
output_file = "processing_results.json"
with open(output_file, "w") as f:
    json.dump(results, f, indent=2)

print(f"Results saved to: {output_file}")

# scorer 실행을 추적하기 위해 weave.op()를 추가합니다
@weave.op()
def check_for_missing_fields_programatically(model_output):
    # 모든 항목에 필요한 키
    required_fields = {"Patient Name", "Date", "Patient ID", "Group Number"}

    for key in required_fields:
        if (
            key not in model_output
            or model_output[key] is None
            or str(model_output[key]).strip() == ""
        ):
            return False  # 이 항목에 누락되거나 비어 있는 필드가 있습니다

    return True  # 모든 필수 필드가 존재하며 비어 있지 않습니다

# LLM-as-a-judge의 시스템 프롬프트

eval_prompt = """
You are an OCR validation system. Your role is to assess whether the structured text extracted from an image accurately reflects the information in that image.
Only validate the structured text and use the image as your source of truth.

Expected input text format:
{"Patient Name": "First Last", "Date": "04/23/25", "Patient ID": "131313JJH", "Group Number": "35453453"}

Evaluation criteria:
- All four fields must be present.
- No field should be empty or contain placeholder/malformed values.
- The "Date" should be in MM/DD/YY format (e.g., "04/07/25") (zero padding the date is allowed)

Scoring:
- Return: {"Correct": true, "Reason": ""} if **all fields** match the information in the image and formatting is correct.
- Return: {"Correct": false, "Reason": "EXPLANATION"} if **any** field is missing, empty, incorrect, or mismatched.

Output requirements:
- Respond with a valid JSON object only.
- "Correct" must be a JSON boolean: true or false (not a string or number).
- "Reason" must be a short, specific string indicating all the problem — e.g., "Patient Name mismatch", "Date not zero-padded", or "Missing Group Number".
- Do not return any additional explanation or formatting.

Your response must be exactly one of the following:
{"Correct": true, "Reason": null}
OR
{"Correct": false, "Reason": "EXPLANATION_HERE"}
"""

# Scorer 실행을 추적하기 위해 weave.op() 추가
@weave.op()
def check_for_missing_fields_with_llm(model_output, image_base64):
    client = OpenAI()
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "developer", "content": [{"text": eval_prompt, "type": "text"}]},
            {
                "role": "user",
                "content": [
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": image_base64,
                        },
                    },
                    {"type": "text", "text": str(model_output)},
                ],
            },
        ],
        response_format={"type": "json_object"},
    )
    response = json.loads(response.choices[0].message.content)
    return response

evaluation = weave.Evaluation(
    dataset=dataset,
    scorers=[
        check_for_missing_fields_with_llm,
        check_for_missing_fields_programatically,
    ],
    name="Evaluate_4.1_NER",
)

print(await evaluation.evaluate(named_entity_recognation))