From 0ab4015bd132ea1bd95b7a199bc79ce72dd3916e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8D=9A=E6=83=9F?= Date: Thu, 22 Jan 2026 16:52:06 +0800 Subject: [PATCH 1/3] feat(workflow): add Anthropic Messages API support for RL training Enable Anthropic-compatible agents to run through AReaL's proxy server for reinforcement learning workflows. This allows using Claude-style agents with the existing RL infrastructure. Key changes: - Add /v1/messages endpoint to proxy server with LiteLLM-based conversion - Introduce MathAgent workflow using Anthropic Messages API format - Add anthropic and litellm dependencies for API conversion --- .../openai/proxy/proxy_rollout_server.py | 73 +++++++++++++++++ areal/experimental/openai/proxy/server.py | 1 + areal/workflow/anthropic/__init__.py | 3 + areal/workflow/anthropic/math_agent.py | 82 +++++++++++++++++++ pyproject.toml | 2 + 5 files changed, 161 insertions(+) create mode 100644 areal/workflow/anthropic/__init__.py create mode 100644 areal/workflow/anthropic/math_agent.py diff --git a/areal/experimental/openai/proxy/proxy_rollout_server.py b/areal/experimental/openai/proxy/proxy_rollout_server.py index 28a8f6105b..b4bd19d592 100644 --- a/areal/experimental/openai/proxy/proxy_rollout_server.py +++ b/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 # ============================================================================= diff --git a/areal/experimental/openai/proxy/server.py b/areal/experimental/openai/proxy/server.py index aa366007cc..91a4a92be5 100644 --- a/areal/experimental/openai/proxy/server.py +++ b/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" diff --git a/areal/workflow/anthropic/__init__.py b/areal/workflow/anthropic/__init__.py new file mode 100644 index 0000000000..1184fa12da --- /dev/null +++ b/areal/workflow/anthropic/__init__.py @@ -0,0 +1,3 @@ +from areal.workflow.anthropic.math_agent import MathAgent + +__all__ = ["MathAgent"] diff --git a/areal/workflow/anthropic/math_agent.py b/areal/workflow/anthropic/math_agent.py new file mode 100644 index 0000000000..a0c8306428 --- /dev/null +++ b/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"], + ) diff --git a/pyproject.toml b/pyproject.toml index 5b15f8c565..ee71f3266a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -51,6 +51,8 @@ dependencies = [ "peft", "qwen_agent", "openai-agents", + "litellm", + "anthropic", # Visualization "pandas", From ff3be96214db073c5b0d4420944d1a3da90cbf25 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8D=9A=E6=83=9F?= Date: Thu, 22 Jan 2026 17:03:53 +0800 Subject: [PATCH 2/3] . --- .../openai/proxy/proxy_rollout_server.py | 14 ++++++- areal/workflow/anthropic/math_agent.py | 42 +++++++++---------- examples/experimental/proxy/config.yaml | 2 +- 3 files changed, 32 insertions(+), 26 deletions(-) diff --git a/areal/experimental/openai/proxy/proxy_rollout_server.py b/areal/experimental/openai/proxy/proxy_rollout_server.py index b4bd19d592..877d7f3943 100644 --- a/areal/experimental/openai/proxy/proxy_rollout_server.py +++ b/areal/experimental/openai/proxy/proxy_rollout_server.py @@ -446,11 +446,21 @@ async def anthropic_messages(raw_request: Request, session_id: str) -> Message: 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}") + except (ValueError, TypeError, KeyError) as e: + logger.warning( + f"Failed to convert Anthropic request to OpenAI format due to invalid input: {e}" + ) raise HTTPException( status_code=400, detail=f"Invalid Anthropic request format: {e}" ) + except Exception as e: + logger.error( + f"Unexpected error converting Anthropic request to OpenAI format: {e}", + exc_info=True, + ) + raise HTTPException( + status_code=500, detail="Internal server error during request conversion." + ) # Call OpenAI-compatible endpoint openai_response = await _call_client_create( diff --git a/areal/workflow/anthropic/math_agent.py b/areal/workflow/anthropic/math_agent.py index a0c8306428..db8c9e2312 100644 --- a/areal/workflow/anthropic/math_agent.py +++ b/areal/workflow/anthropic/math_agent.py @@ -1,16 +1,15 @@ +from math_verify import parse, verify + 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 +def math_reward_fn(completions: str, answer: str) -> float: + ans = parse(completions) + gold = parse(answer) + return float(verify(ans, gold)) class MathAgent(AgentWorkflow): @@ -19,8 +18,6 @@ class MathAgent(AgentWorkflow): 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. @@ -59,24 +56,23 @@ async def run(self, data: dict, **extra_kwargs) -> float: ) # 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 + response = await client.messages.create( + model="default", + messages=anthropic_messages, + system=system_prompt if system_prompt else anthropic.NOT_GIVEN, + **self.kwargs, + ) # Extract response text - completion_text = "" - for block in response.content: - if hasattr(block, "text"): - completion_text += block.text + completion_text = "".join( + block.text + for block in response.content + if isinstance(block, anthropic.types.TextBlock) + ) # Calculate reward - return await self.reward_fn( + reward_fn = AsyncRewardWrapper(math_reward_fn) + return await reward_fn( result=completion_text, answer=data["answer"], ) diff --git a/examples/experimental/proxy/config.yaml b/examples/experimental/proxy/config.yaml index d6b6c5e37e..e8b50fa924 100644 --- a/examples/experimental/proxy/config.yaml +++ b/examples/experimental/proxy/config.yaml @@ -34,7 +34,7 @@ rollout: openai: mode: inline tool_call_parser: qwen25 - reasoning_parser: qwen25 + reasoning_parser: qwen3 export_style: individual turn_discount: 1.0 From e497942f996bd4cde43caa71d638f524a5bbb7a4 Mon Sep 17 00:00:00 2001 From: "bowei.fw" Date: Thu, 22 Jan 2026 21:16:36 +0800 Subject: [PATCH 3/3] fix --- areal/workflow/anthropic/math_agent.py | 8 +++----- areal/workflow/openai/math_agent.py | 10 +++++++--- examples/experimental/proxy/config.yaml | 1 + examples/experimental/proxy/train.py | 9 ++++++--- 4 files changed, 17 insertions(+), 11 deletions(-) diff --git a/areal/workflow/anthropic/math_agent.py b/areal/workflow/anthropic/math_agent.py index db8c9e2312..98dc1f25ad 100644 --- a/areal/workflow/anthropic/math_agent.py +++ b/areal/workflow/anthropic/math_agent.py @@ -17,7 +17,8 @@ class MathAgent(AgentWorkflow): def __init__(self, **kwargs): # Store kwargs for client.messages.create call - self.kwargs = kwargs + self.kwargs = kwargs.copy() + self.kwargs.pop("max_completion_tokens", None) async def run(self, data: dict, **extra_kwargs) -> float: """Run the agent on a single problem. @@ -72,7 +73,4 @@ async def run(self, data: dict, **extra_kwargs) -> float: # Calculate reward reward_fn = AsyncRewardWrapper(math_reward_fn) - return await reward_fn( - result=completion_text, - answer=data["answer"], - ) + return await reward_fn(completion_text, data["answer"]) diff --git a/areal/workflow/openai/math_agent.py b/areal/workflow/openai/math_agent.py index 6126d7a2ab..a8628be750 100644 --- a/areal/workflow/openai/math_agent.py +++ b/areal/workflow/openai/math_agent.py @@ -23,7 +23,9 @@ def math_reward_fn(completions: str, answer: str) -> float: class MathAgent(AgentWorkflow): def __init__(self, **kwargs): - self.kwargs = kwargs + self.kwargs = kwargs.copy() + self.kwargs.pop("max_tokens", None).copy() + self.kwargs.pop("max_tokens", None) async def run(self, data: dict, **extra_kwargs): http_client = extra_kwargs.get("http_client", None) @@ -42,7 +44,8 @@ async def run(self, data: dict, **extra_kwargs): class MultiTurnMathAgent(AgentWorkflow): def __init__(self, max_turns: int = 8, **kwargs): self.max_turns = max_turns - self.kwargs = kwargs + self.kwargs = kwargs.copy() + self.kwargs.pop("max_tokens", None) async def run(self, data: dict, **extra_kwargs): http_client = extra_kwargs.get("http_client", None) @@ -116,7 +119,8 @@ def sqrt(a: float) -> float: class MathToolAgent(AgentWorkflow): def __init__(self, **kwargs): - self.kwargs = kwargs + self.kwargs = kwargs.copy() + self.kwargs.pop("max_tokens", None) async def run(self, data: dict, **extra_kwargs): http_client = extra_kwargs.get("http_client", None) diff --git a/examples/experimental/proxy/config.yaml b/examples/experimental/proxy/config.yaml index e8b50fa924..1a5ffe2ff5 100644 --- a/examples/experimental/proxy/config.yaml +++ b/examples/experimental/proxy/config.yaml @@ -42,6 +42,7 @@ gconfig: n_samples: 4 min_new_tokens: 0 max_new_tokens: 1024 + max_tokens: 2048 greedy: false temperature: 1.0 diff --git a/examples/experimental/proxy/train.py b/examples/experimental/proxy/train.py index 4cbd43cf64..6404ceb18c 100644 --- a/examples/experimental/proxy/train.py +++ b/examples/experimental/proxy/train.py @@ -23,9 +23,12 @@ def main(args): ) workflow_kwargs = dict( - max_completion_tokens=config.gconfig.max_new_tokens, temperature=config.gconfig.temperature, top_p=config.gconfig.top_p, + # For anthtropic + max_tokens=config.gconfig.max_tokens, + # For openai + max_completion_tokens=config.gconfig.max_new_tokens, ) eval_workflow_kwargs = workflow_kwargs.copy() eval_workflow_kwargs["temperature"] = 0.6 @@ -36,8 +39,8 @@ def main(args): valid_dataset=valid_dataset, ) as trainer: trainer.train( - workflow="areal.workflow.openai.math_agent.MathAgent", - eval_workflow="areal.workflow.openai.math_agent.MathAgent", + workflow="areal.workflow.anthropic.math_agent.MathAgent", + eval_workflow="areal.workflow.anthropic.math_agent.MathAgent", workflow_kwargs=workflow_kwargs, eval_workflow_kwargs=eval_workflow_kwargs, )