Agentic OPD Development

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 时）

# 通用化工具定义模板
# 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: <question>.
If no tool is appropriate, reply exactly with NO_TOOL: <answer>.
Destructive actions require explicit user confirmation first; otherwise reply with CONFIRM: <question>."""


# 工具定义列表，格式与 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",
    # 按任务添加更多高危工具名
}

# 通用化 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__,  # 上报异常类型名，便于排查
            )

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

# 通用化 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="<student-model-id>",          # 学生模型 ID，被蒸馏对象
        teacher_model="<teacher-model-id>",  # 教师模型 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

# 通用化 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,
        )

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="<student-model-id>",          # 学生模型 ID
    teacher_model="<teacher-model-id>",  # 教师模型 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_* 部署参数见训练配置
    },
)

# 运行评分函数本地测试，-v 显示每个用例明细
python -m pytest test_scoring.py -v

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="<student-model-id>",       # 学生模型 ID
        base_url="<inference-endpoint>",        # 推理服务地址
        api_key="<your-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，验证输出结构

# 通用化评分函数本地测试骨架
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()