> ## Documentation Index > Fetch the complete documentation index at: https://docs.bailian.console.aliyun.com/llms.txt > Use this file to discover all available pages before exploring further. # Agentic OPD Development > Agentic OPD requires a custom Rollout (you control the generation process yourself, such as multi-turn interaction, passing the tools parameter to produce structured tool_call). Reward is optional. This article introduces function development for both the Reward-disabled and Reward-enabled modes. This article introduces function development for Agentic OPD. Agentic OPD requires a custom Rollout (you control the generation process yourself, such as multi-turn dialogue, custom interaction, passing the `tools` parameter to produce structured `tool_call` (tool calls), etc.); Reward is optional. It consists of three parts: tool definitions + Rollout (sampling to produce trajectories) + Reward (scoring, optional). For the training overview, see [Online Policy Distillation Training Overview](opd-training-overview.md). ## Development Overview The key difference between Agentic OPD and Model OPD is whether tool calls, multi-turn, or custom interactions are involved: 1. **Model OPD** is used for pure-text tasks, where the student model directly generates text output. 2. **Agentic OPD** is used for tasks requiring tool calls or multi-turn interaction. By passing the `tools` parameter, the framework passes tool information directly to the model, and the model returns a structured `tool_calls` field (native tool calls). When Reward is not enabled, the functions list only contains `RolloutFunctionComponent`, and the training signal is dominated by teacher-model distillation. The differences between the two modes are as follows.

Dimension

No Reward (only Rollout)

With Reward (Rollout + Reward)

Development Work

Write the Rollout function (custom generation process)

Additionally write a Reward scoring function

Training Signal

Teacher-model distillation as the only supervision

Teacher distillation + task reward

Applicable Scenario

Only tool-call shape alignment needed, teacher model already the ceiling

Need to optimize tool selection and parameters by business goal

## Data Format One JSONL record per line, containing two fields: `messages` (conversation) and `rollout_extra` (additional information). - `messages`: the conversation list, arranged with alternating user/assistant turns; the first turn can be user or system. - `rollout_extra`: additional information that can store sample\_id, category, solution, etc. `solution` is a JSON string storing the standard answer for Reward scoring; when Reward is not enabled it serves only as a marker and is not consumed. Example (actually a single line; expanded here for readability). The tools in the example (such as `query_status`, `delete_item`) are defined in the [Tool Definitions](#tool-definitions) section below, and Rollout injects tools by passing the `tools` parameter—see the [Rollout Function Structure](#rollout-structure) section: The user asks about a project status; the standard answer is to call the `query_status` tool to query: ```json { "messages": [ {"role": "user", "content": "查询项目 P100 的状态"} ], "rollout_extra": { "sample_id": "train-query-001", "solution": "{\"decision\":\"tool_call\",\"tool_calls\":[{\"name\":\"query_status\",\"arguments\":{\"item_id\":\"P100\"}}]}" } } ``` The user wants to delete a project; the assistant confirms first, and after the user confirms, the standard answer is to call the `delete_item` tool: ```json { "messages": [ {"role": "user", "content": "删除项目 P200"}, {"role": "assistant", "content": "CONFIRM: 请确认删除项目 P200"}, {"role": "user", "content": "确认删除"} ], "rollout_extra": { "solution": "{\"decision\":\"tool_call\",\"tool_calls\":[{\"name\":\"delete_item\",\"arguments\":{\"item_id\":\"P200\"}}]}" } } ``` The user provides insufficient information (no project number); the standard answer is to clarify rather than call a tool: ```json { "messages": [ {"role": "user", "content": "项目编号是什么？"} ], "rollout_extra": { "sample_id": "train-clarify-003", "solution": "{\"decision\":\"clarify\",\"mentions\":[\"编号\"]}" } } ``` ## Rollout Function Development Agentic OPD requires a custom Rollout (you control the generation process yourself). This section covers Rollout function development: tool definitions, Rollout function structure, and Tracing integration. ### Tool Definitions Tool definitions are a prerequisite for Rollout. First define the tool\_spec, then implement the Rollout processor; both belong to the `functions/` subpackage. The directory structure is as follows (`__init__.py` makes the directory an importable Python package; without it, classpath resolution fails): ```plaintext functions/ ├── __init__.py ├── tool_spec.py # SYSTEM_PROMPT、TOOL_DEFINITIONS、DESTRUCTIVE_TOOLS ├── rollout/ │ ├── __init__.py │ └── my_rollout.py # MyRolloutProcessor └── reward/ ├── __init__.py └── my_reward.py # MyRewardProcessor（启用 Reward 时） ``` The `classpath` of the submission script points to `functions.rollout.my_rollout.MyRolloutProcessor` (i.e., the `MyRolloutProcessor` class inside `functions/rollout/my_rollout.py`); the Reward class likewise points to `functions.reward.my_reward.MyRewardProcessor`. Tool definitions consist of three parts: - **System Prompt**: constrains when to call a tool and when to reply with a text prefix - **Tool Definition List**: a JSON Schema list, directly used by passing the `tools` parameter inside Rollout; the framework automatically renders tool information to the model; the format is consistent with OpenAI function calling - **High-Risk Operation Set**: marks destructive tool names so that Reward can judge whether the model mis-invoked a dangerous tool Difference from Model OPD: Model OPD is for pure-text tasks; Agentic OPD is for tasks requiring tool calls or multi-turn interaction, producing native `tool_calls` by passing the `tools` parameter through the framework. ```plaintext # 通用化工具定义模板 # TOOL_DEFINITIONS：工具定义列表，供 Rollout 传入 tools 参数，框架自动渲染给模型 # DESTRUCTIVE_TOOLS：破坏性工具名集合，供 Reward 判断模型是否误调危险工具 # 系统提示词：约束模型何时调工具、何时回退为文本前缀（CLARIFY/NO_TOOL/CONFIRM） SYSTEM_PROMPT = """You are a tool-routing assistant. Call tools only when the request is clear and all required arguments are known. If required information is missing, reply exactly with CLARIFY: . If no tool is appropriate, reply exactly with NO_TOOL: . Destructive actions require explicit user confirmation first; otherwise reply with CONFIRM: .""" # 工具定义列表，格式与 OpenAI function calling 一致，每项含 type/function(name/description/parameters) TOOL_DEFINITIONS = [ { "type": "function", "function": { "name": "query_status", # 查询类工具：按 item_id 返回状态 "description": "Look up the status of an item by its identifier.", "parameters": { "type": "object", "properties": { "item_id": { # 必填参数：待查询项目的标识符 "type": "string", "description": "Identifier of the item to query.", } }, "required": ["item_id"], # 声明 item_id 为必填 "additionalProperties": False, # 禁止额外字段，校验更严格 }, }, }, { "type": "function", "function": { "name": "delete_item", # 破坏性工具：删除项目，须用户确认后才执行 "description": "Delete an item after the user explicitly confirms.", "parameters": { "type": "object", "properties": { "item_id": {"type": "string"}, # 待删除项目标识符 "reason": {"type": "string"}, # 删除原因，留痕审计 }, "required": ["item_id", "reason"], # 两参数均必填 "additionalProperties": False, }, }, }, # 按任务添加更多工具定义 ] # 高危操作名称集合，供 Reward 判断模型是否误调危险工具 DESTRUCTIVE_TOOLS = { "delete_item", # 按任务添加更多高危工具名 } ``` ### Rollout Function Structure Inherit `AbstractRolloutProcessor` and implement `async process(input: RolloutInput) -> RolloutOutput`. Decorate `process` with `@observe_processor` (automatically integrates Tracing). Key steps: 1. **Retrieve model resources**: take `model_resource` (api\_key/base\_url/model\_name) and `sampling_params` from `RolloutInput` 2. **Construct and wrap the LLM**: construct `ChatOpenAI`, call `trace_client` to wrap it 3. **Sample**: pass the `tools` parameter to inject tool definitions, `ainvoke` to sample (single turn, without executing tools) 4. **Construct output**: construct `AgentOutput` (containing native `tool_calls`, `rollout_metrics`), placed in `RolloutOutput.agent_output` 5. **Exception fallback**: return the `ROLLOUT_ERROR` fallback text and set status to FAILED Data Flow: `RolloutInput → AgentOutput → RolloutOutput`. ```plaintext # 通用化 Agentic OPD Rollout 处理器模板 import time import json from typing import Any from langchain_core.messages import AIMessage, HumanMessage, SystemMessage # langchain 消息类型，对应 user/assistant/system from langchain_openai import ChatOpenAI # 封装 OpenAI 兼容接口的 LLM 客户端 from dashscope.finetune.reinforcement import RolloutInput, RolloutOutput # 输入/输出数据模型 from dashscope.finetune.reinforcement.component.data.base_data_model import ( AgentOutput, # 学生输出，含 messages/tool_calls/rollout_metrics TaskStatus, # 任务状态枚举：SUCCESS / FAILED ) from dashscope.finetune.reinforcement.component.observability import ( observe_processor, # 装饰器：自动为 process 接入 Tracing Span trace_client, # 包装自构造 LLM 客户端，使其请求产生 LLM Span ) from dashscope.finetune.reinforcement.component.processor.abstract_rollout_processor import ( AbstractRolloutProcessor, # 抽象基类，子类须实现 async process ) from functions.tool_spec import SYSTEM_PROMPT, TOOL_DEFINITIONS # 工具定义与系统提示词 def _to_langchain_messages(messages): """将 dict messages 转为 langchain 消息对象，首条注入 SYSTEM_PROMPT。""" converted = [SystemMessage(content=SYSTEM_PROMPT)] # 默认塞入系统提示词 for msg in messages: role = msg.get("role") content = str(msg.get("content") or "") if role == "system": converted[0] = SystemMessage(content=content) # 数据自带 system 则覆盖默认 elif role == "assistant": converted.append(AIMessage(content=content)) else: # user 或其他角色按 user 处理 converted.append(HumanMessage(content=content)) return converted def _assistant_to_openai(message): """将 langchain AIMessage 转为 OpenAI 格式 dict，含 tool_calls。""" tool_calls = [] for call in message.tool_calls or []: # 遍历模型产出的原生工具调用 tool_calls.append({ "id": call.get("id") or "tool-call", # 调用 ID，缺失时给默认值 "type": "function", "function": { "name": call.get("name"), "arguments": json.dumps(call.get("args") or {}, ensure_ascii=False), # args 序列化为 JSON 字符串 }, }) result = {"role": "assistant", "content": str(message.content or "")} if tool_calls: result["tool_calls"] = tool_calls # 有工具调用时挂到 tool_calls 字段 return result class MyRolloutProcessor(AbstractRolloutProcessor): """Agentic OPD Rollout 处理器。 传入 tools 参数让学生模型产生原生 tool_call，chat-completions parser 生效，content 干净。""" def setup(self) -> None: pass # 初始化钩子，无资源需预热时留空 @observe_processor # 装饰后 process 自动接入 Tracing async def process(self, input: RolloutInput) -> RolloutOutput: started = time.perf_counter() # 计时起点，用于上报 latency try: # 1. 取模型资源：api_key/base_url/model_name 来自 RolloutInput.model_resource resource = input.model_resource api_key = ( # Secret 类型取真实值，普通字符串直用 resource.api_key.get_secret_value() if hasattr(resource.api_key, "get_secret_value") else resource.api_key ) sampling = input.sampling_params or {} # 采样参数字典，可能为空 # 2. 构造 LLM 客户端：ChatOpenAI 封装 OpenAI 兼容接口 llm = ChatOpenAI( model=resource.model_name, openai_api_key=api_key, openai_api_base=resource.base_url, temperature=float(sampling.get("temperature", 0.2)), max_tokens=int(sampling.get("max_tokens", 512)), request_timeout=float(sampling.get("timeout", 60.0)), extra_body={"enable_thinking": False}, # 关闭思考链输出，content 干净 streaming=False, # 非流式，便于一次性取 tool_calls ) # Rollout 自构造 LLM，trace_client 须显式调用，否则请求不产生 LLM Span trace_client(llm) # 3. 采样：传入 tools 参数注入工具定义，tool_choice=auto 由模型自行决定是否调用 model = llm.bind_tools(TOOL_DEFINITIONS, tool_choice="auto") response = await model.ainvoke(_to_langchain_messages(input.messages)) latency = round(time.perf_counter() - started, 4) # 本轮采样耗时（秒） # 4. 构造输出：原生 tool_calls 与 rollout_metrics 一并放入 AgentOutput return RolloutOutput( agent_output=AgentOutput( messages=[*input.messages, _assistant_to_openai(response)], # 追加学生回复 rollout_extra=input.rollout_extra, # 透传附加信息供 Reward 读取 rollout_metrics={ "latency": latency, # 采样延迟 "tool_call_count": float(len(response.tool_calls or [])), # 本轮工具调用次数 }, reward_score=0.0, # 占位，由 Reward 覆写 ), status=TaskStatus.SUCCESS, error=None, ) except Exception as exc: # 5. 异常兜底：返回 ROLLOUT_ERROR 文本 + FAILED，保轨迹不中断，Reward 据前缀跳过评分 latency = round(time.perf_counter() - started, 4) return RolloutOutput( agent_output=AgentOutput( messages=[ *input.messages, {"role": "assistant", "content": "ROLLOUT_ERROR: rollout failed"}, ], rollout_extra=input.rollout_extra, rollout_metrics={"latency": latency, "tool_call_count": 0.0}, reward_score=0.0, ), status=TaskStatus.FAILED, error=type(exc).__name__, # 上报异常类型名，便于排查 ) ``` RolloutInput and AgentOutput field table:

Field

Source

Meaning

model\_resource

RolloutInput

Contains api\_key/base\_url/model\_name; used to construct ChatOpenAI

sampling\_params

RolloutInput

Sampling parameters such as temperature/max\_tokens/timeout

messages

RolloutInput

Input conversation list, containing system/user/assistant

rollout\_extra

RolloutInput

Passed through to AgentOutput for Reward to read

messages

AgentOutput

Output conversation; the last assistant turn contains tool\_calls

rollout\_metrics

AgentOutput

Reports latency (seconds) and tool\_call\_count (float)

reward\_score

AgentOutput

Placeholder 0.0, overwritten by Reward

status

RolloutOutput

SUCCESS or FAILED

error

RolloutOutput

Exception type name; None when normal

### Tracing Integration Rollout constructs its own LLM client and, unlike Reward (where `@observe_processor` implicitly injects it), must explicitly call `trace_client` to wrap the client. Integration steps: 1. Take `model_resource` (containing api\_key/base\_url/model\_name) and `sampling_params` from `RolloutInput`, and construct the LLM client (such as `ChatOpenAI`) 2. Call `trace_client(llm)` to wrap it—after that, all LLM requests from this client automatically produce an LLM Span (recording model name, request content, Token usage, latency) 3. After wrapping, pass the `tools` parameter and `ainvoke` to sample Minimal snippet (full code in [Rollout Function Structure](#rollout-structure)): ```plaintext from langchain_openai import ChatOpenAI from dashscope.finetune.reinforcement.component.observability import trace_client resource = input.model_resource # 从 RolloutInput 取模型凭证与地址 llm = ChatOpenAI(model=resource.model_name, openai_api_key=resource.api_key, openai_api_base=resource.base_url, temperature=0.2, max_tokens=512) trace_client(llm) # 包装后该客户端所有 LLM 请求自动产生 LLM Span（模型名/Token/延迟） model = llm.bind_tools(TOOL_DEFINITIONS, tool_choice="auto") # 传入 tools 参数注入工具定义 response = await model.ainvoke(messages) # 异步采样，返回含 tool_calls 的 AIMessage ``` The example uses LangChain `ChatOpenAI` (recommended; wraps the OpenAI-compatible interface); you can also use the `openai` SDK directly—just replace `ChatOpenAI` and the message classes. If `trace_client` is not called, LLM requests produce no Span and the trajectory page will not show the tool-call chain (training is unaffected). For full Tracing configuration (decorators, Span types, metric paths), see [Observable Configuration](opd-observable-config.md). ## Development Reference Without Reward When Reward is not enabled, the functions list only contains the Rollout component, and teacher-model distillation is the only supervision signal. Submit with the `AgenticRL` client; the functions list only contains `RolloutFunctionComponent`, without passing `RewardFunctionComponent`. Construct `function_components` to specify the working directory and classpath, and configure `teacher_model` (teacher model), the student model, dataset paths, and resource specifications. The `teacher_model` field triggers distillation; for hyperparameter configuration, see [Training Configuration](opd-training-config.md). ```plaintext # 通用化 Agentic OPD 提交脚本（不启用 Reward） import asyncio from pathlib import Path from dashscope.finetune.agentic_rl import AgenticRL # Agentic RL/OPD 提交客户端 from dashscope.finetune.reinforcement import ( DataSourceType, # 数据源类型枚举（FILE_ID 表示按文件 ID 传数据） FunctionComponentModel, # 组件模型：指定打包目录与 classpath FunctionComponentRuntime, # 组件运行时：cpu/内存/并发等规格 RolloutFunctionComponent, # Rollout 组件，functions 列表元素之一 TrainingDataset, ValidationDataset, ) # 工作目录：须含 functions/ 子包 + requirements.txt，提交时整体打包上传 WORKSPACE = Path(__file__).resolve().parent def build_runtime(): """函数运行时资源规格。""" return { "cpu": 2, # vCPU 数 "memory_size": 4096, # 内存（MB） "disk_size": 512, # 磁盘（MB） "concurrency": 20, # 单实例并发请求数 "capacity": 10, # 初始实例数 "min_capacity": 10, # 最小实例数（缩容下限） "max_capacity": 30, # 最大实例数（扩容上限） "memory_scale_threshold": 0.6, # 内存利用率达 60% 触发扩容 "concurrency_scale_threshold": 0.6, # 并发利用率达 60% 触发扩容 "env": {}, # 注入函数运行环境的环境变量 } async def submit(): client = AgenticRL() # 不启用 Reward：functions 只含 RolloutFunctionComponent function_components = [ RolloutFunctionComponent( name="my-rollout", # 组件名，需在 functions 列表内唯一 timeout=120, # 单次采样超时（秒），超时计 FAILED fcmodel=FunctionComponentModel( zipdir=str(WORKSPACE.resolve()), # 打包根目录（含 functions/） classpath="functions.rollout.my_rollout.MyRolloutProcessor", # 处理器全限定类名 ), runtime=FunctionComponentRuntime(**build_runtime()), ), ] result = await client.run( job_name="my-agentic-opd", model="", # 学生模型 ID，被蒸馏对象 teacher_model="", # 教师模型 ID，触发蒸馏 training_datasets=[ TrainingDataset( data_source_type=DataSourceType.FILE_ID, file_name=str(WORKSPACE / "data" / "train.jsonl"), # 训练集 JSONL 路径 ) ], validation_datasets=[ ValidationDataset( data_source_type=DataSourceType.FILE_ID, file_name=str(WORKSPACE / "data" / "validation.jsonl"), # 验证集 JSONL 路径 ) ], functions=function_components, hyper_parameters={ # 超参与 opd_teacher_* 部署参数见训练配置 }, resources={ "charge_type": "mtu_postpaid", # 计费方式：MTU 后付费 "mtu_spec_code": "MTU4", # MTU 规格编码 "mtu_capacity": 24, # MTU 容量 }, ) if result.status_code != 200: # 非 200 视为提交失败，抛错便于上层捕获 raise RuntimeError(f"submission failed: status={result.status_code}") return result.output.job_id # 返回训练任务 ID if __name__ == "__main__": print(f"job_id={asyncio.run(submit())}") # 同步等待异步提交完成并打印 job_id ``` ## Development With Reward Enabled Reward is an optional component; once enabled, `RewardFunctionComponent` is appended to the functions list. Agentic Reward is simpler than Model OPD: via the chat-completions interface, the parser takes effect, and the output content is clean with no need to strip thinking text. ### Reward Function Structure Inherit `AbstractRewardProcessor` and implement `async process(input: RewardInput) -> RewardOutput`; decorate `process` with `@observe_processor` (automatically integrates Tracing). The process does five things: 1. **Skip failed samples**: when Rollout fails (the last-turn content starts with `ROLLOUT_ERROR:`), return 0 points directly without scoring 2. **Retrieve the standard answer**: `ground_truth` (the scoring baseline; taken from `RewardInput` first, or read from the data field `rollout_extra.solution` when missing) 3. **Retrieve the student output**: take `tool_calls` from the last assistant message (the tool calls produced by the student model) 4. **Score**: compare the standard answer with the student output to compute `reward_score` (main score) and `reward_metrics` (per-dimension scores) 5. **Construct the return**: place `reward_score` separately in `Reward.reward_score`, put the remaining dimensions into `reward_metrics`, and wrap as `RewardOutput` `agent_output` (the student output) is produced by Rollout; Reward only reads it, never writes. For full Tracing configuration, see [Observable Configuration](opd-observable-config.md). ```plaintext # 通用化 Agentic OPD Reward 处理器模板 import json from typing import Any, Iterable, Mapping from dashscope.finetune.reinforcement import ( AbstractRewardProcessor, # 抽象基类，子类实现 async process Reward, # 评分载体：reward_score 主分 + reward_metrics 各维度 RewardInput, # 输入：含 ground_truth 与 agent_output RewardOutput, # 输出：Reward + status + error TaskStatus, ) from dashscope.finetune.reinforcement.component.observability import observe_processor from functions.tool_spec import DESTRUCTIVE_TOOLS # 破坏性工具名集合，评分可据此判误调 def _parse_args(value): """将 arguments 解析为 dict，支持 dict/JSON 字符串/None。""" if isinstance(value, Mapping): return dict(value) # 已是 dict 直接转 if not value: return {} # None/空串返回空 dict parsed = json.loads(str(value)) # JSON 字符串解析 return dict(parsed) if isinstance(parsed, Mapping) else {} def _normalized_calls(calls): """将 tool_calls 规范化为可比较的排序列表。 argument 整体序列化相等：json.dumps + sort_keys 后字符串比较。""" normalized = [] for call in calls: normalized.append(( str(call.get("name", "")), # 工具名 json.dumps( # 参数整体序列化为紧凑字符串，键排序后比较 dict(call.get("arguments") or {}), ensure_ascii=False, sort_keys=True, # 键排序消除顺序差异 separators=(",", ":"), # 紧凑输出，无多余空白 ), )) return sorted(normalized) # 排序后列表比较，与调用顺序无关 def your_scoring_function(messages, ground_truth): """通用评分函数。Agentic 经由 chat-completions，parser 生效，content 干净无需剥离思考文本。 以下给最小可计算实现（decision + tool_name + argument 三项全对得 1.0），维度与权重按业务扩展。 返回 dict 含 reward_score（主分）+ 若干自定义维度（float）。""" # ground_truth 可能是 dict 或 JSON 字符串，统一成 dict expected = ( ground_truth if isinstance(ground_truth, Mapping) else json.loads(str(ground_truth)) ) # 取末轮 assistant 消息（学生模型的最终输出） assistant = next( ( m for m in reversed(messages) if str(m.get("role", "")).lower() == "assistant" ), messages[-1] if messages else {}, # 无 assistant 时退化取末轮 ) # 从 tool_calls 提取预测调用（chat-completions parser 生效，结构化提取） predicted_calls = [] for raw in assistant.get("tool_calls") or []: func = raw.get("function", raw) # 兼容扁平结构与 OpenAI 嵌套 function 结构 predicted_calls.append({ "name": str(func.get("name", "")), "arguments": _parse_args(func.get("arguments")), }) # 最小可计算实现：decision + tool_name + argument 三项全对得 1.0 exp_decision = str(expected.get("decision", "")) # 标准决策：tool_call / clarify exp_calls = expected.get("tool_calls") or [] # 标准工具调用列表 # decision_ok：标准要求 tool_call 且学生确有产出调用 decision_ok = exp_decision == "tool_call" and len(predicted_calls) > 0 # name_ok：逐个比对工具名是否一致 name_ok = all( pc["name"] == str(ec.get("name", "")) for pc, ec in zip(predicted_calls, exp_calls) ) if (predicted_calls and exp_calls) else False # arg_ok：name 全对后比对参数整体序列化相等 arg_ok = all( _normalized_calls([pc]) == _normalized_calls([{ "name": str(ec.get("name", "")), "arguments": ec.get("arguments", {}), }]) for pc, ec in zip(predicted_calls, exp_calls) ) if name_ok else False # 主分：三项全对得 1.0，否则 0.0（按业务改加权/连续值） reward_score = 1.0 if (decision_ok and name_ok and arg_ok) else 0.0 return { "reward_score": float(reward_score), "decision_accuracy": float(decision_ok), # 决策是否正确（是否该调工具） "tool_name_accuracy": float(name_ok), # 工具名是否选对 "argument_accuracy": float(arg_ok), # 参数是否全对 } class MyRewardProcessor(AbstractRewardProcessor): """Agentic OPD Reward 处理器。 chat-completions parser 生效，content 干净无需剥离思考文本。""" @observe_processor # 装饰后 process 自动接入 Tracing async def process(self, input: RewardInput) -> RewardOutput: # 0. Rollout 失败样本跳过评分：检查末轮 content 是否 ROLLOUT_ERROR 兜底前缀 last_msg = (input.agent_output.messages or [{}])[-1] last_content = str(last_msg.get("content", "")) if last_content.startswith("ROLLOUT_ERROR:"): # Rollout 失败直接返回 0 分，状态 FAILED，不进入评分逻辑 return RewardOutput( reward=Reward(reward_score=0.0, reward_metrics={}), status=TaskStatus.FAILED, error="rollout failed", ) # 1. 取 ground_truth：优先 RewardInput，缺失时从 rollout_extra.solution 读 ground_truth = input.ground_truth if ground_truth is None and input.agent_output.rollout_extra: ground_truth = input.agent_output.rollout_extra.get("solution") # 2. 评分：调用评分函数得到主分 + 各维度 dict metrics = your_scoring_function(input.agent_output.messages, ground_truth) # 3. reward_score 单独经由 Reward.reward_score 上报，其余进 reward_metrics（不重复） reward_score = metrics.pop("reward_score") return RewardOutput( reward=Reward(reward_score=reward_score, reward_metrics=metrics), status=TaskStatus.SUCCESS, error=None, ) ``` RewardInput and RewardOutput field table:

Field

Source

Meaning

ground\_truth

RewardInput

Scoring baseline; can be None

agent\_output

RewardInput

Produced by Rollout; Reward only reads

messages

agent\_output.messages

Conversation trajectory containing tool\_calls

rollout\_extra

agent\_output.rollout\_extra

Passed-through additional information; solution can be read

reward\_score

Reward

Main score, reported separately, not in metrics

reward\_metrics

Reward

Other dimensions dict, each value a float

status

RewardOutput

SUCCESS or FAILED

error

RewardOutput

Exception info; None when normal

### Scoring Logic `reward_score` is the main reward scalar, and `reward_metrics` is a dictionary of additional dimensions (each value a float). The scoring function implementation is in the [Reward Function Structure](#reward-structure) section above. Key conventions: - **Main score reported separately**: `reward_score` is assigned separately to `Reward.reward_score` and does not duplicate into `reward_metrics` (`metrics.pop("reward_score")`) - **Dimensions are custom**: dimensions and weights are defined by the task; implement them per your business needs - **Parameter comparison**: argument comparison uses whole-serialization equality: compare strings after `json.dumps(args, sort_keys=True, separators=(",", ":"))` - **Focus on business constraints**: teacher-model distillation already provides dense distribution-level supervision, so Reward need not cover everything—focus on the business constraints that the teacher model cannot cover (such as whether `safe_action` mis-invoked a destructive tool, or whether `format_valid` is compliant) ### Submit a Training Task With Reward Enabled After writing `MyRewardProcessor`, register both `RolloutFunctionComponent` and `RewardFunctionComponent` in the `functions` list. `run()` completes registration → upload data → submit the task in one step. ```plaintext from dashscope.finetune.agentic_rl import AgenticRL from dashscope.finetune.reinforcement import ( DataSourceType, FunctionComponentModel, FunctionComponentRuntime, RolloutFunctionComponent, RewardFunctionComponent, # Reward 组件，启用 Reward 时追加 TrainingDataset, ValidationDataset, ) client = AgenticRL() result = await client.run( job_name="my-agentic-opd-reward", model="", # 学生模型 ID teacher_model="", # 教师模型 ID，触发蒸馏 training_datasets=[TrainingDataset( data_source_type=DataSourceType.FILE_ID, file_name="data/train.jsonl")], # 训练集 JSONL validation_datasets=[ValidationDataset( data_source_type=DataSourceType.FILE_ID, file_name="data/validation.jsonl")], # 验证集 JSONL functions=[ # Rollout 组件：与无 Reward 模式一致 RolloutFunctionComponent( name="my-rollout", timeout=120, # 采样超时 120s fcmodel=FunctionComponentModel( zipdir=".", # 工作区根目录（含 functions/） classpath="functions.rollout.my_rollout.MyRolloutProcessor"), # Rollout 处理器类 runtime=FunctionComponentRuntime( cpu=2, memory_size=4096, disk_size=512, concurrency=20, capacity=10, min_capacity=10, max_capacity=30)), # Reward 组件：weight 为奖励权重，多个 Reward 时按权重加权 RewardFunctionComponent( name="my-reward", weight=1.0, timeout=120, # 权重 1.0，评分超时 120s fcmodel=FunctionComponentModel( zipdir=".", classpath="functions.reward.my_reward.MyRewardProcessor"), # Reward 处理器类 runtime=FunctionComponentRuntime( cpu=2, memory_size=4096, disk_size=512, concurrency=20, capacity=10, min_capacity=10, max_capacity=30)), ], hyper_parameters={ # 超参与 opd_teacher_* 部署参数见训练配置 }, ) ``` `RolloutFunctionComponent` is consistent with the no-Reward mode, and `RewardFunctionComponent` points to the module where `MyRewardProcessor` resides. For full field descriptions, see [Training Configuration](opd-training-config.md). ### Function Testing Before submitting for training, locally validate the scoring logic and Rollout output structure first, to avoid discovering errors only inside the training task (which wastes compute). Feed sample messages + ground\_truth into `your_scoring_function` and assert `reward_score` and each dimension value. For Rollout, construct a mock `RolloutInput` and call `process` locally to validate the output structure. Local unit tests only validate function logic, not remote training effects. Run tests with `pytest` (install first with `pip install pytest` if not installed): ```plaintext # 运行评分函数本地测试，-v 显示每个用例明细 python -m pytest test_scoring.py -v ``` To construct a mock `RolloutInput` and call the Rollout `process`, fill three fields: `model_resource` (api\_key/base\_url/model\_name), `sampling_params` (sampling parameters), and `messages` (conversation list). Example: ```plaintext from dashscope.finetune.reinforcement import RolloutInput from dashscope.finetune.reinforcement.component.data.base_data_model import ( ModelResource, # 模型凭证与地址封装 ) # 构造 mock RolloutInput：填充三个必填字段 + 透传的 rollout_extra mock_input = RolloutInput( model_resource=ModelResource( model_name="", # 学生模型 ID base_url="", # 推理服务地址 api_key="", # 推理服务 API Key ), sampling_params={"temperature": 0.2, "max_tokens": 512, "timeout": 60.0}, # 采样参数 messages=[{"role": "user", "content": "查询项目 P100 的状态"}], # 输入对话 rollout_extra={"sample_id": "test-001"}, # 附加信息，Reward 也可读 ) output = await MyRolloutProcessor().process(mock_input) # 本地调用 process，验证输出结构 ``` ```plaintext # 通用化评分函数本地测试骨架 import json import unittest # 按实际落盘路径 import（your_scoring_function 定义见上方 Reward 模板，可抽到 functions/reward/scoring.py 或与处理器同模块） from functions.reward.scoring import your_scoring_function # 路径按实际调整 class ScoringTest(unittest.TestCase): def test_correct_tool_call(self): """工具调用完全匹配时 reward_score 应为 1.0。""" # 标准答案：决策为 tool_call，调 query_status 且 item_id=P100 expected = { "decision": "tool_call", "tool_calls": [ {"name": "query_status", "arguments": {"item_id": "P100"}} ], } # 学生输出：与标准完全一致的 tool_calls（OpenAI 嵌套结构） messages = [ { "role": "assistant", "content": "", "tool_calls": [ { "type": "function", "function": { "name": "query_status", "arguments": '{"item_id": "P100"}', # JSON 字符串形式 }, } ], } ] score = your_scoring_function(messages, json.dumps(expected)) # expected 传 JSON 字符串也兼容 self.assertEqual(1.0, score["reward_score"]) # 全对应得满分 1.0 def test_wrong_argument(self): """参数错误时 reward_score 应低于 1.0。""" # 标准答案同上 expected = { "decision": "tool_call", "tool_calls": [ {"name": "query_status", "arguments": {"item_id": "P100"}} ], } # 学生输出：工具名对但 item_id 错（P999 ≠ P100） messages = [ { "role": "assistant", "content": "", "tool_calls": [ { "type": "function", "function": { "name": "query_status", "arguments": '{"item_id": "P999"}', # 参数值错误 }, } ], } ] score = your_scoring_function(messages, expected) # expected 传 dict 也兼容 self.assertLess(score["reward_score"], 1.0) # 参数错则不得满分 if __name__ == "__main__": unittest.main() ``` ## Next Steps - Training parameters and submission configuration → [Online Policy Distillation Training Configuration](opd-training-config.md) - Observable configuration → [Online Policy Distillation Observable Configuration](opd-observable-config.md) - Review the training overview → [Online Policy Distillation Training Overview](opd-training-overview.md) ## FAQ Troubleshooting for high-frequency Rollout and Reward issues encountered in Agentic OPD development. Check whether the correct tool definition list is passed (the `tools` parameter), whether `tool_choice` is auto, and whether api\_key and base\_url are valid. Check whether the tool definition list is complete, whether the system prompt clearly indicates when to call tools, and whether the student model supports function calling. Common cause: `teacher_model` is filled with a model ID not registered in Bailian (such as a HuggingFace repository name). Troubleshooting: use a Bailian-supported model ID; for the complete list and more causes, see [Training Configuration FAQ](opd-training-config.md#opd-faq). Check whether `reward_score` is popped out to be reported separately via `Reward.reward_score`, whether the remaining dimension values are all floats, and whether `ground_truth` is correctly read from `rollout_extra.solution`. Check whether each `rollout_extra.solution` in the data JSONL has a value. Reward takes from `input.ground_truth` first, and falls back to `input.agent_output.rollout_extra.solution` when missing; if both are empty, the score is 0. Confirm that the `functions/` directory contains `__init__.py` (to make the directory an importable Python package); the `classpath` aligns with the file path, e.g., `functions.rollout.my_rollout.MyRolloutProcessor` corresponds to the `MyRolloutProcessor` class inside `functions/rollout/my_rollout.py`.