1import torch
2import json
3from transformers import AutoTokenizer, AutoModel
4
5# Constants
6MASK_TOKEN_ID = 126336
7
8def add_gumbel_noise(logits, temperature):
9 '''
10 The Gumbel max is a method for sampling categorical distributions.
11 For diffusion models, low-precision Gumbel Max affects generation quality.
12 '''
13 if temperature <= 0:
14 return logits
15
16 logits = logits.to(torch.float64)
17 noise = torch.rand_like(logits, dtype=torch.float64)
18 gumbel_noise = (- torch.log(noise)) ** temperature
19 return logits.exp() / gumbel_noise
20
21def get_num_transfer_tokens(mask_index, steps):
22 '''
23 In the reverse process, we precompute the number of tokens to transition at each step.
24 '''
25 mask_num = mask_index.sum(dim=1, keepdim=True)
26
27 # Ensure we have at least one step
28 if steps == 0:
29 steps = 1
30
31 base = mask_num // steps
32 remainder = mask_num % steps
33
34 num_transfer_tokens = torch.zeros(mask_num.size(0), steps, device=mask_index.device, dtype=torch.int64) + base
35
36 for i in range(mask_num.size(0)):
37 if remainder[i] > 0:
38 num_transfer_tokens[i, :remainder[i]] += 1
39
40 return num_transfer_tokens
41
42def generate(model, prompt, steps=128, gen_length=128, block_length=32, temperature=0.,
43 remasking='low_confidence', mask_id=MASK_TOKEN_ID):
44 '''
45 Generate text using LLaDA's diffusion-based generation process.
46 '''
47 device = next(model.parameters()).device
48 prompt = prompt.to(device)
49
50 x = torch.full((1, prompt.shape[1] + gen_length), mask_id, dtype=torch.long).to(device)
51 x[:, :prompt.shape[1]] = prompt.clone()
52
53 prompt_index = (x != mask_id)
54
55 assert gen_length % block_length == 0
56 num_blocks = gen_length // block_length
57
58 assert steps % num_blocks == 0
59 steps_per_block = steps // num_blocks
60
61 for num_block in range(num_blocks):
62 block_mask_index = (x[:, prompt.shape[1] + num_block * block_length: prompt.shape[1] + (num_block + 1) * block_length:] == mask_id)
63 num_transfer_tokens = get_num_transfer_tokens(block_mask_index, steps_per_block)
64
65 for i in range(steps_per_block):
66 mask_index = (x == mask_id)
67 if not mask_index.any():
68 break
69
70 outputs = model(x)
71 logits = outputs.logits
72
73 logits_with_noise = add_gumbel_noise(logits, temperature=temperature)
74 x0 = torch.argmax(logits_with_noise, dim=-1) # b, l
75
76 if remasking == 'low_confidence':
77 p = torch.nn.functional.softmax(logits.to(torch.float64), dim=-1)
78 x0_p = torch.squeeze(
79 torch.gather(p, dim=-1, index=torch.unsqueeze(x0, -1)), -1) # b, l
80 elif remasking == 'random':
81 x0_p = torch.rand((x0.shape[0], x0.shape[1]), device=x0.device)
82 else:
83 raise NotImplementedError(remasking)
84
85 x0_p[:, prompt.shape[1] + (num_block + 1) * block_length:] = -float('inf')
86
87 x0 = torch.where(mask_index, x0, x)
88 confidence = torch.where(mask_index, x0_p, -float('inf'))
89
90 transfer_index = torch.zeros_like(x0, dtype=torch.bool, device=x0.device)
91 for j in range(confidence.shape[0]):
92 _, select_index = torch.topk(confidence[j], k=num_transfer_tokens[j, i])
93 transfer_index[j, select_index] = True
94 x[transfer_index] = x0[transfer_index]
95
96 return x
97
98def chat_completion(model, tokenizer, messages, temperature=0.1, gen_length=128, steps=128):
99 """
100 Generate a chat completion.
101
102 Args:
103 model: The LLaDA tool calling model
104 tokenizer: The tokenizer
105 messages: List of message dictionaries with 'role' and 'content' keys
106 temperature: Temperature for generation (0 for greedy)
107 gen_length: Maximum length of generated text
108 steps: Number of denoising steps
109
110 Returns:
111 The generated response text
112 """
113 # Format input for the model
114 formatted_input = tokenizer.apply_chat_template(
115 messages,
116 tokenize=False,
117 add_generation_prompt=True
118 )
119
120 # Tokenize input
121 input_ids = tokenizer(formatted_input, return_tensors="pt")["input_ids"]
122
123 # Generate response
124 with torch.no_grad():
125 output_ids = generate(
126 model,
127 input_ids,
128 steps=steps,
129 gen_length=gen_length,
130 block_length=32,
131 temperature=temperature,
132 remasking='low_confidence'
133 )
134
135 # Decode the generated output
136 generated_text = tokenizer.decode(output_ids[0, input_ids.shape[1]:], skip_special_tokens=False).split("<|")[0]
137 return generated_text
138
139# Example usage
140if __name__ == "__main__":
141 # Load the base model and tokenizer
142 model_name = "Proximile/LLaDA-8B-Tools"
143 tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
144 model = AutoModel.from_pretrained(model_name, trust_remote_code=True, device_map="auto")
145
146 # Define tool calling function schema
147 tool_schema = [
148 {
149 "type": "function",
150 "function": {
151 "name": "get_weather",
152 "description": "Get the current weather in a given location",
153 "parameters": {
154 "type": "object",
155 "properties": {
156 "location": {
157 "type": "string",
158 "description": "The city and state, e.g. San Francisco, CA"
159 },
160 "unit": {
161 "type": "string",
162 "enum": ["celsius", "fahrenheit"],
163 "description": "The unit of temperature"
164 }
165 },
166 "required": ["location", "unit"]
167 }
168 }
169 }
170 ]
171
172 # Create conversation with system prompt including tool description
173 system_prompt = """You are a helpful assistant with tool calling capabilities. When you receive a tool call response, use the output to format an answer to the orginal user question.
174
175If you choose to use one or more of the following tool functions, respond with a list of JSON function calls, each with the proper arguments that best answers the given prompt.
176
177Each tool request within the list should be in the exact format {"name": function name, "parameters": {dictionary of argument names and values}}. Do not use variables. Just a list of two-key dictionaries, each starting with the function name, followed by a dictionary of parameters.
178
179Here are the tool functions available to you:
180
181""" + json.dumps(tool_schema, indent=4) + """
182
183After receiving the results back from a function call, you have to formulate your response to the user. If the information needed is not found in the returned data, either attempt a new function call, or inform the user that you cannot answer based on your available knowledge. The user cannot see the function results. You have to interpret the data and provide a response based on it.
184
185If the user request does not necessitate a function call, simply respond to the user's query directly."""
186
187 messages = [
188 {"role": "system", "content": system_prompt},
189 {"role": "user", "content": "What's the weather like in New York?"}
190 ]
191
192 # Generate assistant response (expecting tool call)
193 assistant_response = chat_completion(model, tokenizer, messages)
194 print(f"Assistant: {assistant_response}")
195
196 # Mock tool response
197 tool_response = json.dumps({
198 "location": "New York, NY",
199 "temperature": 72,
200 "unit": "fahrenheit",
201 "condition": "Partly Cloudy",
202 "humidity": 65,
203 "wind_speed": 8,
204 "wind_direction": "NE"
205 })
206
207 # Add assistant and tool responses to the conversation
208 messages.append({"role": "assistant", "content": assistant_response})
209 messages.append({"role": "ipython", "content": tool_response})
210
211 # Generate final assistant response
212 final_response = chat_completion(model, tokenizer, messages)
213 print(f"Assistant (with tool data): {final_response}")
214
215# Assistant: [{"name": "get_weather", "parameters": {"location": "New York", "unit": "fahrenheit"}}]
216# Assistant (with tool data): The current weather in New York is as follows:
217# - Temperature: 72°F
218# - Weather Condition: Partly Cloudy
219# - Humidity: 65%
220# - Wind Speed: 8 miles per hour
221# - Wind Direction: Northeast