feat(workflow): add Anthropic Messages API support for RL training

areal/experimental/openai/proxy/proxy_rollout_server.py

@@ -8,8 +8,13 @@
 from typing import TYPE_CHECKING, Any
 
 import uvicorn
+from anthropic.types.message import Message
 from fastapi import Depends, FastAPI, HTTPException, Request
 from fastapi.exceptions import RequestValidationError
+from litellm.llms.anthropic.experimental_pass_through.adapters.transformation import (
+    AnthropicAdapter,
+)
+from litellm.types.utils import ModelResponse as LitellmModelResponse
 from pydantic import BaseModel
 
 from openai.types.chat import ChatCompletion
@@ -27,6 +32,7 @@
 from areal.utils.network import find_free_ports, gethostip
 
 from .server import (
+    ANTHROPIC_MESSAGES_PATHNAME,
     CHAT_COMPLETIONS_PATHNAME,
     RESPONSES_PATHNAME,
     RL_END_SESSION_PATHNAME,
@@ -75,6 +81,8 @@
 _nfs_record_root: str = "/tmp/areal/name_resolve"
 _etcd3_addr: str = "localhost:2379"
 
+# Adapter to convert Anthropic request to OpenAI format
+_adapter = AnthropicAdapter()
 
 # =============================================================================
 # Request Validation
@@ -408,6 +416,71 @@ async def responses(request: ResponseCreateParams, session_id: str) -> Response:
     )
 
 
+@app.post(
+    "/{session_id}/" + ANTHROPIC_MESSAGES_PATHNAME,
+    dependencies=[Depends(validate_json_request)],
+)
+async def anthropic_messages(raw_request: Request, session_id: str) -> Message:
+    """Anthropic Messages API compatible endpoint.
+
+    Converts Anthropic format requests to OpenAI format, processes through
+    the OpenAI-compatible endpoint, then converts the response back to
+    Anthropic format.
+
+    Uses LiteLLM's AnthropicAdapter for format conversion.
+    """
+
+    if _openai_client is None:
+        raise HTTPException(
+            status_code=500,
+            detail='Proxy server not initialized. Send requests to /create_engine then /call "initialize" first.',
+        )
+
+    # Parse Anthropic request
+    anthropic_request = await raw_request.json()
+
+    try:
+        openai_request = _adapter.translate_completion_input_params(
+            anthropic_request.copy()
+        )
+        if openai_request is None:
+            raise ValueError("Failed to translate request")
+        openai_request = dict(openai_request)
+    except Exception as e:
+        logger.error(f"Failed to convert Anthropic request to OpenAI format: {e}")
+        raise HTTPException(
+            status_code=400, detail=f"Invalid Anthropic request format: {e}"
+        )
+
+    # Call OpenAI-compatible endpoint
+    openai_response = await _call_client_create(
+        create_fn=_openai_client.chat.completions.create,
+        request=openai_request,
+        session_id=session_id,
+    )
+
+    # Convert OpenAI response to Anthropic format using LiteLLM's adapter
+    try:
+        # Convert ChatCompletion to LitellmModelResponse
+        openai_response_dict = openai_response.model_dump()
+        model_response = LitellmModelResponse(**openai_response_dict)
+        anthropic_response = _adapter.translate_completion_output_params(model_response)
+        if anthropic_response is None:
+            raise ValueError("Failed to translate response")
+
+        # LiteLLM returns Pydantic BaseModel objects in content list,
+        # Convert them to dict.
+        if "content" in anthropic_response and anthropic_response["content"]:
+            anthropic_response["content"] = [
+                block.model_dump() if hasattr(block, "model_dump") else block
+                for block in anthropic_response["content"]
+            ]
+        return Message(**anthropic_response)
+    except Exception as e:
+        logger.error(f"Failed to convert OpenAI response to Anthropic format: {e}")
+        raise HTTPException(status_code=500, detail=f"Failed to convert response: {e}")
+
+
 # =============================================================================
 # Trajectory Export
 # =============================================================================

areal/experimental/openai/proxy/server.py

@@ -160,3 +160,4 @@ def deserialize_interactions(
 RL_SET_REWARD_PATHNAME = "rl/set_reward"
 CHAT_COMPLETIONS_PATHNAME = "chat/completions"
 RESPONSES_PATHNAME = "responses"
+ANTHROPIC_MESSAGES_PATHNAME = "v1/messages"

areal/workflow/anthropic/__init__.py

@@ -0,0 +1,3 @@
+from areal.workflow.anthropic.math_agent import MathAgent
+
+__all__ = ["MathAgent"]

areal/workflow/anthropic/math_agent.py

@@ -0,0 +1,82 @@
+import anthropic
+
+from areal.api.reward_api import AsyncRewardWrapper
+from areal.api.workflow_api import AgentWorkflow
+from areal.reward import get_math_verify_worker
+
+
+def gsm8k_reward_fn(result, answer):
+    try:
+        worker = get_math_verify_worker()
+        return worker.verify(str(result), str(answer))
+    except Exception:
+        return 0.0
+
+
+class MathAgent(AgentWorkflow):
+    """Simple single-turn math agent using Anthropic Messages API."""
+
+    def __init__(self, **kwargs):
+        # Store kwargs for client.messages.create call
+        self.kwargs = kwargs
+        # Cache AsyncRewardWrapper to avoid creating new ProcessPoolExecutor per call
+        self.reward_fn = AsyncRewardWrapper(gsm8k_reward_fn)
+
+    async def run(self, data: dict, **extra_kwargs) -> float:
+        """Run the agent on a single problem.
+
+        Args:
+            data: Input data containing "messages" and "answer"
+            **extra_kwargs: Contains base_url and http_client from proxy
+
+        Returns:
+            Reward value for this trajectory
+        """
+        http_client = extra_kwargs.get("http_client", None)
+        base_url = extra_kwargs.get("base_url", None)
+
+        client = anthropic.AsyncAnthropic(
+            api_key="placeholder",
+            base_url=base_url,
+            http_client=http_client,
+            max_retries=0,
+        )
+
+        # Convert OpenAI-style messages to Anthropic format
+        messages = data["messages"]
+        system_prompt = None
+        anthropic_messages = []
+
+        for msg in messages:
+            if msg["role"] == "system":
+                system_prompt = msg["content"]
+            else:
+                anthropic_messages.append(
+                    {
+                        "role": msg["role"],
+                        "content": msg["content"],
+                    }
+                )
+
+        # Call the Anthropic Messages API (via proxy)
+        try:
+            response = await client.messages.create(
+                model="default",
+                messages=anthropic_messages,
+                system=system_prompt if system_prompt else anthropic.NOT_GIVEN,
+                **self.kwargs,
+            )
+        except Exception:
+            return 0.0
+
+        # Extract response text
+        completion_text = ""
+        for block in response.content:
+            if hasattr(block, "text"):
+                completion_text += block.text
+
+        # Calculate reward
+        return await self.reward_fn(
+            result=completion_text,
+            answer=data["answer"],
+        )

pyproject.toml

@@ -51,6 +51,8 @@ dependencies = [
     "peft",
     "qwen_agent",
     "openai-agents",
+    "litellm",
+    "anthropic",
 
     # Visualization
     "pandas",