Feature request: Send Traces to Observability Frameworks

[maxritter]

Use case

For production use-cases, we need some way to send traces of the orchestrator and agents to an Observability Framework like Langfuse, so they can be visualized there nicely and combined with user feedback.

Right now, it is unclear to me how this is possible. Only getting text from logs is not sufficient for debugging when a lot of users utilize the system.

Solution/User Experience

Need some kind of mechanism to write traces to external SDKs like the one from Langfuse (https://langfuse.com/docs/integrations/amazon-bedrock).

Alternative solutions

No response

[cornelcroi]

Hi @maxritter actually you can create a custom logger (or use your own) that can send the logs where you need.
Have a look here https://awslabs.github.io/multi-agent-orchestrator/cookbook/monitoring/logging/

[maxritter]

@cornelcroi Thanks a lot for the hint!. With that I could at least use https://github.com/kislyuk/watchtower to push those logs to Cloudwatch and then search for them with the trace information that I record via the Python SDK before / after a model invication in LangFuse (https://langfuse.com/docs/sdk/python/example).

If anybody knows a smarter way to handle observability on Bedrock Agents, I am curious to know!

[brnaba-aws]

For observability with Bedrock Agents, we could also forward the trace to the Logger.

[maxritter]

@brnaba-aws Thanks for your help. I will try to implement this to send the input, output and reasoning steps to LangFuse like in this example from AWS (https://aws.amazon.com/de/blogs/machine-learning/accelerate-analysis-and-discovery-of-cancer-biomarkers-with-amazon-bedrock-agents/):

[brnaba-aws]

Hi @maxritter
Please let us know how it goes, we might provide this as an example in our repo for others.

[brnaba-aws]

@maxritter
Here is a suggestion:

from typing import Optional

def create_observability_decorator(
    framework_name: str = "langfuse",
    config: Optional[Dict] = None
):
    """
    Factory function that returns the appropriate decorator based on the framework
    with optional configuration
    """
    config = config or {}
    
    if framework_name == "langfuse":
        from langfuse.decorators import observe
        return observe(**config)
    elif framework_name == "opentelemetry":
        from opentelemetry.trace import trace
        return trace(**config)
    elif framework_name == "honeycomb":
        from honeycomb.trace import span
        return span(**config)
    else:
        def noop_decorator(func):
            @wraps(func)
            async def wrapper(*args, **kwargs):
                return await func(*args, **kwargs) if inspect.iscoroutinefunction(func) else func(*args, **kwargs)
            return wrapper
        return noop_decorator

# Usage example with configuration:
observe_llm = create_observability_decorator(
    framework_name="langfuse",
    config={"name": "llm_request", "capture_input": True}
)

@observe_llm
async def handle_single_response(self, input_data: Dict) -> Any:
    # ... rest of the code

Do you believe this is sufficient?
Side note: I didn't test :-)