'스스로 개선하는' 에이전트 만들어보기

바이브 코딩 (Vibe Coding), 많이 들어보시고 해 보신 분들도 많으리라 생각합니다. 바이브 코딩이라고 하면, 대략 ‘직접 코딩을 하는게 아니라, 원하는 결과물의 느낌 (바이브)을 말로 설명해서 개발하는 방식’ 정도가 되겠죠?

바이브 코딩은 그 나름대로 굉장한 의미와 역할을 하게 되겠지만, 많은 경우에 바이브 코딩을 하다가 보면 개발의 다양한 과정, 특히 디버깅의 관점에서 벽에 부딪히게 될 겁니다. 특히 LLM 기반의 에이전트를 만들 때, 이 문제가 더 심각해지리라 예상됩니다.

아무래도 LLM이라는 기술, 도구는 ‘예측 불가능하다’는 특성이 있죠. 그래서 LLM의 Response (응답)를 평가하는 작업 자체가 하나의 독립된 분야가 되기도 했는데요. 이 작업 때문에, 실제로 LLM 에이전트를 처음 만드는데 걸리는 시간과 자원 대비, 프로덕션에 배포하고 운영할 수 있는 수준으로 다듬는데 더 많은 시간과 자원이 들어가는 경우도 많이 보게 됩니다.

이 ‘다듬는’ 과정은, 다른 각도에서 보자면 ‘반복적’인 작업이고, ‘비판적인 사고가 필요한’ 작업이죠 - 바로, 자연스럽게 “AI가 도움을 줄 수 있지 않을까?”하는 생각이 드는 영역 아니겠어요?

한 번, 해 볼까요? 스스로의 성능 (Performance)을 자가 분석 (Self-Analyze)해서 (사람의 개입이 없어도) 프롬프트를 최적화해가면서 데이터베이스에 질의해서 답을 얻어내는 에이전트를 만들어 봅시다. 그리고 이걸 MCP 아키텍처를 이용해서 만들면 어떻게 될지도 간단히 알아보겠습니다.

이 글을 통해서, AI 에이전트를 지속적으로 자가 개선하는 사이클을 만들기 위한 평가, 추적 (Tracing), 실험, 프롬프트 최적화 개념들을 살펴보실 수 있기를 바랍니다.

🌱 이 글에서 만들 에이전트, 그리고 필요한 도구

오늘, 데이터베이스에 쿼리를 날리는 에이전트를 만들어 볼 겁니다.

여기서 핵심은, 이 에이전트가 스스로 자신의 성능을 분석하고, 프롬프트를 최적화하고, 사람이 도와주지 않아도 점점 더 똑똑해지도록 하는 것입니다. 그렇게 하기 위해서, 아래의 네 가지 개념을 합쳐서 하나의 ‘루프’로 만들려고 합니다:

1. 에이전트의 평가 (Evaluation)

2. 동작 추적 (Tracing)

3. 실험 및 테스트 (Experimentation)

4. 프롬프트 최적화 (Prompt Optimization)

이 네 가지 요소가 함께 작동하면서, 계속해서 진화하는 시스템, 즉 선순환 개선 구조(Virtuous Cycle)가 형성됩니다.

이번 글의 예제에서는, 아래와 같은 도구들을 사용합니다:

🔁 선순환 구조의 ‘10단계 자기 개선’ 루프

선순환 개선 구조를 가진 이 에이전트는, 이렇게 작동합니다:

1. 정답 레이블이 포함된 테스트 케이스셋을 생성 (자동 또는 수동): Ground Truth Dataset

2. DSPy를 사용해서, 테스트 케이스를 기반으로 최적화된 프롬프트 생성

3. Phoenix에 프롬프트 저장 (버전 관리 가능)

4. 다양한 실험을 해 보면서 에이전트의 성능 테스트

5. 일정한 수준을 넘는 - 기준을 만족하는 - 성능을 보여주면 프로덕션용으로 태깅

6. 실제 서비스에서 에이전트를 실행하면서 동작 추적

7. LLM 평가 도구를 사용해서 응답을 자동 레이블링

8. (Optional) 사람이 직접 응답 레이블을 검토

9. 새로 레이블링된 데이터를 학습 데이터로 추가

10. 다시 2번으로 돌아가서 루프 반복

💡 자, 이제 만들어 봅시다.

지금부터 그럼 여러 가지 도구들을 호출하는 에이전트(Agent with Tools)의 프롬프트 최적화를 중점으로 해서 작업을 시작해 봅시다. 그리고, Tool - 도구 - 호출 부분과 관련해서, MCP로 하는 방법도 한 번 글의 말미에 생각해 보겠습니다.

(참고로, 위에서 이야기한 이 루프는 에이전트가 사용하는 도구의 성능 문제, 에이전트의 의사결정 경로 선택을 포함해서 기타 에이전트 구성 요소에도 그대로 적용할 수 있으니, 관심있으신 분은 각자 한 번 해 보시기를 권합니다.)

에이전트 셋업

평가를 하든지 하려면 (^.^;) 작동하는 에이전트가 있어야겠죠? 기본적인 도구를 호출하는 에이전트 구성, 추적법을 말씀드릴 텐데, 이 글의 목적 상 ‘에이전트 도구’를 정의하는 코드는 여기서는 건너뛰겠습니다. 전체 구현 코드는 여기에 있으니 참고하세요.

환경 설정에 필요한 패키지는 아래와 같이 설치할 수 있습니다:

!pip install uv
!uv pip install -q openai "arize-phoenix>=8.8.0" "arize-phoenix-otel>=0.8.0" openinference-instrumentation-openai python-dotenv duckdb "openinference-instrumentation>=0.1.21" tqdm dspy

필요한 라이브러리를 Import합니다:

import dotenv
dotenv.load_dotenv()

import json
import os
from getpass import getpass

import duckdb
import pandas as pd
from openai import OpenAI
from openinference.instrumentation.openai import OpenAIInstrumentor
from opentelemetry.trace import StatusCode
from pydantic import BaseModel, Field
from tqdm import tqdm

from phoenix.otel import register
from phoenix.client import Client as PhoenixClient

API와 디폴트값 등을 설정합니다:

if os.getenv("OPENAI_API_KEY") is None:
    os.environ["OPENAI_API_KEY"] = getpass("Enter your OpenAI API key: ")

client = OpenAI()
model = "gpt-4o-mini"
project_name = "self-improving-agent"

물론, GPT-4o-mini가 아니라도, 도구 호출 (Function Calling)을 지원하는 어떤 모델이든 사용할 수 있습니다.

Phoenix Tracing 설정

그럼 다음으로는, 만들어 놓은 에이전트를 Phoenix에 연결을 해야겠죠?

Phoenix 인스턴스가 이미 있는 분은 거의 없으실 텐데, 직접 호스팅을 하거나 온라인에서 무료로 계정을 받아서 만들어서 사용할 수도 있습니다. 여기서는 온라인 계정을 사용합니다.

if os.getenv("PHOENIX_API_KEY") is None:
    os.environ["PHOENIX_API_KEY"] = getpass("Enter your Phoenix API key: ")

os.environ["PHOENIX_COLLECTOR_ENDPOINT"] = "https://app.phoenix.arize.com/"
os.environ["PHOENIX_CLIENT_HEADERS"] = f"api_key={os.getenv('PHOENIX_API_KEY')}"

그 다음, Tracer를 등록합니다:

tracer_provider = register(
    project_name=project_name,
    auto_instrument=True,
)

tracer = tracer_provider.get_tracer(__name__)

위 코드로 Phoenix 인스턴스에 연결됩니다. 코드 중에 ‘auto_instrument’ 기능은 사용자의 환경에서 OpenInference 트레이싱 (Tracing) 라이브러리를 스캔해서, 해당 라이브러리가 있으면 그 라이브러리를 호출해서 관련한 라이브러리에 대한 모든 호출 (Call)을 캡처합니다. ‘openinference-instrumentation-openai’라는 라이브러리가 있기 때문에, 자동으로 OpenAI에 대한 모든 호출을 기록하게 되죠.

Phoenix에 초기 라우터 프롬프트 저장

그 다음으로는, Versioning (버전 관리)와 Traceability (추적 가능성)를 위해서, Phoenix에 프롬프트를 저장합니다. 이 작업은 두 가지 이유 때문에 중요한데요:

첫번째, Phoenix가 프롬프트가 반복적으로 만들어지거나 업데이트될 때마다, 자동으로 새 버전으로 생성해서 변경 사항이 많더라도 종합적인 로그를 만들어줍니다. 자주, 그리고 반복적으로 프롬프트를 변경하게 되는 상황이라면 아주 유용하고 필요한 기능이죠.

두번째, Phoenix를 사용하면, 대시보드나 코드를 통해서 프롬프트의 버전을 ‘Production (실제 환경)’, ‘Staging (스테이징 환경)’, ‘Development (개발 환경)’ 등으로 태깅할 수가 있어요. 코드를 프롬프트 태그와 연결할 수가 있어서, 코드를 다시 배포하지 않아도 에이전트가 어떤 프롬프트를 사용할지 정교하게 매치시킬 수가 있습니다.

import phoenix as px
from phoenix.client.types import PromptVersion
from openai.types.chat.completion_create_params import CompletionCreateParamsBase

params = CompletionCreateParamsBase(
    model="gpt-4o-mini",
    tools=tools,
    messages=[
        {"role": "system", "content": "You are a helpful assistant that can answer questions about the Store Sales Price Elasticity Promotions dataset."},
        {"role": "user", "content": "{user_query}"},
    ],
)

prompt_name = "self-improving-agent-router"
prompt = px.Client().prompts.create(
    name=prompt_name,
    version=PromptVersion.from_openai(params),
)

# Tag your prompt as ready for production
px.Client().prompts.tags.create(
    prompt_version_id=prompt.id,
    name="production",
    description="Ready for production environment"
)

에이전트 라우팅 로직 정의

그럼, 이제 도구 호출 (Function Call)과 에이전트 라우팅 (Agent Routing)을 처리하는, 핵심적인 에이전트 로직을 구현합니다:

@tracer.chain()
def handle_tool_calls(tool_calls, messages):
    for tool_call in tool_calls:
        function = tool_implementations[tool_call.function.name]
        function_args = json.loads(tool_call.function.arguments)
        result = function(**function_args)

        messages.append({"role": "tool", "content": result, "tool_call_id": tool_call.id})
    return messages

에이전트의 메인 실행 루프는 아래와 같이 구성됩니다:

def run_agent(messages):
    if isinstance(messages, str):
        messages = [{"role": "user", "content": messages}]

    # Check and add system prompt if needed
    if not any(
        isinstance(message, dict) and message.get("role") == "system" for message in messages
    ):
	# Retrieve the production tagged prompt in Phoenix
        phoenix_production_router_prompt = PhoenixClient().prompts.get(prompt_identifier="self-improving-agent-router", tag="production")
        
        system_prompt = {
            "role": "system",
            "content": phoenix_production_router_prompt,
        }
        messages.append(system_prompt)

    while True:
        # Router call instrumentation
        with tracer.start_as_current_span(
            "router_call",
            openinference_span_kind="chain",
        ) as span:
            span.set_input(value=messages)

		# Call the agent router
            response = client.chat.completions.create(
                model=model,
                messages=messages,
                tools=tools,
            )

            messages.append(response.choices[0].message.model_dump())
            tool_calls = response.choices[0].message.tool_calls
            span.set_status(StatusCode.OK)

		# Handle tool calls, or respond to the user
            if tool_calls:
                # Tool calls instrumentation
                messages = handle_tool_calls(tool_calls, messages)
                span.set_output(value=tool_calls)
            else:
                span.set_output(value=response.choices[0].message.content)
                return response.choices[0].message.content

에이전트 실행하기

이제 몇 가지 질문으로 에이전트를 테스트해 봅시다:

ret = run_agent([{"role": "user", "content": "Create a line chart showing sales in 2021"}])

제대로 에이전트가 실행되고 Phoenix에서 트레이스를 생성하고 있다면, 아래와 같은 시각화된 내용을 볼 수 있을 겁니다:

자, 여기까지 해서, Phoenix 플랫폼을 통해서 원격으로 Function Call을 캡처하면서 작동하는 에이전트를 구성해 봤습니다.

다음의 몇 개 단계들에서는, ‘Experimentation Suite (실험 체계)’, ‘Prompt Optimization (프롬프트 최적화)’, 그리고 ‘Production Evaluation (운영 환경에서의 작동 평가)’ 요소를 구성하는 방법을 알아봅니다:

개발 단계에서 에이전트 테스트하기

Production 환경으로 배포하기 전에, 에이전트의 성능을 일단 평가해 봐야 하겠죠. 레이블이 지정된 데이터셋과 비교해서, 에이전트의 Function Calling 선택을 비교하는 실험을 해 보면 될 겁니다.

import nest_asyncio

import phoenix as px
from phoenix.evals import TOOL_CALLING_PROMPT_TEMPLATE, OpenAIModel, llm_classify
from phoenix.experiments import run_experiment
from phoenix.experiments.types import Example
from phoenix.trace import SpanEvaluations
from phoenix.trace.dsl import SpanQuery

nest_asyncio.apply()

px_client = px.Client()
eval_model = OpenAIModel(model="gpt-4o-mini")

함수 호출 (Function Call) 실험 설정하기

실험 (Experiment)은 다음의 3가지 구성요소로 이루어집니다:

1. 알려진 출력이 있는 테스트 케이스 데이터셋

2. 각각의 테스트 케이스에서 실행할 작업 - 이 경우에는 라우터 호출입니다.

3. 해당 작업의 결과를 평가할, 하나 이상의 Evaluator

먼저, 잘 알려진 예상 출력값이 있는 테스트 케이스 데이터셋을 만듭니다:

import uuid
id = str(uuid.uuid4())

# Create a list of tuples with input_messages and next_tool_call
data = [
    (
        [
            {
                "role": "user",
                "content": "Plot daily sales volume over time"
            },
            {
                "role": "system",
                "content": "You are a helpful assistant that can answer questions about the Store Sales Price Elasticity Promotions dataset."
            },
            {
                "role": "assistant",
                "tool_calls": [
                    {
                        "id": "call_1",
                        "type": "function",
                        "function": {
                            "name": "lookup_sales_data",
                            "arguments": "{\"prompt\":\"Plot daily sales volume over time\"}"
                        }
                    }
                ]
            },
            {
                "role": "tool",
                "tool_call_id": "call_1",
                "content": "     Sold_Date  Daily_Sales_Volume\n0   2021-11-01              1021.0\n1   2021-11-02              1035.0\n2   2021-11-03               900.0"
            }
        ],
        "analyze_sales_data"
    ),
    # ... more test cases here ...
]

dataframe = pd.DataFrame(data, columns=["input_messages", "next_tool_call"])

dataset = px_client.upload_dataset(
    dataframe=dataframe,
    dataset_name=f"tool_calling_ground_truth_{id}",
    input_keys=["input_messages"],
    output_keys=["next_tool_call"],
)

다음으로, 에이전트의 라우터 단계만 실행하는 작업 함수를 생성합니다:

def run_router_step(example: Example) -> str:
    input_messages = example.input.get("input_messages")

    phoenix_production_router_prompt = PhoenixClient().prompts.get(prompt_identifier="self-improving-agent-router", tag="development")
    
    system_prompt = {
        "role": "system",
        "content": phoenix_production_router_prompt,
    }
    
    # Replace the system message in input_messages with our production router prompt
    # or add it if no system message exists
    system_message_index = None
    
    for i, message in enumerate(input_messages):
        if message.get("role") == "system":
            system_message_index = i
            break
    
    if system_message_index is not None:
        # Replace existing system message
        input_messages[system_message_index] = system_prompt
    else:
        # Add system message if none exists
        input_messages.insert(0, system_prompt)
    
    response = client.chat.completions.create(
        model=model,
        messages=input_messages,
        tools=tools,
    )
    
    if response.choices[0].message.tool_calls is None:
        return "no tool called"
    
    tool_calls = []
    for tool_call in response.choices[0].message.tool_calls:
        tool_calls.append(tool_call.function.name)
    return tool_calls

그리고 마지막으로, 예상되는 출력과 실제 출력값을 비교하는 Evaluator를 만듭니다:

def tools_match(expected: str, output: str) -> bool:
    if not isinstance(output, list):
        return False
    
    # Check if all expected tools are in output and no additional tools are present
    expected_tools = expected.get("next_tool_call").split(", ")
    expected_set = set(expected_tools)
    output_set = set(output)
    
    # Return True if the sets are identical (same elements, no extras)
    return expected_set == output_set

이제 ‘실험 (Experiment)’을 돌려볼 수 있습니다:

experiment = run_experiment(
    dataset,
    run_router_step,
    evaluators=[tools_match],
    experiment_name="Tool Calling Eval",
    experiment_description="Evaluating the tool calling step of the agent",
)

제공되는 링크를 눌러보면, 아래와 같이 ‘실험’의 요약된 내용을 볼 수 있습니다.

DSPy로 에이전트 최적화하기

자, 이 부분이 이제 ‘스스로 개선’하는 마법같은 과정이 일어나는 곳입니다.

DSPy를 사용하면, Bootstrapped Few Shot 프롬프트라든가 MiProV2 같은 다양한 최적화 기술, 그리고 레이블이 지정된 테스트 케이스셋을 사용해서 ‘프롬프트를 최적화’할 수 있습니다.

먼저, DSPy로 우리의 에이전트 라우터를 나타내는 시그니처를 생성합니다.

import dspy

# Configure DSPy to use OpenAI
dspy_lm = dspy.LM(model="gpt-4o-mini")
dspy.settings.configure(lm=dspy_lm)

# Define the prompt classification task
class RouterPromptSignature(dspy.Signature):
    """Route a user prompt to the correct tool based on the task requirements.
    
    Available tools:
    1. analyze_sales_data: Use for complex analysis of sales data, including trends, patterns, and insights
    2. lookup_sales_data: Use for simple data retrieval or filtering of sales records
    3. generate_visualization: Use when the user needs visual representation of data
    4. no tool called: Use when no tool is needed
    
    The tool selection should be based on:
    - The complexity of the analysis needed
    - Whether raw data or processed insights are required
    - If visualization would help communicate the results
    """

    input_messages = dspy.InputField(desc="The routers input messages. Can include the user's query and any tool calls that have already been made.")
    tool_call = dspy.OutputField(
        desc="A list of tool calls to execute in sequence. Each tool call should include: "
             "1. tool_name: The name of the tool to use "
    )

router = dspy.Predict(RouterPromptSignature)

기본 라우터를 테스트합니다:

result = router(input_messages=[{"role": "user", "content": "Which stores had the highest sales volume?"}])

이제 이전 예제에서 학습 데이터셋을 만듭니다. 적은 수의 예제만으로도 충분한 효과를 얻을 수 있습니다.

trainset = []

for input_messages, next_tool_call in dataframe.values:
    trainset.append(dspy.Example(input_messages=input_messages, tool_call=next_tool_call).with_inputs("input_messages"))

print(trainset[:3])

마지막으로, DSPy로 라우터를 최적화합니다:

# BootstrapFinetune을 통한 최적화
optimizer = dspy.BootstrapFewShot(metric=(lambda x, y, trace=None: x.tool_call == y.tool_call))
optimized = optimizer.compile(router, trainset=trainset)

optimized(input_messages=[{"role": "user", "content": "Which stores had the highest sales volume?"}])

이 과정에서, 성능 수치에 따라 프롬프트 업데이트를 하거나 안 하거나 결정하도록 제한할 수 있습니다 - 성능이 특정한 수준만큼 향상된 경우에만 진행한다든가 하는 거죠. 만족스러운 수치의 성능에 도다하면, 최적화된 프롬프트를 추출, Phoenix에 저장합니다:

# 최적화된 라우터에서 프롬프트 가져오기
new_prompt = optimized.signature.instructions
print(new_prompt)

params = CompletionCreateParamsBase(
    model="gpt-4o-mini",
    tools=tools,
    messages=[
        {"role": "system", "content": new_prompt},
        {"role": "user", "content": "{user_query}"},
    ],
)

# Phoenix에서 기존 프롬프트를 업데이트
prompt_name = "self-improving-agent-router"
prompt = px.Client().prompts.create(
    name=prompt_name,
    prompt_description="Router prompt for the self-improving agent",
    version=PromptVersion.from_openai(params),
)

# 프롬프트 버전에 대한 태그 생성
px.Client().prompts.tags.create(
    prompt_version_id=prompt.id,
    name="production",
    description="Ready for production environment"
)

Phoenix 플랫폼을 사용할 때의 장점은, 저장된 프롬프트를 즉시, 바로 적용할 수 있다는 거죠.

자, 물론, 프로덕션 환경에서는 예외없이 항상 새로운 문제가 발생합니다. 그러니, 이제 프로덕션 환경에서의 ‘학습’을 에이전트의 라우팅 단계에 연계하는 방법을 살펴봐야겠죠?

프로덕션 (Production) 환경에서 에이전트 평가하기

프로덕션 환경에서 에이전트를 평가하는 한 가지 방법은, 바로 ‘사람이 직접 평가하는’ 것이죠. 그런데 이게 시간과 비용이 많이 들겠죠?

그래서, 대안으로 많이 사용하는 접근 방법이, LLM Judge - LLM을 평가자로 활용하는 것 - 방식입니다. LLM Judge가 항상 정확한 건 아니겠지만, 정확한 또는 잘못된 Trace에 대해서 아주 정확한 방향성과 피드백을 제공할 수 있습니다.

또는, 정확성과 효율성을 최적화하기 위해서, 사람 평가자와 LLM Judge를 결합해서 하이브리드 방식으로 사용할 수도 있겠죠.

어쨌든, 함수 호출에 대한 평가를 할 LLM Judge를 만들기 위해서, 먼저 Phoenix에서 평가하려는 데이터를 내보냅니다:

# Phoenix에서 도구 호출 추출하기
def get_tool_calls():
    query = (
        SpanQuery()
        .where(
            "span_kind == 'LLM'",
        )
        .select(question="input.value", output_messages="llm.output_messages")
    )

    # Phoenix 클라이언트는 이 쿼리를 받아 데이터프레임을 반환
    tool_calls_df = px.Client().query_spans(query, project_name=project_name, timeout=None)
    tool_calls_df.dropna(subset=["output_messages"], inplace=True)

    def get_tool_call(outputs):
        if outputs[0].get("message").get("tool_calls"):
            return (
                outputs[0]
                .get("message")
                .get("tool_calls")[0]
                .get("tool_call")
                .get("function")
                .get("name")
            )
        else:
            return "No tool used"

    tool_calls_df["tool_call"] = tool_calls_df["output_messages"].apply(get_tool_call)
    tool_definitions_list = [tools] * len(tool_calls_df)
    tool_calls_df["tool_definitions"] = tool_definitions_list
    return tool_calls_df

그 다음, Function Call을 누가 평가할지를 정의하는데, 이 경우는 LLM Judge겠죠. 따라서 Phoenix에서, 도구 호출을 평가할 판정자로 LLM을 정의해야 하는데, Phoenix는 LLM이 도구 호출을 잘 했는지 LLM Judge로 작동하도록 하는 내장 (Build-in)의 LLM 프롬프트 템플릿을 제공합니다:

def eval_tool_calls(dataframe):
    # Phoenix 헬퍼 함수로 데이터의 각 행에 도구 호출 템플릿을 적용
    tool_call_eval = llm_classify(
        data=dataframe,
        template=TOOL_CALLING_PROMPT_TEMPLATE,
        rails=["correct", "incorrect"],
        model=eval_model,
        provide_explanation=True,
    )

    tool_call_eval["score"] = tool_call_eval.apply(
        lambda x: 1 if x["label"] == "correct" else 0, axis=1
    )

    return tool_call_eval, dataframe

def eval_and_log_tool_calls():
    tool_calls_df = get_tool_calls()
    tool_call_eval, dataframe = eval_tool_calls(tool_calls_df)
    # Phoenix UI에 평가 기록하기
    px.Client().log_evaluations(
        SpanEvaluations(eval_name="Tool Calling Eval", dataframe=tool_call_eval),
    )
    
    # context.span_id를 기준으로 평가 결과와 원본 데이터프레임 병합
    merged_df = pd.merge(
        tool_call_eval,
        dataframe,
        left_index=True,
        right_index=True,
        how='inner'
    )
    
    # 병합된 데이터프레임 반환
    return merged_df

이제 평가를 실행합니다:

tool_call_eval = eval_and_log_tool_calls()

이 함수를 실행한 이후, Phoenix에서 아래 이미지와 유사한 평가 결과를 볼 수가 있습니다:

자동적인 개선 루프 (Automatic Improvement Loop) 만들기

자, 지금까지 살펴본 것들을 연결해서, 하나의 완결적인 ‘자기 개선 루프’를 만들면 되겠죠. 몇 가지 헬퍼 함수만 더 정의하구요:

def create_trainset(tool_call_eval):
    trainset = []
    for _, row in tool_call_eval.iterrows():
        if row["label"] == "correct":
            trainset.append(dspy.Example(input_messages=row["question"], tool_call=row["tool_call"]).with_inputs("input_messages"))
    return trainset

def save_trainset(trainset):
    trainset_df = pd.DataFrame(trainset)
    px.Client().upload_dataset(
        dataframe=trainset_df,
        dataset_name="self-improving-agent-trainset-{}".format(uuid.uuid4()),
    )

def optimize_router(trainset):
    optimizer = dspy.BootstrapFewShot(metric=(lambda x, y, trace=None: x.tool_call == y.tool_call))
    optimized = optimizer.compile(router, trainset=trainset)
    new_prompt = optimized.signature.instructions
    return new_prompt

def run_experiment():
    experiment = run_experiment(
        dataset,
        run_router_step,
        evaluators=[tools_match],
        experiment_name="Tool Calling Eval",
        experiment_description="Evaluating the tool calling step of the agent",
    )
    return experiment.eval_summaries()

def save_prompt(prompt):
    params = CompletionCreateParamsBase(
        model="gpt-4o-mini",
        tools=tools,
        messages=[
            {"role": "system", "content": new_prompt},
            {"role": "user", "content": "{user_query}"},
        ],
    )

    # 이것은 Phoenix에서 기존 프롬프트를 업데이트
    prompt_name = "self-improving-agent-router"
    prompt = px.Client().prompts.create(
        name=prompt_name,
        prompt_description="Router prompt for the self-improving agent",
        version=PromptVersion.from_openai(params),
    )
    
    px.Client().prompts.tags.create(
        prompt_version_id=prompt.id,
        name="production",
        description="Ready for production environment"
    )

그리고 드디어 마지막으로, 자동화 루프 그 자체입니다:

def automated_loop():
    # 1단계: 프로덕션 성능 평가
    tool_call_eval = eval_and_log_tool_calls()
    
    # 2단계: 성공적인 예제에서 학습 세트 생성
    trainset = create_trainset(tool_call_eval)
    save_trainset(trainset)
    
    # 3단계: DSPy를 사용해 라우터 프롬프트 최적화
    new_prompt = optimize_router(trainset)
    
    # 4단계: 새 프롬프트 벤치마킹을 위한 실험 실행
    experiment_results = run_experiment()
    print(experiment_results.eval_summaries())
    
    # 5단계: 사용자에게 새 프롬프트 적용 여부 질문
    apply_prompt = input("새 프롬프트를 적용하시겠습니까? (yes/no): ")
    
    if apply_prompt.lower() not in ["yes", "y"]:
        print("프롬프트 업데이트가 취소되었습니다.")
        return
    
    print("새 프롬프트 적용 중...")
    save_prompt(new_prompt)

아, 너무나 간단하고 아름답죠? ^.^;

이렇게, 평가를 실행하고, 제대로 된 실행 결과를 내보내서 학습 셋에 추가하고, 학습 셋에 최적화된 프롬프트를 만들고, 프롬프트를 프로덕션 환경에 배포하는 함수가 만들어진 것이죠.

이 함수를, 특정한 주기로 실행하도록 설정을 하든, 아니면 어떤 조건에 따라서 실행하도록 설정하면 ‘자동적으로 스스로 개선하는’ 에이전트 구조가 완성됩니다.

✅ 마무리: ‘자기 개선’, 에이전트의 미래 방향

자, 조금 길기는 했지만, ‘스스로 작동을 하는 과정에서 프롬프트를 최적화해 가는, Self-Improving 에이전트’를 구축해 봤습니다:

이 과정에서 Arize의 Phoenix, DSPy 등 몇 가지 도구를 활용했죠:

1. Phoenix의 추적 (Tracing) 기능을 사용해서 자체적인 성능을 모니터링합니다.

2. 실제 정답, 그리고 LLM이 Evaluator 역할을 하는 구조로 에이전트의 Function Call을 평가합니다.

3. DSPy를 사용해서 프롬프트를 자동으로 최적화합니다.

4. Phoenix에서 프롬프트의 버전을 관리합니다.

5. 이 에이전트를 더 많이 사용할 수록 더 개선되는, 지속적인 개선 루프를 만듭니다.

이 글에서 말씀드린 구조와 접근 방식은, 사실상 ‘모든 에이전트 시스템에 적용’할 수 있습니다. Observability (관찰 가능성), 그리고 Optimization (최적화) 개념과 기술을 결합해서, 에이전트와 사용자 간의 모든 상호작용을 개선의 기회로 활용하는, 상당히 괜찮은 피드백 루프를 만들 수 있다는 겁니다.

물론, 일부 구성 요소 - DSPy가 사용하는 최적화 기법을 다른 것으로 선택한다거나, 에이전트 내부의 다른 측면을 최적화한다거나, 프로덕션에서 최적화된 프롬프트를 적용하는 플로우를 변경한다거나 - 의 변경을 통해서, 이 접근 방식을 수없이 다양한 방향으로 확장할 수도 있습니다.

또, 최근 아주 핫하게 떠오른 MCP (Model Context Protocol) 대응 트렌드를 반영해서, Arize에서 Phoenix MCP Server도 3일 전에 공개를 해 놨으니, MCP 클라이언트 - MCP 서버 구조로 Phoenix 플랫폼을 사용하는 코드로 변경도 가능할 것 같네요.

에이전트 기술, MCP, 그리고 며칠 전 구글이 발표한 A2A. 모두 “LLM 기반 에이전트는 정적인 시스템이 아니라 다이나믹하게 변화하는 시스템이다”라고 외치는 듯합니다. 그런 관점에서, 프롬프트 튜닝, 평가, Observability 등의 기능을 잘 합쳐서, 하나의 ‘끝없이, 스스로 개선하는 선순환 루프’를 가진 에이전트의 비전을 여러분의 현장에서도 실험해 보실 수 있기를 바랍니다.

읽어주셔서 감사합니다. 재미있게 보셨다면 친구와 동료 분들에게도 뉴스레터를 추천해 주세요.