| def __init__(self, **kwargs): |
| self.request_id = kwargs.get("request_id", "req") |
| self.prompt = kwargs.get("prompt") |
| self.prompt_token_ids = kwargs.get("prompt_token_ids") |
| self.messages = kwargs.get("messages") |
| self.eos_token_ids = kwargs.get("eos_token_ids") |
| self.chat_template = kwargs.get("chat_template") |
| self.enable_thinking = kwargs.get("enable_thinking") |
| self.history = kwargs.get("history") |
| self.tools = kwargs.get("tools") |
| self.system = kwargs.get("system") |
| self.sampling_params = SimpleNamespace( |
| top_p=kwargs.get("top_p"), |
| temperature=kwargs.get("temperature"), |
| repetition_penalty=kwargs.get("repetition_penalty"), |
| frequency_penalty=kwargs.get("frequency_penalty"), |
| presence_penalty=kwargs.get("presence_penalty"), |
| stop=kwargs.get("stop"), |
| stop_token_ids=kwargs.get("stop_token_ids"), |
| stop_seqs_len=kwargs.get("stop_seqs_len"), |
| bad_words=kwargs.get("bad_words"), |
| bad_words_token_ids=kwargs.get("bad_words_token_ids"), |
| max_tokens=kwargs.get("max_tokens"), |
| ) |
| chat_template_kwargs = {"chat_template": "Hello!"} |
| result = self.processor.process_request(request, 100, chat_template_kwargs=chat_template_kwargs) |
| self.assertEqual(result.prompt_token_ids, [1]) |
|
| def test_process_request_dict(self): |
| request_dict = { |
| "messages": [{"role": "user", "content": "Hello!"}], |
| "chat_template_kwargs": {"chat_template": "Hello!"}, |
| "eos_token_ids": [1], |
| "temperature": 1, |
| "top_p": 1, |
|
| def get(self, key, default=None): |
| if hasattr(self, key) and getattr(self, key) is not None: |
| return getattr(self, key) |
| return getattr(self.sampling_params, key, default) |
|
| def set(self, key, value): |
| if hasattr(self.sampling_params, key): |
| setattr(self.sampling_params, key, value) |
| else: |
| setattr(self, key, value) |
|
| def to_dict(self): |
| return { |
| "request_id": self.request_id, |
| "messages": self.messages, |
| "prompt": self.prompt, |
| "system": self.system, |
| "history": self.history, |
| "tools": self.tools, |
| "chat_template": self.chat_template, |
| "enable_thinking": self.enable_thinking, |
| } |
| result = self.processor.process_request_dict(request_dict, 100) |
| self.assertEqual(result["prompt_token_ids"], [1]) |
|
| def test_process_response_dict_normal(self): |
| self.processor.tokenizer.decode_token = MagicMock(return_value=("Mock decoded text", 0, 0)) |
| self.processor.reasoning_parser.extract_reasoning_content = MagicMock( |
| return_value=("Mock reasoning content", "Mock final text") |
| def __getitem__(self, key): |
| return self.get(key) |
|
| def __setitem__(self, key, value): |
| self.set(key, value) |
|