AdalFlow 실전·API — 튜토리얼 인덱스
AdalFlow 실전·API
AdalFlow 튜토리얼 인덱스를 따라 실제 사용법을 볼게요. 기본 파이프라인 구축부터 에이전트·평가·최적화까지 단계적으로 익힐 수 있어요.
튜토리얼 개요
- 태스크 파이프라인 구축 — 챗봇, RAG, 구조화 출력
- 에이전트 — 도구·멀티스텝 워크플로
- 평가 — LLM 평가
- 훈련·옵티마이저 — 자동 프롬프트 최적화
헬로 월드 에이전트 예
from adalflow import Agent, Runner
from adalflow.components.model_client.openai_client import OpenAIClient
from adalflow.core.types import (
ToolCallActivityRunItem,
RunItemStreamEvent,
ToolCallRunItem,
ToolOutputRunItem,
FinalOutputItem
)
import asyncio
# 도구 정의
def calculator(expression: str) -> str:
# Evaluate a mathematical expression.
try:
result = eval(expression)
return f"The result of {expression} is {result}"
except Exception as e:
return f"Error: {e}"
async def web_search(query: str = "what is the weather in SF today?") -> str:
await asyncio.sleep(0.5)
return "San Francisco will be mostly cloudy today..."
# 에이전트 생성
agent = Agent(
name="MyAgent",
tools=[calculator, web_search],
model_client=OpenAIClient(),
model_kwargs={"model": "gpt-4o", "temperature": 0.3},
max_steps=5
)
runner = Runner(agent=agent)
동기 호출
result = runner.call(
prompt_kwargs={"input_str": "Calculate 15 * 7 + 23 and count to 5"}
)
print(result.answer)
비동기·스트리밍
result = await runner.acall(
prompt_kwargs={"input_str": "What's the weather in SF and calculate 42 * 3"}
)
streaming_result = runner.astream(
prompt_kwargs={"input_str": "Calculate 100 + 50 and count to 3"},
)
async for event in streaming_result.stream_events():
if isinstance(event, RunItemStreamEvent):
if isinstance(event.item, ToolCallRunItem):
print(f"🔧 Calling: {event.item.data.name}")
OPENAI_API_KEY 환경 변수 설정이 필요해요.