고급 도구 사용 튜토리얼

고급 도구 사용 튜토리얼 (Advanced Tool Use)

고급 도구 사용을 위한 DSPy 에이전트를 만들고 프롬프트를 최적화하는 간단한 예시를 함께 살펴볼게요. 까다로운 벤치마크인 ToolHop을 대상으로 하되, 원래 논문보다 훨씬 엄격한 평가 기준을 적용해 볼 거예요.

설치는 pip install -U dspy로 최신 DSPy를 받아서 따라오시면 되고, 추가로 pip install func_timeout datasets도 필요해요. 이 튜토리얼은 dspy.SIMBA를 쓰는데, 이건 numpy가 필요해서 pip install dspy[numpy]로 설치해야 해요.

권장사항: 내부에서 무슨 일이 벌어지는지 이해하려면 MLflow Tracing을 설정해 두세요.

출처: Advanced Tool Use

MLflow DSPy 통합

MLflow는 DSPy와 기본적으로 통합되는 LLMOps 도구로, 설명 가능성과 실험 추적을 제공해요. 이 튜토리얼에서 MLflow를 쓰면 프롬프트와 최적화 진행 상황을 trace로 시각화해 DSPy의 동작을 더 잘 이해할 수 있어요. 위 샘플 코드를 참고해 MLflow를 설치하고(%pip install mlflow>=2.20), mlflow ui --port 5000으로 UI를 띄우고, mlflow.set_tracking_uri("http://localhost:5000")과 mlflow.dspy.autolog()로 연결하면 돼요. MLflow DSPy 문서에 더 자세한 내용이 있어요.

여기서는 새로 나온 실험용 dspy.SIMBA 프롬프트 옵티마이저를 보여줄게요. 큰 LLM과 어려운 작업에서 특히 강력한 경향이 있어요. 이걸로 우리 에이전트의 정확도를 35%에서 60%까지 끌어올릴 거예요.

모델 설정과 데이터 준비

import dspy
import orjson
import random

gpt4o = dspy.LM("openai/gpt-4o", temperature=0.7)
dspy.configure(lm=gpt4o)

이제 데이터를 내려받을게요.

from dspy.utils import download

download("https://huggingface.co/datasets/bytedance-research/ToolHop/resolve/main/data/ToolHop.json")

data = orjson.loads(open("ToolHop.json").read())
random.Random(0).shuffle(data)

ToolHop 작업의 흥미로운 점은 에이전트가 요청마다 고유한 도구(함수) 집합을 받는다는 거예요. 그래서 실제로 어떤 도구든 효과적으로 쓰는 방법을 학습해야 해요. 정리된 예시 집합을 준비할게요.

import re
import inspect

examples = []
fns2code = {}

def finish(answer: str):
    """Conclude the trajectory and return the final answer."""
    return answer

for datapoint in data:
    func_dict = {}
    for func_code in datapoint["functions"]:
        cleaned_code = func_code.rsplit("\n\n# Example usage", 1)[0]
        fn_name = re.search(r"^\s*def\s+([a-zA-Z0-9_]+)\s*\(", cleaned_code)
        fn_name = fn_name.group(1) if fn_name else None

        if not fn_name:
            continue

        local_vars = {}
        exec(cleaned_code, {}, local_vars)
        fn_obj = local_vars.get(fn_name)

        if callable(fn_obj):
            func_dict[fn_name] = fn_obj
            assert fn_obj not in fns2code, f"Duplicate function found: {fn_name}"
            fns2code[fn_obj] = (fn_name, cleaned_code)

    func_dict["finish"] = finish

    example = dspy.Example(question=datapoint["question"], answer=datapoint["answer"], functions=func_dict)
    examples.append(example.with_inputs("question", "functions"))

trainset, devset, testset = examples[:100], examples[100:400], examples[400:]

각 데이터 포인트의 함수 코드를 파싱해 실행 가능한 함수 객체로 만들고, finish 함수를 더해 trajectory를 끝낼 수 있게 해요. train 100, dev 300, test는 나머지로 나눠요.

헬퍼와 metric 정의

몇 가지 헬퍼를 정의할게요. 여기서 metric은 원래 논문보다 훨씬 엄격해요. 정규화 후 예측이 정답과 정확히 일치해야 해요. 또 두 번째로 엄격한 점은, 효율적인 배포를 위해 에이전트가 총 5단계만 밟도록 제한하는 거예요.

from func_timeout import func_set_timeout

def wrap_function_with_timeout(fn):
    @func_set_timeout(10)
    def wrapper(*args, **kwargs):
        try:
            return {"return_value": fn(*args, **kwargs), "errors": None}
        except Exception as e:
            return {"return_value": None, "errors": str(e)}

    return wrapper

def fn_metadata(func):
    signature = inspect.signature(func)
    docstring = inspect.getdoc(func) or "No docstring."
    return dict(function_name=func.__name__, arguments=str(signature), docstring=docstring)

def metric(example, pred, trace=None):
    gold = str(example.answer).rstrip(".0").replace(",", "").lower()
    pred = str(pred.answer).rstrip(".0").replace(",", "").lower()
    return pred == gold  # stricter than the original paper's metric!

evaluate = dspy.Evaluate(devset=devset, metric=metric, num_threads=24, display_progress=True, display_table=0, max_errors=999)

wrap_function_with_timeout은 각 함수 호출에 10초 타임아웃을 걸고 결과나 오류를 dict로 감싸요. fn_metadata는 함수 시그니처와 docstring을 뽑아요. metric은 정규화 후 정확히 일치하면 참을 돌려줘요.

에이전트 정의

에이전트의 핵심은 ReAct 루프예요. 모델이 지금까지의 trajectory와 호출 가능한 함수 집합을 보고, 다음에 부를 도구를 결정하죠. 최종 에이전트를 빠르게 유지하려고 max_steps를 5로 제한하고, 각 함수 호출은 타임아웃으로 실행해요.

class Agent(dspy.Module):
    def __init__(self, max_steps=5):
        self.max_steps = max_steps
        instructions = "For the final answer, produce short (not full sentence) answers in which you format dates as YYYY-MM-DD, names as Firstname Lastname, and numbers without leading 0s."
        signature = dspy.Signature('question, trajectory, functions -> next_selected_fn, args: dict[str, Any]', instructions)
        self.react = dspy.ChainOfThought(signature)

    def forward(self, question, functions):
        tools = {fn_name: fn_metadata(fn) for fn_name, fn in functions.items()}
        trajectory = []

        for _ in range(self.max_steps):
            pred = self.react(question=question, trajectory=trajectory, functions=tools)
            selected_fn = pred.next_selected_fn.strip('"').strip("'")
            fn_output = wrap_function_with_timeout(functions[selected_fn])(**pred.args)
            trajectory.append(dict(reasoning=pred.reasoning, selected_fn=selected_fn, args=pred.args, **fn_output))

            if selected_fn == "finish":
                break

        return dspy.Prediction(answer=fn_output.get("return_value", ''), trajectory=trajectory)

forward는 최대 max_steps만큼 반복하며, 매 단계 ChainOfThought로 다음에 부를 함수와 인자를 고르고, 타임아웃 래퍼로 실행해 trajectory에 기록해요. finish가 선택되면 루프를 끝내요.

제로샷 평가와 SIMBA 최적화

먼저 GPT-4o 기반 에이전트를 dev set에서 평가해 볼게요.

agent = Agent()
evaluate(agent)

결과: Average Metric: 105.0 / 300 (35.0%) — 35% 정확도예요.

이제 dspy.SIMBA로 에이전트를 최적화할게요. SIMBA는 Stochastic Introspective Mini-Batch Ascent의 약자예요. 이 프롬프트 옵티마이저는 우리 에이전트 같은 임의의 DSPy 프로그램을 받아들이고, 프롬프트 지시문이나 few-shot 예시를 점진적으로 개선하는 미니배치 시퀀스를 진행해요.

simba = dspy.SIMBA(metric=metric, max_steps=12, max_demos=10)
optimized_agent = simba.compile(agent, trainset=trainset, seed=6793115)

최적화를 마친 뒤 다시 평가하면 상당한 개선이 보여요. 무려 71% 상대 이득으로 정확도가 60%까지 올라가요.

evaluate(optimized_agent)

결과: Average Metric: 182.0 / 300 (60.7%) — 60.67% 정확도예요.

더 알아보기 (Learn more)