prompt: You will be provided with a partial code base and an issue statement explaining a problem to resolve.
  <issue>
  Add typical sampling
  HF has TypicalWarper and `typical_p`. Would be nice to be able to pass that.
  
  </issue>
  <code>
  [start of server/text_generation_server/utils/tokens.py]
  1 import re
  2 import torch
  3 
  4 from transformers import (
  5     LogitsProcessorList,
  6     TemperatureLogitsWarper,
  7     TopKLogitsWarper,
  8     TopPLogitsWarper,
  9     RepetitionPenaltyLogitsProcessor,
  10     PreTrainedTokenizerBase,
  11 )
  12 from typing import List, Tuple, Optional
  13 
  14 from text_generation_server.pb import generate_pb2
  15 from text_generation_server.pb.generate_pb2 import FinishReason
  16 from text_generation_server.utils.watermark import WatermarkLogitsProcessor
  17 
  18 
  19 class Sampling:
  20     def __init__(self, seed: int, device: str = "cpu"):
  21         self.generator = torch.Generator(device)
  22         self.generator.manual_seed(seed)
  23         self.seed = seed
  24 
  25     def __call__(self, logits):
  26         probs = torch.nn.functional.softmax(logits)
  27         next_tokens = torch.multinomial(probs, num_samples=1, generator=self.generator)
  28         return next_tokens
  29 
  30 
  31 class Greedy:
  32     def __call__(self, logits):
  33         return logits.argmax()
  34 
  35 
  36 class NextTokenChooser:
  37     def __init__(
  38         self,
  39         watermark=False,
  40         temperature=1.0,
  41         repetition_penalty=1.0,
  42         top_k=None,
  43         top_p=None,
  44         do_sample=False,
  45         seed=0,
  46         device="cpu",
  47     ):
  48         warpers = LogitsProcessorList()
  49         # the following idea is largely copied from this PR: https://github.com/huggingface/transformers/pull/5420/files
  50         # all samplers can be found in `generation_utils_samplers.py`
  51         sampling = do_sample
  52 
  53         if watermark:
  54             warpers.append(WatermarkLogitsProcessor(device=device))
  55         if repetition_penalty is not None and repetition_penalty != 1.0:
  56             warpers.append(RepetitionPenaltyLogitsProcessor(penalty=repetition_penalty))
  57         if temperature is not None and temperature != 1.0:
  58             temperature = float(temperature)
  59             warpers.append(TemperatureLogitsWarper(temperature))
  60             sampling = True
  61         if top_k is not None and top_k != 0:
  62             warpers.append(TopKLogitsWarper(top_k=top_k))
  63             sampling = True
  64         if top_p is not None and top_p < 1.0:
  65             warpers.append(TopPLogitsWarper(top_p=top_p))
  66             sampling = True
  67 
  68         self.warpers = warpers
  69         self.choice = Sampling(seed, device) if sampling else Greedy()
  70 
  71     def __call__(self, input_ids, scores):
  72         # Warp logits
  73         scores = self.warpers(input_ids, scores)
  74 
  75         # Compute logprobs
  76         logprobs = torch.log_softmax(scores, -1)
  77 
  78         # Choose tokens
  79         next_id = self.choice(scores[-1])
  80 
  81         return next_id.view(1, 1), logprobs
  82 
  83     @classmethod
  84     def from_pb(
  85         cls,
  86         pb: generate_pb2.NextTokenChooserParameters,
  87         device: torch.device,
  88     ) -> "NextTokenChooser":
  89         return NextTokenChooser(
  90             watermark=pb.watermark,
  91             temperature=pb.temperature,
  92             repetition_penalty=pb.repetition_penalty,
  93             top_k=pb.top_k,
  94             top_p=pb.top_p,
  95             do_sample=pb.do_sample,
  96             seed=pb.seed,
  97             device=device,
  98         )
  99 
  100 
  101 class StopSequenceCriteria:
  102     def __init__(self, stop_sequence: str):
  103         self.regex = re.compile(f".*{stop_sequence}$")
  104 
  105     def __call__(self, output: str) -> bool:
  106         if self.regex.findall(output):
  107             return True
  108         return False
  109 
  110 
  111 class StoppingCriteria:
  112     def __init__(
  113         self,
  114         eos_token_id: int,
  115         stop_sequence_criterias: List[StopSequenceCriteria],
  116         max_new_tokens=20,
  117     ):
  118         self.eos_token_id = eos_token_id
  119         self.stop_sequence_criterias = stop_sequence_criterias
  120         self.max_new_tokens = max_new_tokens
  121         self.current_tokens = 0
  122         self.current_output = ""
  123 
  124     def __call__(self, last_token: int, last_output: str) -> Tuple[bool, Optional[str]]:
  125         self.current_tokens += 1
  126         if self.current_tokens >= self.max_new_tokens:
  127             return True, FinishReason.FINISH_REASON_LENGTH
  128 
  129         if last_token == self.eos_token_id:
  130             return True, FinishReason.FINISH_REASON_EOS_TOKEN
  131 
  132         self.current_output += last_output
  133         for stop_sequence_criteria in self.stop_sequence_criterias:
  134             if stop_sequence_criteria(self.current_output):
  135                 return True, FinishReason.FINISH_REASON_STOP_SEQUENCE
  136 
  137         return False, None
  138 
  139     @classmethod
  140     def from_pb(
  141         cls,
  142         pb: generate_pb2.StoppingCriteriaParameters,
  143         tokenizer: PreTrainedTokenizerBase,
  144     ) -> "StoppingCriteria":
  145         stop_sequence_criterias = [
  146             StopSequenceCriteria(sequence) for sequence in pb.stop_sequences
  147         ]
  148         return StoppingCriteria(
  149             tokenizer.eos_token_id, stop_sequence_criterias, pb.max_new_tokens
  150         )
  151
  [end of server/text_generation_server/utils/tokens.py]
  </code>
  
  I need you to solve this issue by generating a single patch file that I can apply directly to this repository using git apply. Please respond with a single patch file in the following format.
  <patch>
  diff --git a/file.py b/file.py
  --- a/file.py
  +++ b/file.py
  @@ -1,27 +1,35 @@
   def euclidean(a, b):
  -    while b:
  -        a, b = b, a % b
  -    return a
  +    if b == 0:
  +        return a
  +    return euclidean(b, a % b)
   
   
   def bresenham(x0, y0, x1, y1):
       points = []
       dx = abs(x1 - x0)
       dy = abs(y1 - y0)
  -    sx = 1 if x0 < x1 else -1
  -    sy = 1 if y0 < y1 else -1
  -    err = dx - dy
  +    x, y = x0, y0
  +    sx = -1 if x0 > x1 else 1
  +    sy = -1 if y0 > y1 else 1
   
  -    while True:
  -        points.append((x0, y0))
  -        if x0 == x1 and y0 == y1:
  -            break
  -        e2 = 2 * err
  -        if e2 > -dy:
  -            err -= dy
  -            x0 += sx
  -        if e2 < dx:
  -            err += dx
  -            y0 += sy
  +    if dx > dy:
  +        err = dx / 2.0
  +        while x != x1:
  +            points.append((x, y))
  +            err -= dy
  +            if err < 0:
  +                y += sy
  +                err += dx
  +            x += sx
  +    else:
  +        err = dy / 2.0
  +        while y != y1:
  +            points.append((x, y))
  +            err -= dx
  +            if err < 0:
  +                x += sx
  +                err += dy
  +            y += sy
  +
  +    points.append((x, y))
       return points
  
  </patch>
golden_diff: diff --git a/server/text_generation_server/utils/tokens.py b/server/text_generation_server/utils/tokens.py
  --- a/server/text_generation_server/utils/tokens.py
  +++ b/server/text_generation_server/utils/tokens.py
  @@ -6,6 +6,7 @@
       TemperatureLogitsWarper,
       TopKLogitsWarper,
       TopPLogitsWarper,
  +    TypicalLogitsWarper,
       RepetitionPenaltyLogitsProcessor,
       PreTrainedTokenizerBase,
   )
  @@ -41,6 +42,7 @@
           repetition_penalty=1.0,
           top_k=None,
           top_p=None,
  +        typical_p=None,
           do_sample=False,
           seed=0,
           device="cpu",
  @@ -64,6 +66,9 @@
           if top_p is not None and top_p < 1.0:
               warpers.append(TopPLogitsWarper(top_p=top_p))
               sampling = True
  +        if typical_p is not None and typical_p < 1.0:
  +            warpers.append(TypicalLogitsWarper(mass=typical_p))
  +            sampling = True
   
           self.warpers = warpers
           self.choice = Sampling(seed, device) if sampling else Greedy()
  @@ -92,6 +97,7 @@
               repetition_penalty=pb.repetition_penalty,
               top_k=pb.top_k,
               top_p=pb.top_p,
  +            typical_p=pb.typical_p,
               do_sample=pb.do_sample,
               seed=pb.seed,
               device=device,
  
verification_info: {"golden_diff": "diff --git a/server/text_generation_server/utils/tokens.py b/server/text_generation_server/utils/tokens.py\n--- a/server/text_generation_server/utils/tokens.py\n+++ b/server/text_generation_server/utils/tokens.py\n@@ -6,6 +6,7 @@\n     TemperatureLogitsWarper,\n     TopKLogitsWarper,\n     TopPLogitsWarper,\n+    TypicalLogitsWarper,\n     RepetitionPenaltyLogitsProcessor,\n     PreTrainedTokenizerBase,\n )\n@@ -41,6 +42,7 @@\n         repetition_penalty=1.0,\n         top_k=None,\n         top_p=None,\n+        typical_p=None,\n         do_sample=False,\n         seed=0,\n         device=\"cpu\",\n@@ -64,6 +66,9 @@\n         if top_p is not None and top_p < 1.0:\n             warpers.append(TopPLogitsWarper(top_p=top_p))\n             sampling = True\n+        if typical_p is not None and typical_p < 1.0:\n+            warpers.append(TypicalLogitsWarper(mass=typical_p))\n+            sampling = True\n \n         self.warpers = warpers\n         self.choice = Sampling(seed, device) if sampling else Greedy()\n@@ -92,6 +97,7 @@\n             repetition_penalty=pb.repetition_penalty,\n             top_k=pb.top_k,\n             top_p=pb.top_p,\n+            typical_p=pb.typical_p,\n             do_sample=pb.do_sample,\n             seed=pb.seed,\n             device=device,\n", "issue": "Add typical sampling\nHF has TypicalWarper and `typical_p`. Would be nice to be able to pass that.\n", "before_files": [{"content": "import re\nimport torch\n\nfrom transformers import (\n    LogitsProcessorList,\n    TemperatureLogitsWarper,\n    TopKLogitsWarper,\n    TopPLogitsWarper,\n    RepetitionPenaltyLogitsProcessor,\n    PreTrainedTokenizerBase,\n)\nfrom typing import List, Tuple, Optional\n\nfrom text_generation_server.pb import generate_pb2\nfrom text_generation_server.pb.generate_pb2 import FinishReason\nfrom text_generation_server.utils.watermark import WatermarkLogitsProcessor\n\n\nclass Sampling:\n    def __init__(self, seed: int, device: str = \"cpu\"):\n        self.generator = torch.Generator(device)\n        self.generator.manual_seed(seed)\n        self.seed = seed\n\n    def __call__(self, logits):\n        probs = torch.nn.functional.softmax(logits)\n        next_tokens = torch.multinomial(probs, num_samples=1, generator=self.generator)\n        return next_tokens\n\n\nclass Greedy:\n    def __call__(self, logits):\n        return logits.argmax()\n\n\nclass NextTokenChooser:\n    def __init__(\n        self,\n        watermark=False,\n        temperature=1.0,\n        repetition_penalty=1.0,\n        top_k=None,\n        top_p=None,\n        do_sample=False,\n        seed=0,\n        device=\"cpu\",\n    ):\n        warpers = LogitsProcessorList()\n        # the following idea is largely copied from this PR: https://github.com/huggingface/transformers/pull/5420/files\n        # all samplers can be found in `generation_utils_samplers.py`\n        sampling = do_sample\n\n        if watermark:\n            warpers.append(WatermarkLogitsProcessor(device=device))\n        if repetition_penalty is not None and repetition_penalty != 1.0:\n            warpers.append(RepetitionPenaltyLogitsProcessor(penalty=repetition_penalty))\n        if temperature is not None and temperature != 1.0:\n            temperature = float(temperature)\n            warpers.append(TemperatureLogitsWarper(temperature))\n            sampling = True\n        if top_k is not None and top_k != 0:\n            warpers.append(TopKLogitsWarper(top_k=top_k))\n            sampling = True\n        if top_p is not None and top_p < 1.0:\n            warpers.append(TopPLogitsWarper(top_p=top_p))\n            sampling = True\n\n        self.warpers = warpers\n        self.choice = Sampling(seed, device) if sampling else Greedy()\n\n    def __call__(self, input_ids, scores):\n        # Warp logits\n        scores = self.warpers(input_ids, scores)\n\n        # Compute logprobs\n        logprobs = torch.log_softmax(scores, -1)\n\n        # Choose tokens\n        next_id = self.choice(scores[-1])\n\n        return next_id.view(1, 1), logprobs\n\n    @classmethod\n    def from_pb(\n        cls,\n        pb: generate_pb2.NextTokenChooserParameters,\n        device: torch.device,\n    ) -> \"NextTokenChooser\":\n        return NextTokenChooser(\n            watermark=pb.watermark,\n            temperature=pb.temperature,\n            repetition_penalty=pb.repetition_penalty,\n            top_k=pb.top_k,\n            top_p=pb.top_p,\n            do_sample=pb.do_sample,\n            seed=pb.seed,\n            device=device,\n        )\n\n\nclass StopSequenceCriteria:\n    def __init__(self, stop_sequence: str):\n        self.regex = re.compile(f\".*{stop_sequence}$\")\n\n    def __call__(self, output: str) -> bool:\n        if self.regex.findall(output):\n            return True\n        return False\n\n\nclass StoppingCriteria:\n    def __init__(\n        self,\n        eos_token_id: int,\n        stop_sequence_criterias: List[StopSequenceCriteria],\n        max_new_tokens=20,\n    ):\n        self.eos_token_id = eos_token_id\n        self.stop_sequence_criterias = stop_sequence_criterias\n        self.max_new_tokens = max_new_tokens\n        self.current_tokens = 0\n        self.current_output = \"\"\n\n    def __call__(self, last_token: int, last_output: str) -> Tuple[bool, Optional[str]]:\n        self.current_tokens += 1\n        if self.current_tokens >= self.max_new_tokens:\n            return True, FinishReason.FINISH_REASON_LENGTH\n\n        if last_token == self.eos_token_id:\n            return True, FinishReason.FINISH_REASON_EOS_TOKEN\n\n        self.current_output += last_output\n        for stop_sequence_criteria in self.stop_sequence_criterias:\n            if stop_sequence_criteria(self.current_output):\n                return True, FinishReason.FINISH_REASON_STOP_SEQUENCE\n\n        return False, None\n\n    @classmethod\n    def from_pb(\n        cls,\n        pb: generate_pb2.StoppingCriteriaParameters,\n        tokenizer: PreTrainedTokenizerBase,\n    ) -> \"StoppingCriteria\":\n        stop_sequence_criterias = [\n            StopSequenceCriteria(sequence) for sequence in pb.stop_sequences\n        ]\n        return StoppingCriteria(\n            tokenizer.eos_token_id, stop_sequence_criterias, pb.max_new_tokens\n        )\n", "path": "server/text_generation_server/utils/tokens.py"}]}