Tutorial: Using watsonx Orchestrate and BeeAI Framework

# watsonx Configuration
WATSONX_URL=https://us-south.ml.cloud.ibm.com
WATSONX_REGION=us-south
WATSONX_PROJECT_ID=your-project-id-here
WATSONX_API_KEY=your-watsonx-api-key-here

# External Agent Security
API_KEY=your-secure-random-key

# Optional: Web Search
TAVILY_API_KEY=your-tavily-api-key-here

from dotenv import load_dotenv
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict

load_dotenv()

__all__ = ["AppSettings"]

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="", str_to_upper=False)
    api_key: str = Field(..., description="Project API Key")
    watsonx_url: str
    watsonx_project_id: str
    watsonx_api_key: str
    tavily_api_key: str
    log_intermediate_steps: bool = Field(default=False)
    watsonx_default_model: str = "ibm/granite-3-3-8b-instruct"

AppSettings = Settings()

from beeai_framework.tools import tool
from tavily import TavilyClient
from wxo_beeai.settings import AppSettings

@tool
def search_web_tool(query: str) -> dict[str, str] | None:
    """
    Searches the web for the given query and returns the most relevant results.
    """
    client = TavilyClient(api_key=AppSettings.tavily_api_key)

    results = client.search(
        query,
        search_depth="advanced",
        max_results=10,
        include_images=False,
        include_answer=False,
    )

    return results

import logging
import sys
import traceback
import warnings

from beeai_framework.agents.requirement import RequirementAgent
from beeai_framework.agents.requirement.requirements.conditional import (
    ConditionalRequirement,
)
from beeai_framework.errors import FrameworkError
from beeai_framework.middleware.trajectory import GlobalTrajectoryMiddleware
from beeai_framework.tools import Tool
from beeai_framework.tools.think import ThinkTool
from beeai_framework.adapters.watsonx_orchestrate import (
    WatsonxOrchestrateServer,
    WatsonxOrchestrateServerConfig,
)
from beeai_framework.memory import UnconstrainedMemory
from beeai_framework.adapters.watsonx import WatsonxChatModel
from wxo_beeai.settings import AppSettings
from wxo_beeai.tools import search_web_tool

warnings.filterwarnings("ignore")

logger = logging.getLogger()
logger.setLevel(logging.INFO)
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)

def main() -> None:

    memory = UnconstrainedMemory()

    llm = WatsonxChatModel(
        model_id=AppSettings.watsonx_default_model.removeprefix("watsonx/").lower(),
        api_key=AppSettings.watsonx_api_key,
        project_id=AppSettings.watsonx_project_id,
        base_url=AppSettings.watsonx_url,
        middlewares=[
            GlobalTrajectoryMiddleware(enabled=AppSettings.log_intermediate_steps)
        ],
    )

    agent = RequirementAgent(
        llm=llm,
        memory=memory,
        tools=[ThinkTool(), search_web_tool],
        role="a deep researcher",
        instructions=[
            "Your task is to conduct in-depth research on the given topic.",
            "Before you start, thoroughly prepare a step-by-step plan for how you will solve the task.",
            "After each action, reflect on what you have obtained and what you need to do to gather evidence for the final answer.",
        ],
        middlewares=[
            GlobalTrajectoryMiddleware(
                included=[Tool], enabled=AppSettings.log_intermediate_steps
            )
        ],
        requirements=[
            ConditionalRequirement(ThinkTool, consecutive_allowed=False),
            ConditionalRequirement(
                ThinkTool, force_at_step=1, force_after=[search_web_tool]
            ),
        ],
    )

    config = WatsonxOrchestrateServerConfig(
        port=8080, host="0.0.0.0", api_key=AppSettings.api_key
    )

    server = WatsonxOrchestrateServer(config=config)
    server.register(agent)
    server.serve()

if __name__ == "__main__":
    try:
        main()
    except FrameworkError as e:
        traceback.print_exc()
        sys.exit(e.explain())

FROM python:3.11-slim AS python-base

# https://python-poetry.org/docs#ci-recommendations
ENV POETRY_VERSION=2.1.1
ENV POETRY_HOME=/opt/poetry
ENV POETRY_VENV=/opt/poetry-venv

# Tell Poetry where to place its cache and virtual environment
ENV POETRY_CACHE_DIR=/opt/.cache

# Create stage for Poetry installation
FROM python-base AS poetry-base

# Creating a virtual environment just for poetry and install it with pip
RUN python3 -m venv $POETRY_VENV \
	&& $POETRY_VENV/bin/pip install -U pip setuptools \
	&& $POETRY_VENV/bin/pip install poetry==${POETRY_VERSION}

# Create a new stage from the base python image
FROM python-base AS example-app

# Copy Poetry to app image
COPY --from=poetry-base ${POETRY_VENV} ${POETRY_VENV}

# Add Poetry to PATH
ENV PATH="${PATH}:${POETRY_VENV}/bin"

WORKDIR /app

# Copy Dependencies
COPY poetry.lock pyproject.toml ./

# Tell Poetry to not use its own venv
RUN poetry config virtualenvs.create false

# [OPTIONAL] Validate the project is properly configured
RUN poetry check

# Copy Application
COPY . /app

# Install Dependencies
RUN poetry install --no-interaction --no-cache

# Run Application
EXPOSE 8080

ENTRYPOINT ["poetry", "run", "python", "wxo_beeai/app.py"]

services:
  app:
    container_name: wxo_beeai_test
    build:
      context: .
      dockerfile: Dockerfile
    ports:
      - "8080:8080"
    env_file:
      - .env
    environment:
      - PYTHONUNBUFFERED=1
    restart: no