Kaggle AI数学奥赛挑战赛 高分思路：GPT-OSS-120b 搭配 思维链 Python工具调用

• 赛题名称：AI Mathematical Olympiad - Progress Prize 3 (AIMO Progress Prize 3)
• 赛题类型：大模型
• 赛题任务：用 LaTeX 格式编写的奥林匹克水平的数学问题

https://www.kaggle.com/competitions/ai-mathematical-olympiad-progress-prize-3/overview

unsetunset赛题背景 unsetunset

AI 数学奥林匹克 (AIMO) 是一项旨在推动人工智能在数学推理领域进步的竞赛，由 $1000 万美元的基金支持。数学推理被认为是 AI 的一个关键里程碑。

• 现状挑战： 尽管闭源模型已在 2025 年的 IMO（国际数学奥林匹克）中达到金牌水平，展现了解决全球最具挑战性高中数学问题的能力，但开源模型的能力仍存在显著差距（例如，商业模型能解决 AIMO2 公共排行榜 50 题中的 50 题，而最高 Kaggle 得分仅为 34/50）。
• 本届目标 (AIMO3)： 本次竞赛是第三届 AIMO 进步奖，旨在加速开源社区的进展，缩小与闭源模型的差距。

unsetunset高分思路unsetunset

“[Way to 30+/50] GPT-OSS - w/ python tool”方案目基于开源的 GPT-OSS-120b大语言模型，最终获得了35分的优异成绩。

https://www.kaggle.com/code/shelterw/way-to-30-50-gpt-oss-w-python-tool

• 要点1：GPT-OSS-120b
• 要点2：Self-Consistency Strategy
• 要点3：Python数学工具库精确计算

unsetunsetGPT-OSS-120b unsetunset

OpenAI最新发布的GPT-OSS系列（Open Source Series）代表了开源大模型领域的重要突破。这一系列模型专门为复杂推理、智能体任务和多样化开发场景而设计，填补了开源社区在高性能推理模型方面的空白。

• 参数规模：1170亿参数，其中51亿为激活参数（MoE架构）
• 硬件要求：单个80GB GPU（如NVIDIA H100或AMD MI300X）
• 适用场景：生产环境、通用目的、高推理要求任务

OpenAI Harmony 是专门为GPT-OSS系列开源模型设计的标准化响应格式系统。它不仅仅是一个简单的对话模板，而是一个完整的结构化通信协议，确保模型能够正确地进行复杂推理、工具调用和思维链展示。

Harmony格式的核心在于多通道并行输出机制，允许模型同时在不同频道中生成内容：

assistant
analysis思考过程：我需要先获取用户位置...
commentary调用工具：get_location()
final答案是：旧金山天气晴朗...

unset unset智能体与工具协同unsetunset

将大语言模型推理能力与外部工具执行能力深度融合，模型不再是单纯的文本生成器，而是能够自主决定何时、如何使用工具的智能代理。

1. 初始化层：资源配置与模型准备

    
def __init__(self, model_path: str, max_model_len: int = MAX_LEN, ...):
    # 模型配置
    self.model_path = model_path
    self.model = "gpt-oss"
    
    # 推理参数（精细控制）
    self.temperature = temperature  # 创造性控制
    self.top_p = top_p              # 核采样
    self.min_p = min_p              # 最小概率阈值
    self.seed = seed                # 可重现性
    self.k = k                      # 自一致性样本数
    
    # 时间预算管理
    self.base_budget = 60 * 5.5    # 基础5.5分钟
    self.budget = 370               # 动态预算
    self.deadline = None            # 截止时间
    
    # 客户端连接
    self.client = OpenAI(base_url="http://127.0.0.1:8000/v1", ...)
    self.tokenizer = AutoTokenizer.from_pretrained(model_path, ...)

创新点：将时间作为第一类资源进行管理，实现智能时间分配。

2. 推理策略层：动态采样与预算分配

• 自适应样本数计算

def get_num_samples(self) -> int:
    


    
"""基于剩余预算智能决定采样数量"""
    if not self.use_budget:
        return self.k  # 固定采样
    
    # 公式：(预算 - 固定开销) / 单次推理耗时
    estimated = (self.budget - 190) / 90
    ret = min(self.k, math.floor(estimated))
    return max(1, ret)  # 至少采样1次

• TIR提示工程多样化

def format_prompts(self, problem: str) -> list[str]:
    """为自一致性生成多样化提示"""
    num_samples = self.get_num_samples()
    prompts = []
    for i in range(num_samples):
        tir_prompt = TIR_PROMPTS[i % len(TIR_PROMPTS)]  # 轮换提示模板
        prompts.append(problem + "\n\n" + tir_prompt)
    return prompts

• 提示多样性：不同角度思考同一问题
• 模板轮换：避免思维定式

3. 执行层：迭代式工具集成推理

• 单次推理的核心循环

def single_generate_tir(self, prompt: str, stop_event: threading.Event) -> str:
    python_tool = PythonTool(execution_backend="jupyter")
    messages = self.apply_chat_template(prompt, python_tool)
    
    for iteration in range(self.max_iter):  # 最多300次迭代
        # 1. 条件检查（截止时间、停止信号等）
        if self.deadline and time.time() >= self.deadline:
            break
        
        # 2. 渲染对话为token序列
        prompt_ids = encoding.render_conversation_for_completion(...)
        max_tokens = self.max_model_len - len(prompt_ids)
        
        # 3. 流式生成
        stream = self.client.completions.create(
            model=self.model,
            prompt=prompt_ids,
            max_tokens=max_tokens,
            temperature=self.temperature,
            stream=True,  # 关键：流式处理
            extra_body=dict(
                stop_token_ids=self.stop_token_ids,  # 停止token控制
                return_token_ids=True,
            )
        )
        
        # 4. 实时监控与中断检测
        token_buffer = []
        token_buffer_str = ""
        for chunk in stream:
            # 检查停止条件
            if stop_event.is_set():
                break
            
            # 收集token和文本
            token_buffer.extend(chunk.choices[0].token_ids)
            token_buffer_str += chunk.choices[0].text
            
            # 关键：实时检测boxed答案
            if"}"in text_chunk and self.extract_boxed_text(token_buffer_str):
                final_answer_found = token_buffer_str
                break
        
        stream.close()
        
        # 5. 解析生成内容
        if token_buffer:
            new_messages = encoding.parse_messages_from_completion_tokens(...)
            messages.extend(new_messages)
            
            last_message = messages[-1]
            
            # 检查生成是否完成
            if last_message.channel == "final":
                break
            
            # 6. 工具调用检测与执行
            if last_message.recipient == "python":  # 模型主动调用工具


    

                print(f"🐍 Executing Python code...")
                response_msgs = python_tool.process_sync_plus(last_message)
                messages.extend(response_msgs)  # 将工具结果加入对话历史

• 流式检测技术

for chunk in stream:
    token_buffer_str += chunk.choices[0].text
    
    # 智能中断：一旦检测到完整答案立即停止
    if "}" in text_chunk and self.extract_boxed_text(token_buffer_str):
        final_answer_found = token_buffer_str
        breaking = True
        break

需等待完整生成，避免不必要的token生成，显著降低延迟。

• 多重答案提取模式

def extract_boxed_text(self, text: str) -> int | None:
    # 模式1：LaTeX格式 \boxed{答案}
    pattern = r'oxed{(.*?)}'# 注意：实际转义为\\boxed{}
    matches = re.findall(pattern, str(text))
    
    # 模式2：自然语言格式
    pattern = r'(?i)final\s+answer\s*(?:is|:)?\s*(\d+)'
    matches = re.findall(pattern, text)
    
    # 清理和验证
    clean_match = match.strip().replace(',', '').replace(' ', '')
    val = int(float(clean_match[:20]))  # 防止溢出
    if0 <= val <= 99999:  # 有效范围检查
        return val

智能多数表决机制

def _inference_parallel(self, prompts: list[str]) -> list[str]:
    """并行推理，采用多数表决早停"""
    stop_event = threading.Event()
    answers_collected = []
    majority_threshold = len(prompts) / 2# 过半阈值
    
    with ThreadPoolExecutor(max_workers=self.k) as executor:
        future_to_idx = {
            executor.submit(self.single_generate_tir, p, stop_event): i
            for i, p in enumerate(prompts)
        }
        
        for future in as_completed(future_to_idx):
            idx = future_to_idx[future]
            result_text = future.result()
            
            ans = self.extract_boxed_text(result_text)
            if ans isnotNone:
                answers_collected.append(ans)
                counts = Counter(answers_collected)
                most_common_ans, count = counts.most_common(1)[0]
                
                # 关键：达成多数一致后立即停止
                if count > majority_threshold:
                    print(f"🎯 Majority reached! {most_common_ans} appeared {count} times")
                    stop_event.set()  # 通知所有线程停止
                    break

unset unset工具集成逻辑unsetunset

对话状态机

初始状态
    ↓
模型思考 → [analysis通道]
    ↓
可能需要工具 → [recipient="python"]
    ↓
工具执行 ← PythonTool.process_sync_plus()
    ↓
结果返回 → [tool通道]  
    ↓
模型继续推理
    ↓
...循环直到final通道...

多通道协调

# 模型可以同时在不同通道输出
analysis我需要计算积分...
commentary调用python计算...
final最终答案是42

#  学习大模型 & 讨论Kaggle  #

每天大模型、算法竞赛、干货资讯