Qwen3-4B-Thinking-2507-GPT-5-Codex-Distill实战:CLI命令行调用封装教程

1. 引言:从Web界面到命令行,解锁模型调用新姿势

如果你已经成功部署了Qwen3-4B-Thinking-2507-GPT-5-Codex-Distill模型,并且通过Chainlit的Web界面体验过它的文本生成能力,那么恭喜你,你已经迈出了第一步。但Web界面只是冰山一角,真正的灵活性和自动化潜力,藏在命令行接口(CLI)里。

想象一下这些场景:你想批量处理几十个文本生成任务,需要把模型集成到自己的自动化脚本里,或者想在服务器上定时运行一些生成任务。这时候,每次都打开浏览器、点击界面、复制粘贴,就显得效率低下了。CLI调用就是解决这些问题的钥匙。

今天这篇文章,我要带你从Web界面使用者,升级为命令行高手。我会手把手教你如何封装一个简单实用的CLI工具,让你能像使用系统命令一样调用这个强大的文本生成模型。无论你是开发者、研究人员,还是自动化爱好者,这套方法都能让你的工作流效率翻倍。

2. 理解基础:模型服务与调用原理

在开始动手之前,我们先花几分钟搞清楚背后的原理。这样你不仅能照着做,还能理解为什么要这么做,以后遇到问题也能自己解决。

2.1 模型是如何被调用的?

当你通过Chainlit界面提问时,背后发生了什么?简单来说,Chainlit这个Web应用,实际上是一个“中间人”。它接收你在浏览器里输入的问题,然后按照特定的格式和协议,发送请求给真正运行模型的vLLM服务,拿到结果后再展示给你看。

这个“特定的格式和协议”,就是HTTP API。vLLM部署的模型通常会提供一个标准的API接口,允许其他程序通过发送HTTP请求来调用模型。Chainlit做的就是这件事——它封装了这些HTTP请求的细节,让你通过友好的界面就能使用。

2.2 为什么需要CLI封装?

既然Chainlit已经很好用了,为什么还要折腾CLI呢?原因有几个:

  • 自动化集成:CLI可以轻松集成到Shell脚本、Python程序、定时任务(cron)中
  • 批量处理:一次性处理多个文件或大量文本,无需手动操作
  • 远程调用:通过SSH连接到服务器,直接命令行调用,无需图形界面
  • 管道操作:可以和其他命令行工具组合使用,形成处理流水线
  • 资源节约:在某些场景下,CLI调用比启动Web界面更轻量

我们的目标,就是绕过Chainlit这个“中间人”,直接和模型的API对话,并且把这个过程封装成简单易用的命令行工具。

3. 环境准备与API探索

在开始封装之前,我们需要先确认一些基本信息,并准备好开发环境。

3.1 确认模型服务状态

首先,确保你的模型服务正在正常运行。按照你提供的部署说明,可以通过以下命令查看日志:

cat /root/workspace/llm.log

你应该能看到类似这样的输出,表明vLLM服务已经启动并加载了模型:

INFO 07-28 10:30:15 llm_engine.py:72] Initializing an LLM engine...
INFO 07-28 10:30:20 model_runner.py:84] Loading model weights...
INFO 07-28 10:30:25 model_runner.py:121] Model loaded successfully.
INFO 07-28 10:30:25 llm_engine.py:189] LLM engine initialized.

如果服务没有运行,你需要先启动它。通常vLLM的启动命令类似这样:

python -m vllm.entrypoints.openai.api_server \
    --model /path/to/your/model \
    --served-model-name qwen3-4b-thinking \
    --port 8000 \
    --host 0.0.0.0

3.2 探索API接口

vLLM默认提供了OpenAI兼容的API接口。这意味着它的API调用方式和OpenAI的官方API非常相似。我们可以先通过简单的curl命令测试一下API是否可用:

curl http://localhost:8000/v1/models

如果API正常,你会得到一个JSON响应,里面包含了模型的信息:

{
  "object": "list",
  "data": [
    {
      "id": "qwen3-4b-thinking",
      "object": "model",
      "created": 1677610602,
      "owned_by": "vllm"
    }
  ]
}

再测试一下聊天接口:

curl http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{
    "model": "qwen3-4b-thinking",
    "messages": [
      {
        "role": "user",
        "content": "你好,请介绍一下你自己"
      }
    ],
    "temperature": 0.7,
    "max_tokens": 100
  }'

如果一切正常,你会收到模型生成的回复。这个curl命令虽然能工作,但每次都要写这么长的JSON,显然不够方便。接下来,我们就来封装一个更友好的CLI工具。

4. 构建Python CLI封装工具

现在进入正题,我们来创建一个Python脚本,把复杂的API调用封装成简单的命令行工具。

4.1 创建基础脚本框架

首先创建一个新的Python文件,比如叫做qwen_cli.py

#!/usr/bin/env python3
"""
Qwen3-4B-Thinking-2507-GPT-5-Codex-Distill 命令行调用工具
一个简单易用的CLI封装,用于调用vLLM部署的文本生成模型
"""

import argparse
import json
import sys
from typing import Optional, List, Dict, Any
import requests
from requests.exceptions import RequestException
import time

class QwenCLIClient:
    """Qwen模型命令行客户端"""
    
    def __init__(self, base_url: str = "http://localhost:8000"):
        """
        初始化客户端
        
        Args:
            base_url: vLLM API服务器地址,默认localhost:8000
        """
        self.base_url = base_url.rstrip('/')
        self.chat_url = f"{self.base_url}/v1/chat/completions"
        
    def test_connection(self) -> bool:
        """测试与API服务器的连接"""
        try:
            response = requests.get(f"{self.base_url}/v1/models", timeout=5)
            return response.status_code == 200
        except RequestException:
            return False
    
    def generate_chat_response(
        self,
        message: str,
        system_prompt: Optional[str] = None,
        temperature: float = 0.7,
        max_tokens: int = 512,
        stream: bool = False
    ) -> str:
        """
        生成聊天回复
        
        Args:
            message: 用户输入的消息
            system_prompt: 系统提示词(可选)
            temperature: 温度参数,控制随机性
            max_tokens: 最大生成token数
            stream: 是否使用流式输出
            
        Returns:
            模型生成的回复文本
        """
        # 构建消息列表
        messages = []
        
        if system_prompt:
            messages.append({
                "role": "system",
                "content": system_prompt
            })
        
        messages.append({
            "role": "user",
            "content": message
        })
        
        # 构建请求数据
        data = {
            "model": "qwen3-4b-thinking",
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens,
            "stream": stream
        }
        
        try:
            if stream:
                return self._stream_response(data)
            else:
                return self._get_complete_response(data)
        except RequestException as e:
            return f"请求失败: {str(e)}"
    
    def _get_complete_response(self, data: Dict[str, Any]) -> str:
        """获取完整的非流式响应"""
        response = requests.post(
            self.chat_url,
            json=data,
            headers={"Content-Type": "application/json"},
            timeout=60
        )
        
        if response.status_code != 200:
            return f"API请求失败: {response.status_code}\n{response.text}"
        
        result = response.json()
        return result['choices'][0]['message']['content']
    
    def _stream_response(self, data: Dict[str, Any]) -> str:
        """处理流式响应"""
        full_response = []
        
        response = requests.post(
            self.chat_url,
            json=data,
            headers={"Content-Type": "application/json"},
            stream=True,
            timeout=60
        )
        
        if response.status_code != 200:
            return f"API请求失败: {response.status_code}"
        
        for line in response.iter_lines():
            if line:
                line = line.decode('utf-8')
                if line.startswith('data: '):
                    data_str = line[6:]
                    if data_str == '[DONE]':
                        break
                    try:
                        data_json = json.loads(data_str)
                        if 'choices' in data_json and data_json['choices']:
                            delta = data_json['choices'][0]['delta']
                            if 'content' in delta:
                                content = delta['content']
                                print(content, end='', flush=True)
                                full_response.append(content)
                    except json.JSONDecodeError:
                        continue
        
        print()  # 换行
        return ''.join(full_response)

def main():
    """主函数:解析命令行参数并执行相应操作"""
    parser = argparse.ArgumentParser(
        description='Qwen3-4B-Thinking-2507-GPT-5-Codex-Distill 命令行调用工具',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
使用示例:
  %(prog)s "请写一首关于春天的诗"
  %(prog)s -f input.txt -o output.txt
  %(prog)s --system "你是一个编程助手" "解释Python的装饰器"
  %(prog)s --stream "讲一个有趣的故事"
        """
    )
    
    # 添加参数
    parser.add_argument(
        'message',
        nargs='?',
        help='直接输入要发送给模型的消息'
    )
    
    parser.add_argument(
        '-f', '--file',
        help='从文件读取输入消息'
    )
    
    parser.add_argument(
        '-o', '--output',
        help='将输出保存到文件'
    )
    
    parser.add_argument(
        '-s', '--system',
        help='系统提示词,用于设定模型角色'
    )
    
    parser.add_argument(
        '-t', '--temperature',
        type=float,
        default=0.7,
        help='温度参数 (0.0-2.0),默认0.7'
    )
    
    parser.add_argument(
        '-m', '--max-tokens',
        type=int,
        default=512,
        help='最大生成token数,默认512'
    )
    
    parser.add_argument(
        '--stream',
        action='store_true',
        help='使用流式输出(逐字显示)'
    )
    
    parser.add_argument(
        '--url',
        default='http://localhost:8000',
        help='API服务器URL,默认http://localhost:8000'
    )
    
    parser.add_argument(
        '--test',
        action='store_true',
        help='测试API连接'
    )
    
    args = parser.parse_args()
    
    # 创建客户端实例
    client = QwenCLIClient(base_url=args.url)
    
    # 测试连接
    if args.test:
        print("正在测试API连接...")
        if client.test_connection():
            print("✓ 连接成功!")
            sys.exit(0)
        else:
            print("✗ 连接失败,请检查:")
            print("  1. vLLM服务是否运行?")
            print(f"  2. URL是否正确: {args.url}")
            print("  3. 防火墙或网络设置")
            sys.exit(1)
    
    # 获取输入消息
    message = ""
    if args.file:
        try:
            with open(args.file, 'r', encoding='utf-8') as f:
                message = f.read().strip()
        except FileNotFoundError:
            print(f"错误:文件 '{args.file}' 不存在")
            sys.exit(1)
    elif args.message:
        message = args.message
    else:
        # 如果没有提供消息,进入交互模式
        print("Qwen CLI 工具 - 交互模式 (输入 'quit' 或 'exit' 退出)")
        print("=" * 50)
        
        while True:
            try:
                user_input = input("\n你: ").strip()
                if user_input.lower() in ['quit', 'exit', '退出']:
                    break
                if not user_input:
                    continue
                
                print("\n模型: ", end='')
                start_time = time.time()
                response = client.generate_chat_response(
                    message=user_input,
                    system_prompt=args.system,
                    temperature=args.temperature,
                    max_tokens=args.max_tokens,
                    stream=args.stream
                )
                elapsed = time.time() - start_time
                
                if not args.stream:
                    print(response)
                
                print(f"\n[生成耗时: {elapsed:.2f}秒]")
                
            except KeyboardInterrupt:
                print("\n\n再见!")
                break
        
        sys.exit(0)
    
    # 生成响应
    if not message:
        print("错误:未提供输入消息")
        parser.print_help()
        sys.exit(1)
    
    print("正在生成响应...")
    start_time = time.time()
    
    response = client.generate_chat_response(
        message=message,
        system_prompt=args.system,
        temperature=args.temperature,
        max_tokens=args.max_tokens,
        stream=args.stream
    )
    
    elapsed = time.time() - start_time
    
    if not args.stream:
        print("\n" + "=" * 50)
        print("模型回复:")
        print("=" * 50)
        print(response)
        print("=" * 50)
    
    # 保存输出到文件
    if args.output:
        try:
            with open(args.output, 'w', encoding='utf-8') as f:
                f.write(response)
            print(f"\n输出已保存到: {args.output}")
        except Exception as e:
            print(f"保存文件失败: {str(e)}")
    
    print(f"\n生成耗时: {elapsed:.2f}秒")

if __name__ == "__main__":
    main()

4.2 安装依赖并测试

这个脚本只需要Python标准库和requests,如果你的环境没有requests,先安装一下:

pip install requests

然后给脚本添加执行权限(Linux/Mac):

chmod +x qwen_cli.py

现在测试一下基本功能:

# 测试API连接
python qwen_cli.py --test

# 简单提问
python qwen_cli.py "你好,请介绍一下你自己"

# 使用系统提示词
python qwen_cli.py --system "你是一个编程助手" "解释Python的装饰器"

# 流式输出
python qwen_cli.py --stream "讲一个有趣的故事"

# 从文件读取输入
echo "写一首关于秋天的诗" > input.txt
python qwen_cli.py -f input.txt -o output.txt

5. 高级功能与实用技巧

基础功能已经实现了,但一个真正好用的CLI工具还需要一些高级功能和实用技巧。让我们来增强它。

5.1 添加批量处理功能

在实际工作中,我们经常需要批量处理多个问题。让我们添加这个功能:

# 在QwenCLIClient类中添加这个方法
def batch_process(
    self,
    input_file: str,
    output_file: str,
    system_prompt: Optional[str] = None,
    temperature: float = 0.7,
    max_tokens: int = 512,
    delay: float = 0.5
) -> None:
    """
    批量处理文件中的多个问题
    
    Args:
        input_file: 输入文件路径,每行一个问题
        output_file: 输出文件路径
        system_prompt: 系统提示词
        temperature: 温度参数
        max_tokens: 最大token数
        delay: 请求间隔(秒),避免过快请求
    """
    try:
        with open(input_file, 'r', encoding='utf-8') as f:
            questions = [line.strip() for line in f if line.strip()]
    except FileNotFoundError:
        print(f"错误:输入文件 '{input_file}' 不存在")
        return
    
    if not questions:
        print("输入文件为空")
        return
    
    print(f"找到 {len(questions)} 个问题,开始批量处理...")
    print("-" * 50)
    
    results = []
    for i, question in enumerate(questions, 1):
        print(f"[{i}/{len(questions)}] 处理: {question[:50]}...")
        
        response = self.generate_chat_response(
            message=question,
            system_prompt=system_prompt,
            temperature=temperature,
            max_tokens=max_tokens,
            stream=False
        )
        
        results.append({
            'question': question,
            'answer': response,
            'index': i
        })
        
        print(f"    完成")
        
        # 延迟一下,避免请求过快
        if i < len(questions):
            time.sleep(delay)
    
    # 保存结果
    try:
        with open(output_file, 'w', encoding='utf-8') as f:
            for result in results:
                f.write(f"问题 {result['index']}:\n")
                f.write(f"Q: {result['question']}\n")
                f.write(f"A: {result['answer']}\n")
                f.write("-" * 50 + "\n\n")
        
        print(f"\n批量处理完成!结果已保存到: {output_file}")
        
    except Exception as e:
        print(f"保存结果失败: {str(e)}")

# 在main函数中添加对应的命令行参数
parser.add_argument(
    '--batch',
    help='批量处理模式,指定输入文件(每行一个问题)'
)

parser.add_argument(
    '--batch-output',
    help='批量处理的输出文件'
)

parser.add_argument(
    '--delay',
    type=float,
    default=0.5,
    help='批量处理时的请求间隔(秒),默认0.5'
)

5.2 添加配置文件和默认参数

为了让工具更易用,我们可以添加配置文件支持:

import os
import configparser
from pathlib import Path

def load_config(config_path: Optional[str] = None) -> Dict[str, Any]:
    """加载配置文件"""
    default_config = {
        'api': {
            'base_url': 'http://localhost:8000',
            'timeout': 60,
            'model': 'qwen3-4b-thinking'
        },
        'generation': {
            'temperature': '0.7',
            'max_tokens': '512',
            'stream': 'false'
        },
        'ui': {
            'show_timing': 'true',
            'color_output': 'true'
        }
    }
    
    config = configparser.ConfigParser()
    config.read_dict(default_config)
    
    # 查找配置文件
    config_files = []
    if config_path:
        config_files.append(config_path)
    
    config_files.extend([
        'qwen_cli.ini',
        str(Path.home() / '.config' / 'qwen_cli.ini'),
        '/etc/qwen_cli.ini'
    ])
    
    for file_path in config_files:
        if os.path.exists(file_path):
            config.read(file_path)
            break
    
    return config

# 修改QwenCLIClient的__init__方法
def __init__(self, base_url: str = None, config_path: str = None):
    """初始化客户端,支持从配置文件读取参数"""
    config = load_config(config_path)
    
    if base_url is None:
        base_url = config.get('api', 'base_url', fallback='http://localhost:8000')
    
    self.base_url = base_url.rstrip('/')
    self.chat_url = f"{self.base_url}/v1/chat/completions"
    self.timeout = config.getint('api', 'timeout', fallback=60)
    self.model_name = config.get('api', 'model', fallback='qwen3-4b-thinking')
    
    # 保存配置供其他方法使用
    self.config = config

5.3 创建更友好的交互模式

让我们增强交互模式,添加历史记录和会话管理:

import readline  # 用于命令行历史记录
import atexit

class InteractiveSession:
    """交互式会话管理器"""
    
    def __init__(self, client: QwenCLIClient):
        self.client = client
        self.history_file = os.path.expanduser('~/.qwen_cli_history')
        self.session_history = []
        self.setup_readline()
    
    def setup_readline(self):
        """设置readline支持历史记录"""
        if os.path.exists(self.history_file):
            readline.read_history_file(self.history_file)
        
        readline.set_history_length(1000)
        atexit.register(readline.write_history_file, self.history_file)
    
    def run(self):
        """运行交互式会话"""
        print("Qwen CLI 交互模式")
        print("=" * 60)
        print("命令:")
        print("  /help     - 显示帮助")
        print("  /history  - 显示会话历史")
        print("  /clear    - 清屏")
        print("  /save     - 保存会话到文件")
        print("  /quit     - 退出")
        print("=" * 60)
        
        system_prompt = None
        temperature = 0.7
        max_tokens = 512
        
        while True:
            try:
                # 显示提示符
                prompt = "\n你"
                if system_prompt:
                    prompt += f" [{system_prompt[:20]}...]"
                prompt += ": "
                
                user_input = input(prompt).strip()
                
                if not user_input:
                    continue
                
                # 处理命令
                if user_input.startswith('/'):
                    self.handle_command(user_input[1:], system_prompt)
                    continue
                
                # 处理普通输入
                print("\n模型: ", end='', flush=True)
                start_time = time.time()
                
                response = self.client.generate_chat_response(
                    message=user_input,
                    system_prompt=system_prompt,
                    temperature=temperature,
                    max_tokens=max_tokens,
                    stream=True  # 交互模式默认使用流式
                )
                
                elapsed = time.time() - start_time
                
                # 保存到历史
                self.session_history.append({
                    'question': user_input,
                    'answer': response,
                    'timestamp': time.time()
                })
                
                print(f"\n[耗时: {elapsed:.2f}秒]")
                
            except KeyboardInterrupt:
                print("\n\n输入 /quit 退出")
            except EOFError:
                print("\n\n再见!")
                break
    
    def handle_command(self, command: str, system_prompt: str):
        """处理交互式命令"""
        cmd_parts = command.split()
        cmd = cmd_parts[0].lower() if cmd_parts else ""
        
        if cmd == 'help':
            print("\n可用命令:")
            print("  /help                   显示此帮助")
            print("  /history                显示当前会话历史")
            print("  /clear                  清屏")
            print("  /save [文件名]         保存会话到文件")
            print("  /system [提示词]       设置系统提示词")
            print("  /temp [温度值]         设置温度参数 (0.0-2.0)")
            print("  /tokens [数量]         设置最大token数")
            print("  /quit                  退出程序")
            
        elif cmd == 'history':
            if not self.session_history:
                print("当前会话没有历史记录")
            else:
                print(f"\n会话历史 ({len(self.session_history)} 条):")
                print("-" * 50)
                for i, item in enumerate(self.session_history, 1):
                    print(f"{i}. Q: {item['question'][:50]}...")
                    print(f"   A: {item['answer'][:50]}...")
                    print()
                    
        elif cmd == 'clear':
            os.system('clear' if os.name == 'posix' else 'cls')
            
        elif cmd == 'save':
            filename = cmd_parts[1] if len(cmd_parts) > 1 else f"qwen_session_{int(time.time())}.txt"
            self.save_session(filename)
            
        elif cmd == 'system':
            if len(cmd_parts) > 1:
                system_prompt = ' '.join(cmd_parts[1:])
                print(f"系统提示词已设置为: {system_prompt}")
            else:
                system_prompt = None
                print("系统提示词已清除")
                
        elif cmd == 'temp' and len(cmd_parts) > 1:
            try:
                temp = float(cmd_parts[1])
                if 0 <= temp <= 2:
                    print(f"温度参数已设置为: {temp}")
                else:
                    print("温度参数必须在 0.0 到 2.0 之间")
            except ValueError:
                print("请输入有效的数字")
                
        elif cmd == 'tokens' and len(cmd_parts) > 1:
            try:
                tokens = int(cmd_parts[1])
                if tokens > 0:
                    print(f"最大token数已设置为: {tokens}")
                else:
                    print("token数必须大于0")
            except ValueError:
                print("请输入有效的整数")
                
        elif cmd == 'quit':
            print("再见!")
            sys.exit(0)
            
        else:
            print(f"未知命令: {cmd},输入 /help 查看可用命令")

6. 实际应用场景与示例

现在工具已经相当完善了,让我们看看在实际工作中怎么用它。

6.1 场景一:自动化文档生成

假设你需要为一批API接口生成文档描述:

# 创建输入文件
cat > api_descriptions.txt << 'EOF'
用户登录接口,接收用户名和密码,返回访问令牌
商品列表查询接口,支持分页和筛选条件
订单创建接口,接收商品ID和数量,生成订单号
用户信息更新接口,支持修改邮箱和手机号
文件上传接口,支持多种格式,返回文件URL
EOF

# 批量生成文档
python qwen_cli.py --batch api_descriptions.txt --batch-output api_docs.txt \
    --system "你是一个专业的API文档编写助手,请为每个接口生成详细的文档说明,包括请求参数、响应格式和示例。"

# 查看结果
head -n 100 api_docs.txt

6.2 场景二:代码审查助手

你可以用这个工具来辅助代码审查:

# 创建一个Python脚本用于代码审查
cat > code_review.py << 'EOF'
#!/usr/bin/env python3
"""
代码审查工具 - 使用Qwen模型分析代码质量
"""

import subprocess
import sys
from pathlib import Path

def review_code(file_path: str):
    """审查单个代码文件"""
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            code_content = f.read()
    except Exception as e:
        print(f"无法读取文件 {file_path}: {e}")
        return
    
    # 构建审查提示
    prompt = f"""请审查以下Python代码,指出:
1. 潜在的错误或bug
2. 代码风格问题
3. 性能优化建议
4. 安全风险
5. 改进建议

代码:
```python
{code_content}

请给出详细的审查报告:"""

# 调用Qwen CLI
cmd = [
    sys.executable, 'qwen_cli.py',
    '--system', '你是一个经验丰富的Python代码审查专家',
    '--temperature', '0.3',
    '--max-tokens', '1024',
    prompt
]

try:
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
    if result.returncode == 0:
        return result.stdout
    else:
        return f"审查失败: {result.stderr}"
except subprocess.TimeoutExpired:
    return "审查超时"
except Exception as e:
    return f"执行失败: {str(e)}"

if name == "main": if len(sys.argv) < 2: print("用法: python code_review.py <代码文件>") sys.exit(1)

for file_path in sys.argv[1:]:
    if Path(file_path).exists():
        print(f"\n{'='*60}")
        print(f"审查文件: {file_path}")
        print(f"{'='*60}\n")
        
        review = review_code(file_path)
        print(review)
        
        # 保存审查结果
        output_file = f"{file_path}.review.txt"
        with open(output_file, 'w', encoding='utf-8') as f:
            f.write(f"文件: {file_path}\n")
            f.write(f"审查时间: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
            f.write("="*60 + "\n\n")
            f.write(review)
        
        print(f"\n审查结果已保存到: {output_file}")
    else:
        print(f"文件不存在: {file_path}")

EOF

使用示例

python code_review.py your_script.py


### 6.3 场景三:Shell脚本集成

将Qwen CLI集成到Shell脚本中,实现更复杂的工作流:

```bash
#!/bin/bash
# analyze_logs.sh - 使用Qwen分析日志文件

LOG_FILE="$1"
OUTPUT_FILE="${LOG_FILE%.*}_analysis.txt"

if [ ! -f "$LOG_FILE" ]; then
    echo "错误: 日志文件不存在: $LOG_FILE"
    exit 1
fi

echo "正在分析日志文件: $LOG_FILE"
echo "开始时间: $(date)"

# 提取关键错误信息
ERRORS=$(grep -i "error\|fail\|exception\|critical" "$LOG_FILE" | head -20)

if [ -z "$ERRORS" ]; then
    SUMMARY="日志文件中未发现明显的错误信息。"
else
    # 使用Qwen分析错误
    echo "发现错误信息,正在分析..."
    
    ANALYSIS=$(python qwen_cli.py --system "你是一个系统运维专家,擅长分析日志文件" \
        --temperature 0.3 \
        --max-tokens 800 \
        "请分析以下日志错误信息,指出可能的原因和解决建议:

${ERRORS}

请按以下格式回复:
1. 主要问题总结
2. 可能的原因分析
3. 解决建议
4. 预防措施")
    
    SUMMARY="$ANALYSIS"
fi

# 生成统计信息
TOTAL_LINES=$(wc -l < "$LOG_FILE")
ERROR_COUNT=$(grep -i "error" "$LOG_FILE" | wc -l)
WARNING_COUNT=$(grep -i "warning" "$LOG_FILE" | wc -l)

# 保存分析报告
cat > "$OUTPUT_FILE" << EOF
日志分析报告
=============

文件: $LOG_FILE
分析时间: $(date)
总行数: $TOTAL_LINES
错误数量: $ERROR_COUNT
警告数量: $WARNING_COUNT

分析结果:
$(echo "$SUMMARY" | sed 's/^/  /')

原始错误样本:
$(echo "$ERRORS" | sed 's/^/  /')

统计摘要:
- 错误率: $(echo "scale=2; $ERROR_COUNT * 100 / $TOTAL_LINES" | bc)%
- 警告率: $(echo "scale=2; $WARNING_COUNT * 100 / $TOTAL_LINES" | bc)%
EOF

echo "分析完成!报告已保存到: $OUTPUT_FILE"

7. 总结

通过这篇文章,我们完成了一个完整的Qwen3-4B-Thinking-2507-GPT-5-Codex-Distill模型CLI调用工具的开发和封装。让我们回顾一下关键要点:

7.1 工具的核心价值

这个CLI工具的价值不仅仅在于"能用命令行调用模型",更重要的是它带来的工作流变革:

  1. 自动化能力:可以集成到各种自动化脚本和流程中
  2. 批量处理:一次性处理大量任务,大幅提升效率
  3. 灵活集成:可以轻松与其他工具和系统结合
  4. 资源友好:在无GUI的服务器环境中也能使用
  5. 可扩展性:基于这个框架,你可以轻松添加更多功能

7.2 关键功能回顾

我们实现的功能包括:

  • 基础API调用:直接与vLLM的OpenAI兼容API通信
  • 多种输入方式:命令行参数、文件读取、交互式输入
  • 参数控制:温度、最大token数、系统提示词等
  • 流式输出:实时显示生成过程
  • 批量处理:处理文件中的多个问题
  • 配置管理:支持配置文件,方便管理默认参数
  • 交互模式:带有历史记录和会话管理的友好界面
  • 错误处理:完善的错误检测和用户提示

7.3 实际应用建议

在实际使用中,我有几个建议:

  1. 从简单开始:先掌握基本用法,再逐步尝试高级功能
  2. 善用配置文件:把常用的参数配置在qwen_cli.ini中,避免每次输入
  3. 结合Shell脚本:CLI工具最大的优势是可以和Shell完美结合,创造自动化工作流
  4. 注意速率限制:如果是生产环境,注意控制请求频率,避免给服务器太大压力
  5. 错误处理:重要的自动化任务要添加适当的错误处理和重试机制

7.4 进一步优化方向

如果你对这个工具还有更多需求,可以考虑:

  1. 添加缓存机制:对相同的问题缓存答案,提高响应速度
  2. 支持多模型:扩展支持其他vLLM部署的模型
  3. 添加监控指标:记录调用次数、响应时间等统计信息
  4. 实现插件系统:允许用户自定义处理流程
  5. 添加Web界面:基于这个CLI工具构建一个简单的Web界面

这个CLI工具只是一个起点,你可以根据自己的需求不断扩展和优化。最重要的是,你现在有了一个强大的文本生成工具,可以无缝集成到你的工作流中,无论是开发、写作、分析还是自动化任务,都能得心应手。


获取更多AI镜像

想探索更多AI镜像和应用场景?访问 CSDN星图镜像广场,提供丰富的预置镜像,覆盖大模型推理、图像生成、视频生成、模型微调等多个领域,支持一键部署。

Logo

小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地,聚焦技能开发、插件实践与部署教程,为开发者提供可直接落地的方案、工具与交流平台,助力高效构建与落地 AI 应用

更多推荐