메인 콘텐츠로 건너뛰기
이 노트북은 대화형 노트북입니다. 로컬에서 실행하거나 아래 링크를 통해 사용할 수 있습니다:

​

​
다중 에이전트 시스템을 위한 구조화된 출력

OpenAI는 사용자가 과하게 제약을 거는 프롬프트 없이도, 모델이 항상 제공된 JSON Schema를 준수하는 형식의 응답을 생성하도록 보장할 수 있게 해 주는 Structured Outputs 기능을 출시했습니다. Structured Outputs 기능을 사용하면, 잘못 포맷된 응답을 검증하거나 재시도할 필요가 없습니다. 새로운 매개변수 strict: true를 사용하면, 응답이 제공된 스키마를 반드시 따르도록 보장할 수 있습니다. 다중 에이전트 시스템에서 구조화된 출력을 사용하면 에이전트 간에 일관되고 쉽게 처리 가능한 데이터를 보장하여 통신을 향상시킵니다. 또한 명시적인 거절을 가능하게 하여 안전성을 높이고, 재시도나 검증이 필요 없도록 하여 성능을 향상시킵니다. 이를 통해 상호작용이 단순해지고 전체 시스템 효율성이 향상됩니다. 이 튜토리얼에서는 다중 에이전트 시스템에서 구조화된 출력을 활용하고 이를 Weave로 추적하는 방법을 보여줍니다.
출처: 이 쿡북은 OpenAI의 Structured Outputs 샘플 코드를 기반으로 하며, Weave를 사용한 시각화를 개선하기 위해 일부 수정이 추가되었습니다.

​
필요한 종속성 설치

이 튜토리얼에는 다음 라이브러리가 필요합니다.
  • 멀티 에이전트 시스템을 만들기 위한 OpenAI
  • LLM 워크플로우를 추적하고 프롬프트 전략을 평가하기 위한 Weave
!pip install -qU openai weave wandb
python
%%capture
# openai 버그 수정을 위한 임시 해결 방법:
# TypeError: Client.__init__() got an unexpected keyword argument 'proxies'
# 참고: https://community.openai.com/t/error-with-openai-1-56-0-client-init-got-an-unexpected-keyword-argument-proxies/1040332/15
!pip install "httpx<0.28"
우리는 WANDB_API_KEY를 환경 변수에 설정하여 wandb.login()으로 쉽게 로그인할 수 있도록 합니다(이는 Colab에 secret으로 제공해야 합니다). name_of_wandb_project에 이 로그를 기록할 W&B 프로젝트를 설정합니다. NOTE: name_of_wandb_project는 트레이스를 기록할 팀을 지정하기 위해 {team_name}/{project_name} 형식으로도 설정할 수 있습니다. 그런 다음 weave.init()을 호출하여 Weave 클라이언트를 가져옵니다. OpenAI API를 사용할 것이므로 OpenAI API key도 필요합니다. OpenAI 플랫폼에서 가입하여 본인의 API key를 발급받을 수 있습니다(이 역시 Colab에 secret으로 제공해야 합니다). 
import base64
import json
import os
from io import BytesIO, StringIO

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import wandb
from google.colab import userdata
from openai import OpenAI

import weave
python
os.environ["WANDB_API_KEY"] = userdata.get("WANDB_API_KEY")
os.environ["OPENAI_API_KEY"] = userdata.get("OPENAI_API_KEY")

wandb.login()
name_of_wandb_project = "multi-agent-structured-output"
weave.init(name_of_wandb_project)

client = OpenAI()
MODEL = "gpt-4o-2024-08-06"

​
에이전트 설정

우리가 살펴볼 사용 사례는 데이터 분석 작업입니다. 먼저 4개 에이전트로 구성된 시스템을 설정해 보겠습니다:
  • 트리아지 에이전트: 어떤 에이전트를 호출할지 결정합니다
  • 데이터 전처리 에이전트: 예를 들어 데이터를 정리해 분석에 사용할 수 있도록 준비합니다
  • 데이터 분석 에이전트: 데이터에 대한 분석을 수행합니다
  • 데이터 시각화 에이전트: 분석 결과를 시각화하여 인사이트를 도출합니다 이제 각 에이전트에 대한 시스템 프롬프트를 정의하는 것부터 시작하겠습니다.
triaging_system_prompt = """You are a Triaging Agent. Your role is to assess the user's query and route it to the relevant agents. The agents available are:
- Data Processing Agent: Cleans, transforms, and aggregates data.
- Analysis Agent: Performs statistical, correlation, and regression analysis.
- Visualization Agent: Creates bar charts, line charts, and pie charts.

Use the send_query_to_agents tool to forward the user's query to the relevant agents. Also, use the speak_to_user tool to get more information from the user if needed."""

processing_system_prompt = """You are a Data Processing Agent. Your role is to clean, transform, and aggregate data using the following tools:
- clean_data
- transform_data
- aggregate_data"""

analysis_system_prompt = """You are an Analysis Agent. Your role is to perform statistical, correlation, and regression analysis using the following tools:
- stat_analysis
- correlation_analysis
- regression_analysis"""

visualization_system_prompt = """You are a Visualization Agent. Your role is to create bar charts, line charts, and pie charts using the following tools:
- create_bar_chart
- create_line_chart
- create_pie_chart"""
그런 다음 각 에이전트에 사용할 도구를 정의하겠습니다. 트리아지 에이전트를 제외한 나머지 에이전트는 각자의 역할에 특화된 도구를 갖추게 됩니다: 데이터 전처리 에이전트: 1. 데이터 정제, 2. 데이터 변환, 3. 데이터 집계 데이터 분석 에이전트: 1. 통계 분석, 2. 상관관계 분석, 3. 회귀 분석 데이터 시각화 에이전트: 1. 막대 차트 생성, 2. 선 차트 생성, 3. 파이 차트 생성 
triage_tools = [
    {
        "type": "function",
        "function": {
            "name": "send_query_to_agents",
            "description": "사용자 쿼리를 에이전트의 기능에 따라 관련 에이전트에 전송합니다.",
            "parameters": {
                "type": "object",
                "properties": {
                    "agents": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "쿼리를 전송할 에이전트 이름의 배열입니다.",
                    },
                    "query": {
                        "type": "string",
                        "description": "전송할 사용자 쿼리입니다.",
                    },
                },
                "required": ["agents", "query"],
            },
        },
        "strict": True,
    }
]

preprocess_tools = [
    {
        "type": "function",
        "function": {
            "name": "clean_data",
            "description": "중복을 제거하고 누락된 값을 처리하여 제공된 데이터를 정제합니다.",
            "parameters": {
                "type": "object",
                "properties": {
                    "data": {
                        "type": "string",
                        "description": "정제할 데이터셋입니다. JSON 또는 CSV와 같은 적절한 형식이어야 합니다.",
                    }
                },
                "required": ["data"],
                "additionalProperties": False,
            },
        },
        "strict": True,
    },
    {
        "type": "function",
        "function": {
            "name": "transform_data",
            "description": "지정된 규칙에 따라 데이터를 변환합니다.",
            "parameters": {
                "type": "object",
                "properties": {
                    "data": {
                        "type": "string",
                        "description": "변환할 데이터입니다. JSON 또는 CSV와 같은 적절한 형식이어야 합니다.",
                    },
                    "rules": {
                        "type": "string",
                        "description": "구조화된 형식으로 지정된 변환 규칙입니다.",
                    },
                },
                "required": ["data", "rules"],
                "additionalProperties": False,
            },
        },
        "strict": True,
    },
    {
        "type": "function",
        "function": {
            "name": "aggregate_data",
            "description": "지정된 열과 연산을 기준으로 데이터를 집계합니다.",
            "parameters": {
                "type": "object",
                "properties": {
                    "data": {
                        "type": "string",
                        "description": "집계할 데이터입니다. JSON 또는 CSV와 같은 적절한 형식이어야 합니다.",
                    },
                    "group_by": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "그룹화 기준 열입니다.",
                    },
                    "operations": {
                        "type": "string",
                        "description": "구조화된 형식으로 지정된 집계 연산입니다.",
                    },
                },
                "required": ["data", "group_by", "operations"],
                "additionalProperties": False,
            },
        },
        "strict": True,
    },
]

analysis_tools = [
    {
        "type": "function",
        "function": {
            "name": "stat_analysis",
            "description": "주어진 데이터셋에 대해 통계 분석을 수행합니다.",
            "parameters": {
                "type": "object",
                "properties": {
                    "data": {
                        "type": "string",
                        "description": "분석할 데이터셋입니다. JSON 또는 CSV와 같은 적절한 형식이어야 합니다.",
                    }
                },
                "required": ["data"],
                "additionalProperties": False,
            },
        },
        "strict": True,
    },
    {
        "type": "function",
        "function": {
            "name": "correlation_analysis",
            "description": "데이터셋의 변수 간 상관 계수를 계산합니다.",
            "parameters": {
                "type": "object",
                "properties": {
                    "data": {
                        "type": "string",
                        "description": "분석할 데이터셋입니다. JSON 또는 CSV와 같은 적절한 형식이어야 합니다.",
                    },
                    "variables": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "상관관계를 계산할 변수 목록입니다.",
                    },
                },
                "required": ["data", "variables"],
                "additionalProperties": False,
            },
        },
        "strict": True,
    },
    {
        "type": "function",
        "function": {
            "name": "regression_analysis",
            "description": "데이터셋에 대해 회귀 분석을 수행합니다.",
            "parameters": {
                "type": "object",
                "properties": {
                    "data": {
                        "type": "string",
                        "description": "분석할 데이터셋입니다. JSON 또는 CSV와 같은 적절한 형식이어야 합니다.",
                    },
                    "dependent_var": {
                        "type": "string",
                        "description": "회귀 분석의 종속 변수입니다.",
                    },
                    "independent_vars": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "독립 변수 목록입니다.",
                    },
                },
                "required": ["data", "dependent_var", "independent_vars"],
                "additionalProperties": False,
            },
        },
        "strict": True,
    },
]

visualization_tools = [
    {
        "type": "function",
        "function": {
            "name": "create_bar_chart",
            "description": "제공된 데이터로 막대 차트를 생성합니다.",
            "parameters": {
                "type": "object",
                "properties": {
                    "data": {
                        "type": "string",
                        "description": "막대 차트에 사용할 데이터입니다. JSON 또는 CSV와 같은 적절한 형식이어야 합니다.",
                    },
                    "x": {"type": "string", "description": "x축에 사용할 열입니다."},
                    "y": {"type": "string", "description": "y축에 사용할 열입니다."},
                },
                "required": ["data", "x", "y"],
                "additionalProperties": False,
            },
        },
        "strict": True,
    },
    {
        "type": "function",
        "function": {
            "name": "create_line_chart",
            "description": "제공된 데이터로 선 차트를 생성합니다.",
            "parameters": {
                "type": "object",
                "properties": {
                    "data": {
                        "type": "string",
                        "description": "선 차트에 사용할 데이터입니다. JSON 또는 CSV와 같은 적절한 형식이어야 합니다.",
                    },
                    "x": {"type": "string", "description": "x축에 사용할 열입니다."},
                    "y": {"type": "string", "description": "y축에 사용할 열입니다."},
                },
                "required": ["data", "x", "y"],
                "additionalProperties": False,
            },
        },
        "strict": True,
    },
    {
        "type": "function",
        "function": {
            "name": "create_pie_chart",
            "description": "제공된 데이터로 파이 차트를 생성합니다.",
            "parameters": {
                "type": "object",
                "properties": {
                    "data": {
                        "type": "string",
                        "description": "파이 차트에 사용할 데이터입니다. JSON 또는 CSV와 같은 적절한 형식이어야 합니다.",
                    },
                    "labels": {
                        "type": "string",
                        "description": "레이블에 사용할 열입니다.",
                    },
                    "values": {
                        "type": "string",
                        "description": "값에 사용할 열입니다.",
                    },
                },
                "required": ["data", "labels", "values"],
                "additionalProperties": False,
            },
        },
        "strict": True,
    },
]

​
Weave를 사용하여 멀티 에이전트 추적 활성화하기

다음과 같은 코드 로직을 작성해야 합니다:
  • 사용자 쿼리를 멀티 에이전트 시스템에 전달하는 로직
  • 멀티 에이전트 시스템의 내부 동작을 처리하는 로직
  • 도구 호출을 실행하는 로직
# 예시 쿼리

user_query = """
Below is some data. I want you to first remove the duplicates then analyze the statistics of the data as well as plot a line chart.

house_size (m3), house_price ($)
90, 100
80, 90
100, 120
90, 100
"""
사용자 쿼리로부터 우리가 호출해야 하는 도구는 clean_data, start_analysis, use_line_chart임을 알 수 있습니다. 이제 도구 호출을 수행하는 실행 함수를 정의하는 것부터 시작하겠습니다. Python 함수에 @weave.op() 데코레이터를 추가하면 언어 모델의 입력, 출력, 그리고 트레이스를 로깅하고 디버깅할 수 있습니다. 멀티 에이전트 시스템을 만들 때는 많은 함수가 필요하지만, 각 함수 위에 @weave.op()만 간단히 추가해 주면 충분합니다. 
@weave.op()
def clean_data(data):
    data_io = StringIO(data)
    df = pd.read_csv(data_io, sep=",")
    df_deduplicated = df.drop_duplicates()
    return df_deduplicated

@weave.op()
def stat_analysis(data):
    data_io = StringIO(data)
    df = pd.read_csv(data_io, sep=",")
    return df.describe()

@weave.op()
def plot_line_chart(data):
    data_io = StringIO(data)
    df = pd.read_csv(data_io, sep=",")

    x = df.iloc[:, 0]
    y = df.iloc[:, 1]

    coefficients = np.polyfit(x, y, 1)
    polynomial = np.poly1d(coefficients)
    y_fit = polynomial(x)

    plt.figure(figsize=(10, 6))
    plt.plot(x, y, "o", label="Data Points")
    plt.plot(x, y_fit, "-", label="Best Fit Line")
    plt.title("Line Chart with Best Fit Line")
    plt.xlabel(df.columns[0])
    plt.ylabel(df.columns[1])
    plt.legend()
    plt.grid(True)

    # 표시하기 전에 플롯을 BytesIO 버퍼에 저장
    buf = BytesIO()
    plt.savefig(buf, format="png")
    buf.seek(0)

    # 플롯 표시
    plt.show()

    # 데이터 URL을 위해 이미지를 base64로 인코딩
    image_data = buf.getvalue()
    base64_encoded_data = base64.b64encode(image_data)
    base64_string = base64_encoded_data.decode("utf-8")
    data_url = f"data:image/png;base64,{base64_string}"

    return data_url

# 도구를 실행하는 함수 정의
@weave.op()
def execute_tool(tool_calls, messages):
    for tool_call in tool_calls:
        tool_name = tool_call.function.name
        tool_arguments = json.loads(tool_call.function.arguments)

        if tool_name == "clean_data":
            # 데이터 정제 시뮬레이션
            cleaned_df = clean_data(tool_arguments["data"])
            cleaned_data = {"cleaned_data": cleaned_df.to_dict()}
            messages.append(
                {"role": "tool", "name": tool_name, "content": json.dumps(cleaned_data)}
            )
            print("Cleaned data: ", cleaned_df)
        elif tool_name == "transform_data":
            # 데이터 변환 시뮬레이션
            transformed_data = {"transformed_data": "sample_transformed_data"}
            messages.append(
                {
                    "role": "tool",
                    "name": tool_name,
                    "content": json.dumps(transformed_data),
                }
            )
        elif tool_name == "aggregate_data":
            # 데이터 집계 시뮬레이션
            aggregated_data = {"aggregated_data": "sample_aggregated_data"}
            messages.append(
                {
                    "role": "tool",
                    "name": tool_name,
                    "content": json.dumps(aggregated_data),
                }
            )
        elif tool_name == "stat_analysis":
            # 통계 분석 시뮬레이션
            stats_df = stat_analysis(tool_arguments["data"])
            stats = {"stats": stats_df.to_dict()}
            messages.append(
                {"role": "tool", "name": tool_name, "content": json.dumps(stats)}
            )
            print("Statistical Analysis: ", stats_df)
        elif tool_name == "correlation_analysis":
            # 상관 분석 시뮬레이션
            correlations = {"correlations": "sample_correlations"}
            messages.append(
                {"role": "tool", "name": tool_name, "content": json.dumps(correlations)}
            )
        elif tool_name == "regression_analysis":
            # 회귀 분석 시뮬레이션
            regression_results = {"regression_results": "sample_regression_results"}
            messages.append(
                {
                    "role": "tool",
                    "name": tool_name,
                    "content": json.dumps(regression_results),
                }
            )
        elif tool_name == "create_bar_chart":
            # 막대 차트 생성 시뮬레이션
            bar_chart = {"bar_chart": "sample_bar_chart"}
            messages.append(
                {"role": "tool", "name": tool_name, "content": json.dumps(bar_chart)}
            )
        elif tool_name == "create_line_chart":
            # 선 차트 생성 시뮬레이션
            line_chart = {"line_chart": plot_line_chart(tool_arguments["data"])}
            messages.append(
                {"role": "tool", "name": tool_name, "content": json.dumps(line_chart)}
            )
        elif tool_name == "create_pie_chart":
            # 파이 차트 생성 시뮬레이션
            pie_chart = {"pie_chart": "sample_pie_chart"}
            messages.append(
                {"role": "tool", "name": tool_name, "content": json.dumps(pie_chart)}
            )
    return messages
다음으로 각 하위 에이전트에 대한 툴 핸들러를 생성합니다. 이 핸들러는 모델에 전달되는 고유한 프롬프트와 툴 집합을 가집니다. 그런 다음 출력이 툴 호출을 실행하는 실행 함수로 전달됩니다. 
# 각 에이전트의 처리를 담당하는 함수 정의
@weave.op()
def handle_data_processing_agent(query, conversation_messages):
    messages = [{"role": "system", "content": processing_system_prompt}]
    messages.append({"role": "user", "content": query})

    response = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        temperature=0,
        tools=preprocess_tools,
    )

    conversation_messages.append(
        [tool_call.function for tool_call in response.choices[0].message.tool_calls]
    )
    execute_tool(response.choices[0].message.tool_calls, conversation_messages)

@weave.op()
def handle_analysis_agent(query, conversation_messages):
    messages = [{"role": "system", "content": analysis_system_prompt}]
    messages.append({"role": "user", "content": query})

    response = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        temperature=0,
        tools=analysis_tools,
    )

    conversation_messages.append(
        [tool_call.function for tool_call in response.choices[0].message.tool_calls]
    )
    execute_tool(response.choices[0].message.tool_calls, conversation_messages)

@weave.op()
def handle_visualization_agent(query, conversation_messages):
    messages = [{"role": "system", "content": visualization_system_prompt}]
    messages.append({"role": "user", "content": query})

    response = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        temperature=0,
        tools=visualization_tools,
    )

    conversation_messages.append(
        [tool_call.function for tool_call in response.choices[0].message.tool_calls]
    )
    execute_tool(response.choices[0].message.tool_calls, conversation_messages)
마지막으로 사용자 쿼리 처리를 담당하는 최상위 도구를 생성합니다. 이 함수는 사용자 쿼리를 입력받아 모델로부터 응답을 받고, 이를 실행하도록 다른 에이전트들에게 전달합니다. 
# 사용자 입력 및 트리아지 처리 함수
@weave.op()
def handle_user_message(user_query, conversation_messages=None):
    if conversation_messages is None:
        conversation_messages = []
    user_message = {"role": "user", "content": user_query}
    conversation_messages.append(user_message)

    messages = [{"role": "system", "content": triaging_system_prompt}]
    messages.extend(conversation_messages)

    response = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        temperature=0,
        tools=triage_tools,
    )

    conversation_messages.append(
        [tool_call.function for tool_call in response.choices[0].message.tool_calls]
    )

    for tool_call in response.choices[0].message.tool_calls:
        if tool_call.function.name == "send_query_to_agents":
            agents = json.loads(tool_call.function.arguments)["agents"]
            query = json.loads(tool_call.function.arguments)["query"]
            for agent in agents:
                if agent == "Data Processing Agent":
                    handle_data_processing_agent(query, conversation_messages)
                elif agent == "Analysis Agent":
                    handle_analysis_agent(query, conversation_messages)
                elif agent == "Visualization Agent":
                    handle_visualization_agent(query, conversation_messages)

    outputs = extract_tool_contents(conversation_messages)

    return outputs

functions = [
    "clean_data",
    "transform_data",
    "stat_analysis",
    "aggregate_data",
    "correlation_analysis",
    "regression_analysis",
    "create_bar_chart",
    "create_line_chart",
    "create_pie_chart",
]

@weave.op()
def extract_tool_contents(data):
    contents = {}
    contents["all"] = data
    for element in data:
        if (
            isinstance(element, dict)
            and element.get("role") == "tool"
            and element.get("name") in functions
        ):
            name = element["name"]
            content_str = element["content"]
            try:
                content_json = json.loads(content_str)
                if "chart" not in element.get("name"):
                    contents[name] = [content_json]
                else:
                    first_key = next(iter(content_json))
                    second_level = content_json[first_key]
                    if isinstance(second_level, dict):
                        second_key = next(iter(second_level))
                        contents[name] = second_level[second_key]
                    else:
                        contents[name] = second_level
            except json.JSONDecodeError:
                print(f"{name}의 JSON 디코딩 오류")
                contents[name] = None

    return contents

​
Weave에서 멀티 에이전트 시스템 실행 및 시각화

마지막으로 사용자의 입력을 전달하여 주요 handle_user_message 함수를 실행하고, 그 결과를 확인합니다. 
handle_user_message(user_query)
Weave URL을 클릭하면 실행이 다음과 같이 추적되고 있는 것을 확인할 수 있습니다. Traces 페이지에서 입력과 출력을 확인할 수 있습니다. 이해를 돕기 위해, 각 출력을 클릭했을 때 표시되는 결과의 스크린샷을 다이어그램에 추가했습니다. Weave는 OpenAI의 API와 연동되어 비용을 자동으로 계산할 수 있습니다. 따라서 오른쪽 끝에서 비용과 지연 시간도 함께 표시되는 것을 확인할 수 있습니다. 1-1.png 한 줄을 클릭하면 멀티 에이전트 시스템 내에서 실행된 중간 과정을 확인할 수 있습니다. 예를 들어 analysis_agent의 입력과 출력을 확인해 보면, 이것이 구조화된 출력 형식임을 알 수 있습니다. OpenAI의 구조화된 출력은 에이전트 간 협업을 용이하게 하지만, 시스템이 복잡해질수록 이러한 상호작용이 어떤 형식으로 이루어지는지 파악하기가 어려워집니다. Weave를 사용하면 이러한 중간 과정과 그 입력 및 출력을 마치 손에 들고 있는 것처럼 직관적으로 이해할 수 있습니다.
3.png
Weave에서 트레이싱이 어떻게 처리되는지 조금 더 자세히 살펴보십시오!

​
결론

이 튜토리얼에서는 OpenAI가 제공하는 구조화된 출력 기능과 Weave를 활용해 입력, 최종 출력, 중간 출력 형식을 추적할 수 있는 멀티 에이전트 시스템을 간편하게 개발하는 방법을 살펴보았습니다.