Bedrock AgentCore 에이전트에 관측성과 트레이싱 추가하기

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Dict, Any
from datetime import datetime

from strands import Agent, tool
from strands.models import BedrockModel

### --- Define your tools
@tool
def word_count(text: str) -> int:
    """Returns the number of words in the text."""
    return len(text.split())

@tool
def reverse(text: str) -> str:
    """Reverses the input string."""
    return text[::-1]

### --- Configure your Bedrock model provider
bedrock_model = BedrockModel(
    model_id="anthropic.claude-3-5-sonnet-20240620-v1:0",
    temperature=0.3,
    streaming=False,  # Set to True if you want streaming output
    region_name="us-east-1",  # <--- your AWS region
)

### --- Initialize the agent WITH tools
strands_agent = Agent(
    model=bedrock_model,
    tools=[word_count, reverse],
    system_prompt="You are a helpful assistant who uses tools when they help."
)

### --- FastAPI app setup
app = FastAPI(
    title="Strands Agent Server",
    version="1.0.0"
)

class InvocationRequest(BaseModel):
    input: Dict[str, Any]

class InvocationResponse(BaseModel):
    output: Dict[str, Any]

@app.post("/invocations", response_model=InvocationResponse)
async def invoke_agent(request: InvocationRequest):
    try:
        user_message = request.input.get("prompt", "")
        if not user_message:
            raise HTTPException(
                status_code=400,
                detail="No prompt found in input. Please provide a 'prompt' key in the input."
            )
        # Call the strands agent synchronously
        result = strands_agent(user_message)

        response = {
            "message": result.message,  # agent reply object
            "timestamp": datetime.utcnow().isoformat(),
            "model": bedrock_model.config["model_id"],
        }
        return InvocationResponse(output=response)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Agent processing failed: {str(e)}")

@app.get("/ping")
async def ping():
    return {"status": "healthy"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)

import os
import json
from datetime import datetime
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Dict, Any

import weave

from strands import Agent, tool
from strands.models import BedrockModel

# Export WANDB key for Weave auth
os.environ["WANDB_API_KEY"] = "your_api_key"

WEAVE_PROJECT = os.getenv("WEAVE_PROJECT", "your_wandb_username/wand_project_name")
weave.init(WEAVE_PROJECT)

@weave.op()
def word_count_op(text: str) -> int:
    return len(text.split())

@weave.op()
def reverse_op(text: str) -> str:
    return text[::-1]

def get_agent():
    @tool
    def word_count(text: str) -> int:
        return word_count_op(text)

    @tool
    def reverse(text: str) -> str:
        return reverse_op(text)

    bedrock_model = BedrockModel(
        model_id="anthropic.claude-3-5-sonnet-20240620-v1:0",
        temperature=0.3,
        streaming=False,
        region_name="us-east-1",
    )

    return Agent(
        model=bedrock_model,
        tools=[word_count, reverse],
        system_prompt="You are a helpful assistant who uses tools when they help.",
    )

@weave.op()
def run_agent(agent: Agent, user_message: str) -> Dict[str, Any]:
    result = agent(user_message)
    return {
        "message": result.message,
        "model": agent.model.config["model_id"],
    }

app = FastAPI()

class InvocationRequest(BaseModel):
    input: Dict[str, Any]

class InvocationResponse(BaseModel):
    output: Dict[str, Any]

@app.post("/invocations", response_model=InvocationResponse)
async def invoke_agent(body: InvocationRequest):
    agent = get_agent()
    user_message = body.input.get("prompt", "")
    if not user_message:
        raise HTTPException(400, "No prompt found; provide 'prompt' key.")

    result = run_agent(agent, user_message)

    response = {
        "message": result["message"],
        "timestamp": datetime.utcnow().isoformat(),
        "model": result["model"],
    }
    return InvocationResponse(output=response)

@app.get("/ping")
async def ping():
    return {"status": "healthy"}

import functools
import base64
import os
import json
from datetime import datetime
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Dict, Any

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace.export import SimpleSpanProcessor

from strands import Agent, tool
from strands.models import BedrockModel

# ─── Weave OTLP setup ────────────────────────────────────────────────────────────
WANDB_API_KEY = os.getenv("WANDB_API_KEY", "your_api_key")
WEAVE_PROJECT = "your_wandb_username/wand_project_name"

auth_b64 = base64.b64encode(f"api:{WANDB_API_KEY}".encode()).decode()
exporter = OTLPSpanExporter(
    endpoint="https://trace.wandb.ai/otel/v1/traces",
    headers={"Authorization": f"Basic {auth_b64}", "project_id": WEAVE_PROJECT},
)
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("strands-agent")
# ────────────────────────────────────────────────────────────────────────────────

def tool_logger(tool_calls):
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            result = fn(*args, **kwargs)
            tool_calls.append({
                "tool_name": fn.__name__,
                "tool_input": {"args": args, "kwargs": kwargs},
                "tool_output": result,
            })
            return result
        return wrapper
    return decorator

def get_agent(tool_calls):
    @tool
    def word_count(text: str) -> int:
        return tool_logger(tool_calls)(lambda t: len(t.split()))(text)

    @tool
    def reverse(text: str) -> str:
        return tool_logger(tool_calls)(lambda t: t[::-1])(text)

    bedrock_model = BedrockModel(
        model_id="anthropic.claude-3-5-sonnet-20240620-v1:0",
        temperature=0.3,
        streaming=False,
        region_name="us-east-1",
    )

    return Agent(
        model=bedrock_model,
        tools=[word_count, reverse],
        system_prompt="You are a helpful assistant who uses tools when they help.",
    )

app = FastAPI()

class InvocationRequest(BaseModel):
    input: Dict[str, Any]

class InvocationResponse(BaseModel):
    output: Dict[str, Any]

@app.post("/invocations", response_model=InvocationResponse)
async def invoke_agent(body: InvocationRequest):
    tool_calls = []
    agent = get_agent(tool_calls)
    user_message = body.input.get("prompt", "")
    if not user_message:
        raise HTTPException(400, "No prompt found; provide 'prompt' key.")

    with tracer.start_as_current_span("invoke_agent") as span:
        span.set_attribute("input.value", json.dumps({"prompt": user_message}))

        result = agent(user_message)

        span.set_attribute("output.value", json.dumps({
            "message": result.message,
            "model": agent.model.config["model_id"],
            "tool_calls": tool_calls,
        }))

    response = {
        "message": result.message,
        "timestamp": datetime.utcnow().isoformat(),
        "model": agent.model.config["model_id"],
        "tool_calls": tool_calls,
    }
    return InvocationResponse(output=response)

@app.get("/ping")
async def ping():
    return {"status": "healthy"}