튜토리얼: 고급 도구 사용
튜토리얼: 고급 도구 사용 (Advanced Tool Use)
고급 도구 사용을 위한 DSPy 에이전트를 구축하고 프롬프트 최적화하는 quick example을 함께 살펴볼게요. 우리가 다룰 과제는 도전적인 ToolHop이지만, 여기서는 원래 논문보다 훨씬 더 엄격한 평가 기준을 적용할 거예요.
최신 DSPy를 pip install -U dspy로 설치하고 따라와 보세요. 또한 pip install func_timeout datasets도 필요해요. 이 튜토리얼은 dspy.SIMBA를 사용하며, numpy가 필요하므로 pip install dspy[numpy]를 실행해야 합니다.
출처: 문서
본문
권장: 내부에서 무슨 일이 일어나는지 이해하려면 MLflow Tracing을 설정하세요.
MLflow DSPy 통합
MLflow는 DSPy와 네이티브로 통합되는 LLMOps 도구로, 설명 가능성(explainability)과 실험 추적을 제공해요. 이 튜토리얼에서는 MLflow를 사용해 프롬프트와 최적화 진행 상황을 트레이스로 시각화해 DSPy의 동작을 더 잘 이해할 수 있어요. 아래 네 단계를 따라 MLflow를 쉽게 설정할 수 있습니다.
- MLflow 설치
%pip install mlflow>=2.20
- 별도 터미널에서 MLflow UI 시작
mlflow ui --port 5000
- 노트북을 MLflow에 연결
import mlflow
mlflow.set_tracking_uri("http://localhost:5000")
mlflow.set_experiment("DSPy")
- 트레이싱 활성화.
mlflow.dspy.autolog()
통합에 대해 더 알아보려면 MLflow DSPy 문서도 방문하세요.
이 튜토리얼에서는 실험적인 새 프롬프트 옵티마이저 dspy.SIMBA를 시연할 거예요. 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)
Downloading 'ToolHop.json'...
그다음 정리된 예제 집합을 준비할게요. 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:]
그리고 작업을 위한 헬퍼 몇 개를 정의할게요. 여기서 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)
이제 에이전트를 정의해 볼게요! 에이전트의 핵심은 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)
바로 사용해 보면서 GPT-4o 기반 에이전트를 dev 세트에서 평가해 보죠.
agent = Agent()
evaluate(agent)
2025/03/23 21:46:10 INFO dspy.evaluate.evaluate: Average Metric: 105.0 / 300 (35.0%)
35.0
이제 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)
2025/03/23 21:46:21 INFO dspy.evaluate.evaluate: Average Metric: 182.0 / 300 (60.7%)
60.67