Azure AI Foundry Agent Service와 W&B Weave로 AI 에이전트 구축 및 평가

brew update && brew install azure-cli

az login 

brew tap azure/functions
brew install azure-functions-core-tools@4
brew link --overwrite azure-functions-core-tools@4

az account show --query id -o tsv

export AZURE_SUBSCRIPTION_ID=your_id_from_above

azd init --template functions-quickstart-python-http-azd -e flexquickstart-py

python3 -m venv .venv
source .venv/bin/activate

az provider register --namespace Microsoft.App

import azure.functions as func
import logging
import requests
import json

API_KEY = "your_serper_api_key"

app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)

@app.route(route="foodsearch", methods=["POST"])
def food_search(req: func.HttpRequest) -> func.HttpResponse:
    logging.info("🔵 food_search triggered")
    try:
        data = req.get_json()
        query = data.get("query")
        latitude = data.get("latitude")
        longitude = data.get("longitude")
        cuisine = data.get("cuisine")

        if not (query and latitude and longitude):
            return func.HttpResponse(
                json.dumps({"error": "Missing required parameters. Must include query, latitude, and longitude."}),
                status_code=400
            )

        search_query = f"{query} {cuisine}" if cuisine else query
        search_params = f"ll=@{latitude},{longitude},14z"
        url = f"https://serpapi.com/search?engine=google_local&q={requests.utils.quote(search_query)}&{search_params}&api_key={API_KEY}"
        logging.info(f"🟠 Requesting: {url}")

        res = requests.get(url)
        res.raise_for_status()
        results = res.json().get("local_results", [])

        if not results:
            return func.HttpResponse(json.dumps({"error": "No food places found"}), status_code=404)

        places = [{
            "name": r.get("title", "Unknown"),
            "rating": r.get("rating", "N/A"),
            "reviews": r.get("reviews", 0)
        } for r in results]

        return func.HttpResponse(json.dumps({"food_places": places}), mimetype="application/json", status_code=200)

    except Exception as e:
        logging.exception("food_search failed")
        return func.HttpResponse(json.dumps({"error": "Server error", "details": str(e)}), status_code=500)

@app.route(route="hotelsearch", methods=["POST"])
def hotel_search(req: func.HttpRequest) -> func.HttpResponse:
    logging.info("🔵 hotel_search triggered")
    try:
        data = req.get_json()
        location = data.get("location")
        check_in = data.get("check_in_date")
        check_out = data.get("check_out_date")
        adults = data.get("adults")
        currency = data.get("currency", "USD")

        if not (location and check_in and check_out and adults):
            return func.HttpResponse(json.dumps({"error": "Missing required parameters", "received": data}), status_code=400)

        url = (
            f"https://serpapi.com/search?engine=google_hotels"
            f"&q={requests.utils.quote(location)}"
            f"&check_in_date={check_in}"
            f"&check_out_date={check_out}"
            f"&adults={adults}"
            f"&currency={currency}"
            f"&api_key={API_KEY}"
        )
        logging.info(f"🟠 Requesting: {url}")

        res = requests.get(url)
        res.raise_for_status()
        properties = res.json().get("properties", [])

        if not properties:
            return func.HttpResponse(json.dumps({"error": "No hotels found"}), status_code=404)

        hotels = [{
            "name": h.get("name", "No Name"),
            "price": h.get("rate_per_night", {}).get("lowest", "N/A"),
            "currency": currency,
            "lat": h.get("gps_coordinates", {}).get("latitude"),
            "lon": h.get("gps_coordinates", {}).get("longitude"),
            "link": h.get("link", "N/A")
        } for h in properties]

        return func.HttpResponse(json.dumps({"hotels": hotels}), mimetype="application/json", status_code=200)

    except Exception as e:
        logging.exception("hotel_search failed")
        return func.HttpResponse(json.dumps({"error": "Server error", "details": str(e)}), status_code=500)


azd up

{
  "openapi": "3.0.0",
  "info": {
    "title": "Hotel Search Tool",
    "version": "1.0.0",
    "description": "Find hotels by location and travel dates."
  },
  "servers": [
    {
      "url": "url to your api"
    }
  ],
  "paths": {
    "/hotelsearch": {
      "post": {
        "operationId": "hotelSearch",
        "summary": "Search for hotels",
        "description": "Search for hotels using location, check-in/check-out dates, and number of adults.",
        "requestBody": {
          "required": true,
          "content": {
            "application/json": {
              "schema": {
                "type": "object",
                "required": [
                  "location",
                  "check_in_date",
                  "check_out_date",
                  "adults"
                ],
                "properties": {
                  "location": {
                    "type": "string",
                    "example": "New York, NY"
                  },
                  "check_in_date": {
                    "type": "string",
                    "format": "date",
                    "example": "2025-07-01"
                  },
                  "check_out_date": {
                    "type": "string",
                    "format": "date",
                    "example": "2025-07-04"
                  },
                  "adults": {
                    "type": "integer",
                    "example": 2
                  },
                  "currency": {
                    "type": "string",
                    "example": "USD"
                  }
                }
              }
            }
          }
        },
        "responses": {
          "200": {
            "description": "Success",
            "content": {
              "application/json": {
                "schema": {
                  "type": "object",
                  "properties": {
                    "hotels": {
                      "type": "array",
                      "items": {
                        "type": "object",
                        "properties": {
                          "name": {
                            "type": "string"
                          },
                          "price": {
                            "type": "string"
                          },
                          "currency": {
                            "type": "string"
                          },
                          "lat": {
                            "type": "number"
                          },
                          "lon": {
                            "type": "number"
                          },
                          "link": {
                            "type": "string"
                          }
                        }
                      }
                    }
                  }
                }
              }
            }
          },
          "400": {
            "description": "Bad request",
            "content": {
              "application/json": {
                "schema": {
                  "type": "object",
                  "properties": {
                    "error": {
                      "type": "string"
                    }
                  }
                }
              }
            }
          },
          "500": {
            "description": "Server error",
            "content": {
              "application/json": {
                "schema": {
                  "type": "object",
                  "properties": {
                    "error": {
                      "type": "string"
                    },
                    "details": {
                      "type": "string"
                    }
                  }
                }
              }
            }
          }
        }
      }
    }
  }
}

{
  "openapi": "3.0.0",
  "info": {
    "title": "Food Search API",
    "version": "1.0.0",
    "description": "Searches for food places using a query and either a location string or coordinates."
  },
  "servers": [
    {
      "url": "url to your api",
      "description": "Azure Function App endpoint"
    }
  ],
  "paths": {
    "/api/foodsearch": {
      "post": {
        "summary": "Search food places",
        "operationId": "foodSearch",
        "requestBody": {
          "required": true,
          "content": {
            "application/json": {
              "schema": {
                "type": "object",
                "required": [
                  "query"
                ],
                "properties": {
                  "query": {
                    "type": "string",
                    "example": "pizza"
                  },
                  "cuisine": {
                    "type": "string",
                    "example": "Mexican"
                  },
                  "location": {
                    "type": "string",
                    "example": "Chicago, IL"
                  },
                  "latitude": {
                    "type": "number",
                    "example": 41.8781
                  },
                  "longitude": {
                    "type": "number",
                    "example": -87.6298
                  }
                }
              }
            }
          }
        },
        "responses": {
          "200": {
            "description": "Successful response",
            "content": {
              "application/json": {
                "schema": {
                  "type": "object",
                  "properties": {
                    "food_places": {
                      "type": "array",
                      "items": {
                        "type": "object",
                        "properties": {
                          "name": {
                            "type": "string"
                          },
                          "rating": {
                            "type": "number"
                          },
                          "reviews": {
                            "type": "integer"
                          }
                        }
                      }
                    }
                  }
                }
              }
            }
          },
          "400": {
            "description": "Bad request",
            "content": {
              "application/json": {
                "schema": {
                  "type": "object",
                  "properties": {
                    "error": {
                      "type": "string"
                    }
                  }
                }
              }
            }
          },
          "500": {
            "description": "Server error",
            "content": {
              "application/json": {
                "schema": {
                  "type": "object",
                  "properties": {
                    "error": {
                      "type": "string"
                    },
                    "details": {
                      "type": "string"
                    }
                  }
                }
              }
            }
          }
        }
      }
    }
  }
}

You are a smart travel assistant designed to help users find hotels and food spots in specific areas. Always respond clearly and helpfully, guiding users step by step if needed. Your main tools are hotel_searcher for hotels and food_searcher for food options. Use them precisely and effectively.

When helping a user:

    Use hotel_searcher to find hotels. Always ensure the location includes the city and state/province (e.g., "Austin, Texas" not just "Austin").

    Use food_searcher to find nearby places to eat. If the user is referencing a hotel, extract the latitude and longitude from the hotel result and pass them to food_searcher. This ensures an accurate nearby search.

    If the user is planning a trip, find the hotel first, then use its coordinates to search for food.

    Always explain what you’re doing in plain language so the user understands. If something is missing (like a city name or query), ask the user clearly and continue once the info is provided.

Be efficient, but conversational - your goal is to solve the user’s request using the tools as needed.


import weave
weave.init("az_agent")

from azure.ai.projects import AIProjectClient
from azure.identity import DefaultAzureCredential

project_client = AIProjectClient.from_connection_string(
    credential=DefaultAzureCredential(),
    conn_str="your_connection_string"
)

agent = project_client.agents.get_agent("asst_MVmqhbpGlAay9Bs2QBf7WLGJ")
thread = project_client.agents.create_thread()

@weave.op()
def ask_foundry_agent(project_client, agent, thread_id, user_input):
    project_client.agents.create_message(
        thread_id=thread_id,
        role="user",
        content=user_input
    )
    project_client.agents.create_and_process_run(
        thread_id=thread_id,
        agent_id=agent.id
    )
    messages = project_client.agents.list_messages(thread_id=thread_id).text_messages
    print(str(messages))
    if messages:
        return messages[-1].text.value
    else:
        return "No response from assistant."

print("Type 'exit' to end the conversation.")

while True:
    user_input = input("You: ")
    if user_input.strip().lower() in ("exit", "quit"):
        print("Chat ended.")
        break
    assistant_reply = ask_foundry_agent(project_client, agent, thread.id, user_input)
    print("Assistant:", assistant_reply)


import os
import asyncio
import base64
import hashlib
import pandas as pd
import json
import weave
from litellm import completion

from azure.ai.projects import AIProjectClient
from azure.identity import DefaultAzureCredential

CONNECTION_STRING = "your_connection_string"

AGENT_IDS = {
    "4o_web_searcher": "asst_vw3S54YgIzoy1WbdJWsceFiz"
}

# One client instance for efficiency
project_client = AIProjectClient.from_connection_string(
    credential=DefaultAzureCredential(),
    conn_str=CONNECTION_STRING
)

##############################
# 1. Data Preparation
##############################

def derive_key(password: str, length: int) -> bytes:
    key = hashlib.sha256(password.encode()).digest()
    return (key * (length // len(key))) + key[: length % len(key)]

def decrypt(ciphertext_b64: str, password: str) -> str:
    encrypted = base64.b64decode(ciphertext_b64)
    key = derive_key(password, len(encrypted))
    decrypted = bytes(a ^ b for a, b in zip(encrypted, key))
    return decrypted.decode(errors="ignore")

def load_eval_dataset(n=10):
    csv_url = "https://openaipublic.blob.core.windows.net/simple-evals/browse_comp_test_set.csv"
    df = pd.read_csv(csv_url)
    dataset = []
    for i, row in df.iterrows():
        if i >= n:
            break
        try:
            question = decrypt(row['problem'], row['canary'])
            answer = decrypt(row['answer'], row['canary'])   # <-- DECRYPT ANSWER TOO!
            dataset.append({"text": question, "label": answer})
        except Exception as e:
            print(f"Failed to decrypt row {i}: {e}")
    return dataset

##############################
# 2. Agent Interface
##############################

def agent_inference(agent_id: str, prompt: str, thread=None) -> str:
    """
    Run an inference with Azure AI agent and return the most recent message string value.
    Uses the last message.
    """
    if not thread:
        thread = project_client.agents.create_thread()
    # Post user's message
    project_client.agents.create_message(
        thread_id=thread.id,
        role="user",
        content=prompt
    )
    project_client.agents.create_and_process_run(
        thread_id=thread.id,
        agent_id=agent_id
    )
    messages = project_client.agents.list_messages(thread_id=thread.id)
    if messages.text_messages:
        latest = messages.text_messages[0].as_dict()
        text = latest.get("text", {}).get("value", "")
        return text
    return "[NO RESPONSE]"

##############################
# 3. Weave Model Wrapper
##############################

class FourOWebSearcherModel(weave.Model):
    @weave.op
    def predict(self, text: str) -> str:
        return agent_inference(AGENT_IDS["4o_web_searcher"], text)

##############################
# 4. Scoring Function
##############################

@weave.op
def gpt4o_scorer(label: str, model_output: str) -> dict:
    """Score the model's output by comparing it with the ground truth."""
    query = f"""
    YOU ARE A LLM JUDGE DETERMINING IF THE FOLLOWING MODEL GENERATED ANSWER IS THE SAME AS THE CORRECT ANSWER

    Model's Answer: {str(model_output)}
    Correct Answer: {label}

    Your task:
    1. State the model's predicted answer (answer only).
    2. State the ground truth (answer only).
    3. Determine if the model's final answer is correct (ignore formatting differences, etc.). RESPOND with the predicted and ground truth answer, followed with a JSON object containing the correctness encapsulated within these delimiters:
       ```json
       {{ "correctness": true/false }}
       ```
    """
    response = completion(
        model="gpt-4o-2024-08-06",
        temperature=0.0,
        messages=[{"role": "user", "content": query}]
    )
    # Parse response
    try:
        result = response.choices[0].message.content
        json_start = result.index("```json") + 7
        json_end = result.index("```", json_start)
        correctness = json.loads(result[json_start:json_end].strip()).get("correctness", False)
    except Exception as e:
        correctness = False

    return {"correctness": correctness, "reasoning": response.choices[0].message.content}

##############################
# 5. Main Evaluation Loop
##############################
async def run_evaluations():
    weave.init("azure_agent_eval")

    dataset = load_eval_dataset(100)
    print(f"Loaded {len(dataset)} examples.")

    models = {
        "4o_web_searcher": FourOWebSearcherModel(),
    }

    scorers = [gpt4o_scorer]

    for model_name, model in models.items():
        print(f"\n\n=== EVALUATING {model_name.upper()} ===")
        evaluation = weave.Evaluation(
            dataset=dataset,
            scorers=scorers,
            name=f"{model_name} Evaluation"
        )
        results = await evaluation.evaluate(model)
        print(f"Results for {model_name}: {results}")

if __name__ == "__main__":
    # os.environ["OPENAI_API_KEY"] = "your_openai_api_key"
    asyncio.run(run_evaluations())