AWS Bedrock Agents

A multi-agent pipeline simulating AWS Bedrock Agent execution with action groups and knowledge base retrieval. The orchestrator prepares an agent invocation, selects an action group, dispatches a runner child agent that retrieves from a knowledge base, invokes action groups, and reasons with an LLM, then dispatches an evaluator child agent that assesses response quality with @reasoning and generates a structured report.

Environment variables

This example requires OPENAI_API_KEY, WAXELL_API_KEY, and WAXELL_API_URL. Use --dry-run to run without any API keys.

Architecture

Key Code

Knowledge base retrieval and action group invocation

The runner child agent performs a multi-step workflow: retrieve from a knowledge base, search with ranking, invoke an action group, and reason about results with an LLM.

@waxell.observe(agent_name="bedrock-agents-runner", workflow_name="bedrock-agent-execution")
async def run_bedrock_agent(query: str, action_group: str, openai_client, waxell_ctx=None):
    waxell.tag("agent_role", "runner")
    waxell.tag("framework", "bedrock_agents")

    # Knowledge base retrieval via @tool
    kb_result = bedrock_kb_retrieve(knowledge_base_id=MOCK_KB_ID, query=query)

    # Ranked search via @retrieval
    ranked_results = search_knowledge_base(query=query, kb_id=MOCK_KB_ID)

    # Action group execution via @tool
    action_result = bedrock_action_invoke(
        action_group=action_group, action="lookupOrder", parameters={"query": query},
    )

    # LLM reasoning about action result (auto-instrumented)
    resp = await openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You are a Bedrock Agent. Process the order lookup."},
            {"role": "user", "content": query},
        ],
    )
    return {"kb_results": ranked_results, "order_data": action_result.get("result", {})}

Tool and retrieval decorators

Both @tool (for function calls) and @retrieval (for search operations) are used to create typed spans.

@waxell.tool(tool_type="retrieval")
def bedrock_kb_retrieve(knowledge_base_id: str, query: str) -> dict:
    """Retrieve results from a Bedrock knowledge base."""
    return {
        "knowledge_base_id": knowledge_base_id,
        "results": MOCK_KB_RESULTS,
        "top_score": max(r["score"] for r in MOCK_KB_RESULTS),
    }


@waxell.retrieval(source="bedrock_knowledge_base")
def search_knowledge_base(query: str, kb_id: str) -> list[dict]:
    """Search Bedrock knowledge base and return ranked results."""
    return [{"text": r["text"], "score": r["score"], "source": r["source"]} for r in MOCK_KB_RESULTS]

What this demonstrates

• @waxell.observe -- parent orchestrator with 2 child agents
• @waxell.step_dec -- agent invocation preparation
• @waxell.decision -- action group selection
• waxell.decide() -- manual inline decision for response routing
• @waxell.tool -- typed tool operations (retrieval, function)
• @waxell.retrieval -- dedicated retrieval decorator for KB search
• @waxell.reasoning_dec -- response quality assessment
• waxell.score() -- accuracy and resolution scores (including boolean type)
• Bedrock Agents patterns -- KB retrieval, action groups, and agent evaluation

Run it

cd dev/waxell-dev
python -m app.demos.bedrock_agents_agent --dry-run

Source

dev/waxell-dev/app/demos/bedrock_agents_agent.py