217 lines
8.9 KiB
Python
217 lines
8.9 KiB
Python
# Copyright (c) Alibaba, Inc. and its affiliates.
|
|
"""T1 骨架 - TaskConfig / TaskState / ReviewResult / Adapter 集成验证.
|
|
|
|
Plan 覆盖点:
|
|
- ``TaskConfig.agent_config`` 开关 (None / dict / NativeAgentConfig)
|
|
- ``TaskState.agent_trace`` 替换旧 ``_trajectory``
|
|
- ``ReviewResult.agent_trace`` 持久化 + 旧 ``trajectory`` 向后兼容丢弃
|
|
- ``DefaultDataAdapter._on_inference`` 根据 ``agent_config`` 自动分支
|
|
- ``AgentLoopAdapter`` 忽略全局 ``agent_config`` 用自己的 ``strategy_name``
|
|
"""
|
|
|
|
import asyncio
|
|
import unittest
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import evalscope # noqa: F401 - trigger strategy registration
|
|
from evalscope.api.agent import AgentTrace, EventType, NativeAgentConfig
|
|
from evalscope.api.benchmark.adapters import AgentLoopAdapter
|
|
from evalscope.api.benchmark.adapters.default_data_adapter import DefaultDataAdapter
|
|
from evalscope.api.dataset import Sample
|
|
from evalscope.api.evaluator import TaskState
|
|
from evalscope.api.evaluator.cache import ReviewResult
|
|
from evalscope.api.messages import ChatMessageAssistant, ChatMessageUser
|
|
from evalscope.api.metric import SampleScore, Score
|
|
from evalscope.api.model.model_output import ChatCompletionChoice, ModelOutput
|
|
from evalscope.api.tool import ToolCall, ToolFunction
|
|
from evalscope.config import TaskConfig
|
|
|
|
|
|
def _mock_model_generate_final(content: str = 'ok'):
|
|
"""Return a MagicMock model whose .generate() yields an assistant with no tool_calls."""
|
|
model = MagicMock()
|
|
output = ModelOutput(
|
|
model='mock',
|
|
choices=[ChatCompletionChoice(message=ChatMessageAssistant(content=content))],
|
|
)
|
|
model.generate.return_value = output
|
|
# AgentLoop awaits ``generate_async``; mock both so sync/agent paths share
|
|
# the same return value.
|
|
model.generate_async = AsyncMock(return_value=output)
|
|
return model
|
|
|
|
|
|
def _mock_model_generate_submit(answer: str = 'ok'):
|
|
"""Return a MagicMock model whose .generate() yields a submit tool call."""
|
|
submit_call = ToolCall(id='sc1', function=ToolFunction(name='submit', arguments={'answer': answer}))
|
|
model = MagicMock()
|
|
output = ModelOutput(
|
|
model='mock',
|
|
choices=[ChatCompletionChoice(message=ChatMessageAssistant(content='', tool_calls=[submit_call]))],
|
|
)
|
|
model.generate.return_value = output
|
|
model.generate_async = AsyncMock(return_value=output)
|
|
return model
|
|
|
|
|
|
class TestTaskConfigNativeAgentConfig(unittest.TestCase):
|
|
"""TaskConfig 增加的 agent_config 字段与 validator."""
|
|
|
|
def test_default_is_none(self):
|
|
cfg = TaskConfig(model='dummy')
|
|
self.assertIsNone(cfg.agent_config)
|
|
|
|
def test_dict_auto_coerced_to_agent_config(self):
|
|
cfg = TaskConfig(
|
|
model='dummy',
|
|
agent_config={'strategy': 'function_calling', 'max_steps': 5, 'tools': []},
|
|
)
|
|
self.assertIsInstance(cfg.agent_config, NativeAgentConfig)
|
|
self.assertEqual(cfg.agent_config.max_steps, 5)
|
|
|
|
def test_accepts_agent_config_instance(self):
|
|
inst = NativeAgentConfig(strategy='function_calling', max_steps=3)
|
|
cfg = TaskConfig(model='dummy', agent_config=inst)
|
|
self.assertIs(cfg.agent_config, inst)
|
|
|
|
def test_invalid_type_raises(self):
|
|
with self.assertRaises(Exception):
|
|
TaskConfig(model='dummy', agent_config=123) # type: ignore[arg-type]
|
|
|
|
|
|
class TestTaskStateAgentTrace(unittest.TestCase):
|
|
"""TaskState 的 agent_trace property/setter (取代旧 _trajectory)."""
|
|
|
|
def _make_state(self):
|
|
sample = Sample(id=1, input='hi', target='ok')
|
|
return TaskState(
|
|
model='mock',
|
|
sample=sample,
|
|
messages=[ChatMessageUser(content='hi')],
|
|
output=ModelOutput(
|
|
model='mock',
|
|
choices=[ChatCompletionChoice(message=ChatMessageAssistant(content='ok'))],
|
|
),
|
|
)
|
|
|
|
def test_default_agent_trace_is_none(self):
|
|
state = self._make_state()
|
|
self.assertIsNone(state.agent_trace)
|
|
|
|
def test_assign_and_read_back(self):
|
|
state = self._make_state()
|
|
trace = AgentTrace(strategy='function_calling', max_steps=3)
|
|
trace.add_event(step=0, type=EventType.MODEL_GENERATE)
|
|
state.agent_trace = trace
|
|
self.assertIs(state.agent_trace, trace)
|
|
self.assertEqual(state.agent_trace.events[0].type, EventType.MODEL_GENERATE)
|
|
|
|
def test_legacy_trajectory_api_removed(self):
|
|
# plan §3.2: 旧的 trajectory / add_trajectory_step 必须已移除
|
|
state = self._make_state()
|
|
self.assertFalse(hasattr(state, 'trajectory'))
|
|
self.assertFalse(hasattr(state, 'add_trajectory_step'))
|
|
|
|
|
|
class TestReviewResultAgentTrace(unittest.TestCase):
|
|
"""ReviewResult 持久化复用: agent_trace 字段替换了 trajectory."""
|
|
|
|
def _sample_score(self):
|
|
return SampleScore(score=Score(value={'acc': 1.0}), sample_id='s1')
|
|
|
|
def test_default_agent_trace_none(self):
|
|
rr = ReviewResult(index=0, sample_score=self._sample_score())
|
|
self.assertIsNone(rr.agent_trace)
|
|
|
|
def test_agent_trace_roundtrip(self):
|
|
trace = AgentTrace(strategy='function_calling', max_steps=1)
|
|
trace.add_event(step=0, type=EventType.SUBMIT, payload={'final_answer': 'x'})
|
|
rr = ReviewResult(index=0, sample_score=self._sample_score(), agent_trace=trace)
|
|
restored = ReviewResult.model_validate_json(rr.model_dump_json())
|
|
self.assertIsNotNone(restored.agent_trace)
|
|
self.assertEqual(restored.agent_trace.events[0].type, EventType.SUBMIT)
|
|
|
|
def test_legacy_trajectory_field_silently_dropped(self):
|
|
# 老缓存带 trajectory 字段; _migrate_legacy_input 应丢弃
|
|
raw = {
|
|
'index': 0,
|
|
'sample_score': self._sample_score().model_dump(),
|
|
'trajectory': [{'role': 'user', 'content': 'legacy'}], # 旧字段
|
|
'messages': [],
|
|
}
|
|
rr = ReviewResult.model_validate(raw)
|
|
self.assertIsNone(rr.agent_trace)
|
|
# trajectory 字段不应复活 (禁止保留)
|
|
self.assertNotIn('trajectory', ReviewResult.model_fields)
|
|
|
|
|
|
class TestDefaultDataAdapterAgentBranch(unittest.TestCase):
|
|
"""DefaultDataAdapter._on_inference 根据 agent_config 自动分支."""
|
|
|
|
def _make_adapter(self, task_config):
|
|
adapter = DefaultDataAdapter.__new__(DefaultDataAdapter)
|
|
adapter._task_config = task_config
|
|
return adapter
|
|
|
|
def test_no_agent_config_uses_plain_generate(self):
|
|
from evalscope.api.evaluator import InferenceResult
|
|
from evalscope.api.model import ModelOutput
|
|
|
|
adapter = self._make_adapter(TaskConfig(model='dummy'))
|
|
model = _mock_model_generate_final('plain')
|
|
sample = Sample(input='hi')
|
|
|
|
out = adapter._on_inference(model, sample)
|
|
|
|
model.generate.assert_called_once()
|
|
# 普通分支返回纯 ModelOutput, 不应是 InferenceResult
|
|
self.assertIsInstance(out, ModelOutput)
|
|
self.assertNotIsInstance(out, InferenceResult)
|
|
self.assertEqual(out.choices[0].message.content, 'plain')
|
|
|
|
|
|
class TestAgentLoopAdapterOverrides(unittest.TestCase):
|
|
"""AgentLoopAdapter._on_inference 忽略全局 agent_config; 用 strategy_name 类属性."""
|
|
|
|
def test_ignores_global_agent_config(self):
|
|
# 即使全局配置切了 react (此刻未注册), AgentLoopAdapter 仍该用自己的 strategy_name=function_calling
|
|
cfg = TaskConfig(
|
|
model='dummy',
|
|
agent_config={'strategy': 'react', 'max_steps': 99}, # 全局
|
|
)
|
|
adapter = AgentLoopAdapter.__new__(AgentLoopAdapter)
|
|
adapter._task_config = cfg
|
|
# AgentLoopAdapter 类默认 strategy_name = 'function_calling' / max_steps_default = 30
|
|
adapter.max_steps = AgentLoopAdapter.max_steps_default
|
|
|
|
model = _mock_model_generate_final('native_agent')
|
|
sample = Sample(input='hi')
|
|
out = adapter._on_inference(model, sample)
|
|
|
|
self.assertEqual(out.output.choices[0].message.content, 'native_agent')
|
|
trace = out.trace
|
|
# 用了 AgentLoopAdapter 自己的 strategy_name, 不是全局 'react'
|
|
self.assertEqual(trace.strategy, 'function_calling')
|
|
self.assertEqual(trace.max_steps, AgentLoopAdapter.max_steps_default)
|
|
|
|
def test_build_initial_messages_handles_str_and_list(self):
|
|
adapter = AgentLoopAdapter.__new__(AgentLoopAdapter)
|
|
s_str = Sample(input='hello')
|
|
msgs = adapter.build_initial_messages(s_str)
|
|
self.assertEqual(len(msgs), 1)
|
|
self.assertEqual(msgs[0].role, 'user')
|
|
self.assertEqual(msgs[0].content, 'hello')
|
|
|
|
s_list = Sample(input=[ChatMessageUser(content='a'), ChatMessageUser(content='b')])
|
|
msgs = adapter.build_initial_messages(s_list)
|
|
self.assertEqual(len(msgs), 2)
|
|
|
|
def test_default_build_tools_and_environment(self):
|
|
adapter = AgentLoopAdapter.__new__(AgentLoopAdapter)
|
|
self.assertEqual(adapter.build_tools(Sample(input='x')), {})
|
|
self.assertIsNone(adapter.build_environment(Sample(input='x')))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|