Qwen3.6-27B 深度解析：27B 密集模型如何超越 397B MoE — 架构创新、本地部署与 Agentic Coding 实战

#!/usr/bin/env python3
"""
Qwen3.6-27B 本地测试脚本
通过 llama.cpp 的 OpenAI 兼容 API 与本地模型交互
"""

import openai
import json

# 配置本地 llama-server 的 API endpoint
client = openai.OpenAI(
    base_url="http://localhost:8080/v1",
    api_key="not-needed"  # 本地部署不需要 API key
)

def test_coding_ability():
    """测试 Qwen3.6-27B 的编程能力"""
    
    test_cases = [
        {
            "name": "Python 算法实现",
            "prompt": "用 Python 实现一个 LRU Cache，要求 O(1) 时间复杂度的 get 和 put 操作。包含完整的类型注解和 docstring。"
        },
        {
            "name": "Bug 修复",
            "prompt": "找出以下 Python 代码中的 bug 并修复：\n\ndef find_duplicates(nums):\n    seen = set()\n    dupes = []\n    for n in nums:\n        if n in seen:\n            dupes.append(n)\n        seen.add(n)\n    return dupes\n\nprint(find_duplicates([1, 2, 2, 3, 1]))"
        },
        {
            "name": "代码重构",
            "prompt": "将以下 Python 函数重构为使用 async/await 的异步版本：\n\ndef fetch_urls(urls):\n    import urllib.request\n    results = []\n    for url in urls:\n        response = urllib.request.urlopen(url)\n        results.append(response.read())\n    return results"
        }
    ]
    
    for test in test_cases:
        print(f"\n{'='*60}")
        print(f"测试: {test['name']}")
        print(f"{'='*60}")
        
        response = client.chat.completions.create(
            model="Qwen3.6-27B",
            messages=[
                {"role": "system", "content": "You are an expert programmer. Provide clear, correct, and well-documented code solutions."},
                {"role": "user", "content": test["prompt"]}
            ],
            temperature=0.3,
            max_tokens=2048
        )
        
        print(response.choices[0].message.content)
        print(f"\n--- Token 使用: prompt={response.usage.prompt_tokens}, completion={response.usage.completion_tokens} ---")

def test_reasoning_ability():
    """测试 Qwen3.6-27B 的推理能力（启用 reasoning 模式）"""
    
    response = client.chat.completions.create(
        model="Qwen3.6-27B",
        messages=[
            {"role": "user", "content": "一个农场有鸡和兔子，总共有 35 个头和 94 只脚。请问鸡和兔子各有多少只？请用方程组求解并解释每一步。"}
        ],
        temperature=0.6,
        max_tokens=2048,
        extra_body={"reasoning": True}  # 启用推理模式
    )
    
    print("\n推理能力测试:")
    print(response.choices[0].message.content)

if __name__ == "__main__":
    print("Qwen3.6-27B 本地测试")
    print("请确保 llama-server 已在 http://localhost:8080 启动")
    test_coding_ability()
    test_reasoning_ability()

#!/usr/bin/env python3
"""
Qwen3.6-27B 性能基准测试脚本
测试推理速度、Token 吞吐量和内存占用
"""

import time
import statistics
import openai
import psutil
import os

class QwenBenchmark:
    def __init__(self, base_url="http://localhost:8080/v1"):
        self.client = openai.OpenAI(
            base_url=base_url,
            api_key="not-needed"
        )
        self.results = []
    
    def measure_throughput(self, prompt: str, max_tokens: int = 512) -> dict:
        """测量单次推理的吞吐量"""
        
        # 记录开始前的内存
        process = psutil.Process(os.getpid())
        mem_before = process.memory_info().rss / 1024 / 1024  # MB
        
        start_time = time.time()
        
        response = self.client.chat.completions.create(
            model="Qwen3.6-27B",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.6,
            max_tokens=max_tokens,
            stream=False
        )
        
        end_time = time.time()
        mem_after = process.memory_info().rss / 1024 / 1024
        
        total_time = end_time - start_time
        output_tokens = response.usage.completion_tokens
        input_tokens = response.usage.prompt_tokens
        
        return {
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "total_time": total_time,
            "throughput": output_tokens / total_time if total_time > 0 else 0,
            "ttft": total_time,  # Time to first token（非流式下近似）
            "memory_mb": mem_after - mem_before,
            "output_text": response.choices[0].message.content[:200]
        }
    
    def run_benchmark_suite(self, num_runs: int = 5):
        """运行完整的基准测试套件"""
        
        test_prompts = [
            "用 Python 写一个快速排序算法，包含类型注解和测试用例",
            "解释 Transformer 架构中的 Self-Attention 机制，用数学公式说明",
            "写一个 Python 脚本，实现一个简单的 HTTP 服务器，支持文件浏览和下载",
            "将以下 SQL 查询优化为更高效的版本：SELECT * FROM users WHERE age > 18 AND city = 'Beijing' ORDER BY name",
            "用 Rust 实现一个线程安全的计数器，支持并发递增和读取"
        ]
        
        print(f"\n{'='*70}")
        print(f"Qwen3.6-27B 基准测试 ({num_runs} 轮)")
        print(f"{'='*70}\n")
        
        for prompt in test_prompts:
            print(f"\n测试: {prompt[:50]}...")
            run_results = []
            
            for i in range(num_runs):
                result = self.measure_throughput(prompt)
                run_results.append(result)
                print(f"  轮次 {i+1}: {result['throughput']:.1f} tok/s, "
                      f"输出 {result['output_tokens']} tokens, "
                      f"耗时 {result['total_time']:.2f}s")
            
            # 统计摘要
            throughputs = [r["throughput"] for r in run_results]
            avg_throughput = statistics.mean(throughputs)
            p50_throughput = statistics.median(throughputs)
            p95_throughput = sorted(throughputs)[int(len(throughputs) * 0.95)]
            
            print(f"  平均: {avg_throughput:.1f} tok/s | "
                  f"P50: {p50_throughput:.1f} tok/s | "
                  f"P95: {p95_throughput:.1f} tok/s")
            
            self.results.append({
                "prompt": prompt[:50],
                "avg_throughput": avg_throughput,
                "median_throughput": p50_throughput
            })
        
        # 总体报告
        print(f"\n{'='*70}")
        print("总体报告:")
        print(f"{'='*70}")
        all_throughputs = [r["avg_throughput"] for r in self.results]
        print(f"  整体平均吞吐: {statistics.mean(all_throughputs):.1f} tokens/s")
        print(f"  最佳测试:     {max(all_throughputs):.1f} tokens/s")
        print(f"  最差测试:     {min(all_throughputs):.1f} tokens/s")

if __name__ == "__main__":
    benchmark = QwenBenchmark()
    benchmark.run_benchmark_suite(num_runs=3)

#!/usr/bin/env python3
"""
Qwen3.6-27B Agentic Coding 实战：自主编程代理
这个代理能接收编程任务、生成代码、运行测试、自动修复
"""

import subprocess
import openai
import tempfile
import os
from typing import Optional

class CodingAgent:
    """基于 Qwen3.6-27B 的自主编程代理"""
    
    def __init__(self, api_url: str = "http://localhost:8080/v1"):
        self.client = openai.OpenAI(
            base_url=api_url,
            api_key="not-needed"
        )
        self.max_iterations = 5  # 最大修复迭代次数
    
    def generate_code(self, task: str, language: str = "python") -> str:
        """根据任务描述生成代码"""
        
        response = self.client.chat.completions.create(
            model="Qwen3.6-27B",
            messages=[
                {
                    "role": "system",
                    "content": f"""You are an expert {language} programmer.
Generate clean, correct, and well-documented code.
Only output the code, no explanations unless asked.
Include type hints and docstrings."""
                },
                {"role": "user", "content": task}
            ],
            temperature=0.3,
            max_tokens=4096
        )
        
        return response.choices[0].message.content
    
    def run_code(self, code: str, test_input: Optional[str] = None) -> tuple[bool, str]:
        """运行生成的代码并返回结果"""
        
        with tempfile.NamedTemporaryFile(
            mode='w', suffix='.py', delete=False
        ) as f:
            f.write(code)
            if test_input:
                f.write(f"\n\n# Test:\n{test_input}")
            f.flush()
            
            try:
                result = subprocess.run(
                    ['python3', f.name],
                    capture_output=True,
                    text=True,
                    timeout=30
                )
                
                success = result.returncode == 0
                output = result.stdout if success else result.stderr
                
                return success, output
            
            except subprocess.TimeoutExpired:
                return False, "代码执行超时（30秒）"
            
            finally:
                os.unlink(f.name)
    
    def fix_code(self, code: str, error: str) -> str:
        """根据错误信息修复代码"""
        
        response = self.client.chat.completions.create(
            model="Qwen3.6-27B",
            messages=[
                {
                    "role": "system",
                    "content": """You are an expert debugger. 
Given code and an error message, fix the bug.
Output only the corrected code."""
                },
                {
                    "role": "user",
                    "content": f"Fix the bug in this code:\n\n{code}\n\nError: {error}"
                }
            ],
            temperature=0.2,
            max_tokens=4096
        )
        
        return response.choices[0].message.content
    
    def solve(self, task: str, test_input: Optional[str] = None) -> dict:
        """完整求解流程：生成 → 测试 → 修复（最多 max_iterations 次）"""
        
        print(f"🤖 接收任务: {task[:80]}...")
        
        # 第一轮：生成代码
        code = self.generate_code(task)
        print(f"📝 代码已生成 ({len(code)} 字符)")
        
        for iteration in range(self.max_iterations):
            # 运行测试
            success, output = self.run_code(code, test_input)
            
            if success:
                print(f"✅ 第 {iteration + 1} 轮测试通过!")
                print(f"输出: {output[:200]}")
                return {
                    "success": True,
                    "code": code,
                    "output": output,
                    "iterations": iteration + 1
                }
            
            print(f"❌ 第 {iteration + 1} 轮测试失败: {output[:100]}")
            
            # 尝试修复
            if iteration < self.max_iterations - 1:
                print("🔧 正在修复...")
                code = self.fix_code(code, output)
        
        return {
            "success": False,
            "code": code,
            "error": output,
            "iterations": self.max_iterations
        }

# 使用示例
if __name__ == "__main__":
    agent = CodingAgent()
    
    # 示例任务
    result = agent.solve(
        task="写一个 Python 函数，实现二分查找。输入一个有序数组和一个目标值，返回目标值的索引，如果不存在返回 -1。",
        test_input="""
arr = [1, 3, 5, 7, 9, 11, 13, 15]
print(binary_search(arr, 7))   # 应输出: 3
print(binary_search(arr, 4))   # 应输出: -1
print(binary_search(arr, 15))  # 应输出: 7
"""
    )
    
    print(f"\n最终结果: {'成功' if result['success'] else '失败'}")
    print(f"迭代次数: {result['iterations']}")