GLM-4.7-Flash代码实例:OpenAI兼容API调用与流式响应实战

1. 快速了解GLM-4.7-Flash

GLM-4.7-Flash是智谱AI推出的新一代大语言模型,采用先进的MoE混合专家架构,总参数量达到300亿。这个模型最大的特点就是既强大又高效——MoE架构让它在推理时只激活部分参数,所以响应速度特别快,特别适合需要实时交互的应用场景。

简单来说,GLM-4.7-Flash就像一个超级智能的对话助手,你问它问题,它能快速给出高质量的回答。而且它针对中文场景做了深度优化,无论是理解中文问题还是生成中文内容,表现都很出色。

2. 环境准备与快速部署

2.1 镜像启动与访问

使用GLM-4.7-Flash镜像非常简单,启动后所有服务都会自动运行。你只需要访问Jupyter界面,然后把端口号换成7860,就能看到Web聊天界面了。

访问地址通常是这样的格式:

https://你的服务器地址-7860.web.gpu.csdn.net/

2.2 检查服务状态

打开Web界面后,先看顶部的状态栏:

  • 如果是🟢绿色显示"模型就绪",说明一切正常,可以开始对话了
  • 如果是🟡黄色显示"加载中",说明模型还在加载,等个30秒左右就好了

3. OpenAI兼容API调用实战

3.1 基础API调用

GLM-4.7-Flash提供了完全兼容OpenAI的API接口,这意味着如果你之前用过OpenAI的API,现在可以直接用同样的方式来调用GLM-4.7-Flash。

先来看一个最简单的调用示例:

import requests
import json

# API接口地址
api_url = "http://127.0.0.1:8000/v1/chat/completions"

# 准备请求数据
payload = {
    "model": "/root/.cache/huggingface/ZhipuAI/GLM-4.7-Flash",
    "messages": [
        {"role": "user", "content": "请用中文介绍一下你自己"}
    ],
    "temperature": 0.7,
    "max_tokens": 1024
}

# 发送请求
response = requests.post(api_url, json=payload)

# 解析响应
if response.status_code == 200:
    result = response.json()
    print(result['choices'][0]['message']['content'])
else:
    print(f"请求失败: {response.status_code}")

这段代码会向模型发送一个简单的问候,然后打印出模型的自我介绍。

3.2 多轮对话实现

在实际应用中,我们经常需要实现多轮对话,让模型记住之前的对话上下文。GLM-4.7-Flash支持长上下文记忆,最多可以处理4096个token。

def multi_turn_chat():
    # 初始化对话历史
    conversation_history = []
    
    print("开始与GLM-4.7-Flash对话(输入'退出'结束)")
    
    while True:
        user_input = input("\n你说: ")
        if user_input.lower() == '退出':
            break
            
        # 添加用户消息到历史
        conversation_history.append({"role": "user", "content": user_input})
        
        # 准备请求
        payload = {
            "model": "/root/.cache/huggingface/ZhipuAI/GLM-4.7-Flash",
            "messages": conversation_history,
            "temperature": 0.7,
            "max_tokens": 1024
        }
        
        # 发送请求
        response = requests.post(api_url, json=payload)
        
        if response.status_code == 200:
            result = response.json()
            ai_response = result['choices'][0]['message']['content']
            print(f"AI: {ai_response}")
            
            # 添加AI回复到历史
            conversation_history.append({"role": "assistant", "content": ai_response})
        else:
            print("对话出错,请重试")

这个例子展示了如何实现一个简单的多轮对话程序,模型会记住之前的对话内容,让交流更加连贯自然。

4. 流式响应实战

4.1 什么是流式响应

流式响应是GLM-4.7-Flash的一个很酷的功能。传统的API调用需要等待模型生成完整回答后才能返回结果,而流式响应可以让答案一个字一个字地实时显示出来,就像真人在打字一样,体验特别好。

4.2 实现流式响应

def stream_chat(prompt):
    payload = {
        "model": "/root/.cache/huggingface/ZhipuAI/GLM-4.7-Flash",
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0.7,
        "max_tokens": 1024,
        "stream": True  # 关键参数:启用流式输出
    }
    
    print("AI: ", end="", flush=True)
    
    # 发送流式请求
    with requests.post(api_url, json=payload, stream=True) as response:
        for line in response.iter_lines():
            if line:
                # 解析SSE格式的数据
                if line.startswith(b'data: '):
                    data = line[6:]  # 去掉"data: "前缀
                    if data != b'[DONE]':
                        try:
                            chunk = json.loads(data)
                            content = chunk['choices'][0]['delta'].get('content', '')
                            print(content, end="", flush=True)
                        except:
                            pass
    print()  # 最后换行

# 使用示例
stream_chat("请用中文写一篇关于人工智能未来发展的短文")

4.3 更完善的流式处理

上面的例子比较简单,下面我们写一个更完整的流式对话函数:

import json
import requests

def advanced_stream_chat(messages, temperature=0.7, max_tokens=1024):
    """
    高级流式对话函数
    :param messages: 对话消息列表
    :param temperature: 创造性程度(0-1)
    :param max_tokens: 最大生成长度
    """
    payload = {
        "model": "/root/.cache/huggingface/ZhipuAI/GLM-4.7-Flash",
        "messages": messages,
        "temperature": temperature,
        "max_tokens": max_tokens,
        "stream": True
    }
    
    full_response = ""
    
    try:
        response = requests.post(api_url, json=payload, stream=True, timeout=30)
        response.raise_for_status()
        
        for line in response.iter_lines(decode_unicode=True):
            if line.startswith('data: '):
                data_str = line[6:]
                if data_str != '[DONE]':
                    try:
                        data = json.loads(data_str)
                        if 'choices' in data and len(data['choices']) > 0:
                            delta = data['choices'][0].get('delta', {})
                            if 'content' in delta:
                                content = delta['content']
                                print(content, end="", flush=True)
                                full_response += content
                    except json.JSONDecodeError:
                        continue
        
        print()  # 最后换行
        return full_response
        
    except requests.exceptions.RequestException as e:
        print(f"\n请求出错: {e}")
        return None

# 使用示例
messages = [
    {"role": "user", "content": "请帮我写一首关于春天的诗"}
]
result = advanced_stream_chat(messages)

5. 实际应用案例

5.1 智能客服机器人

下面我们用一个实际的例子来展示如何用GLM-4.7-Flash构建一个简单的智能客服系统:

class CustomerServiceBot:
    def __init__(self):
        self.api_url = "http://127.0.0.1:8000/v1/chat/completions"
        self.system_prompt = """你是一个专业的客服助手,请用友好、专业的态度回答用户问题。
        回答要简洁明了,重点突出。如果遇到不确定的问题,不要编造信息,而是建议用户联系人工客服。"""
        
    def respond_to_customer(self, user_query, conversation_history=[]):
        # 构建消息列表
        messages = [
            {"role": "system", "content": self.system_prompt}
        ]
        
        # 添加历史对话
        messages.extend(conversation_history)
        
        # 添加当前用户问题
        messages.append({"role": "user", "content": user_query})
        
        # 调用API
        payload = {
            "model": "/root/.cache/huggingface/ZhipuAI/GLM-4.7-Flash",
            "messages": messages,
            "temperature": 0.3,  # 温度调低,让回答更稳定
            "max_tokens": 512
        }
        
        try:
            response = requests.post(self.api_url, json=payload, timeout=10)
            if response.status_code == 200:
                result = response.json()
                return result['choices'][0]['message']['content']
            else:
                return "抱歉,系统暂时无法响应,请稍后再试。"
                
        except requests.exceptions.Timeout:
            return "请求超时,请检查网络连接或稍后再试。"
        except Exception as e:
            return f"系统错误: {str(e)}"

# 使用示例
bot = CustomerServiceBot()
response = bot.respond_to_customer("我的订单什么时候能发货?")
print(response)

5.2 内容生成助手

GLM-4.7-Flash在内容创作方面表现特别出色,下面是一个内容生成助手的例子:

def generate_content(content_type, topic, style="专业", length="中等"):
    """
    内容生成助手
    :param content_type: 内容类型(文章、邮件、文案等)
    :param topic: 主题
    :param style: 风格(专业、幽默、正式等)
    :param length: 长度(短篇、中等、长篇)
    """
    
    prompt = f"""请生成一篇{style}风格的{content_type},主题是:{topic}。
    要求:{length}长度,内容要有价值,结构清晰。"""
    
    print(f"正在生成{content_type}...")
    print("=" * 50)
    
    messages = [{"role": "user", "content": prompt}]
    result = advanced_stream_chat(messages, temperature=0.8, max_tokens=2048)
    
    print("=" * 50)
    return result

# 使用示例
# 生成一篇技术博客
article = generate_content("技术博客", "人工智能在医疗领域的应用", "专业", "中等")

# 生成营销文案
ad_copy = generate_content("营销文案", "新款智能手机", "吸引人", "短篇")

6. 错误处理与优化建议

6.1 完善的错误处理

在实际使用中,我们需要处理各种可能出现的错误:

def robust_api_call(messages, max_retries=3):
    """
    健壮的API调用函数,包含重试机制
    """
    for attempt in range(max_retries):
        try:
            payload = {
                "model": "/root/.cache/huggingface/ZhipuAI/GLM-4.7-Flash",
                "messages": messages,
                "temperature": 0.7,
                "max_tokens": 1024
            }
            
            response = requests.post(api_url, json=payload, timeout=15)
            
            if response.status_code == 200:
                return response.json()
            elif response.status_code == 429:
                print("请求过于频繁,稍后重试...")
                time.sleep(2 ** attempt)  # 指数退避
            else:
                print(f"API错误: {response.status_code}")
                time.sleep(1)
                
        except requests.exceptions.Timeout:
            print(f"请求超时,第{attempt + 1}次重试...")
            time.sleep(1)
        except requests.exceptions.ConnectionError:
            print(f"连接错误,检查服务是否启动")
            break
        except Exception as e:
            print(f"未知错误: {str(e)}")
            time.sleep(1)
    
    return None

6.2 性能优化建议

  1. 批量处理请求:如果需要处理大量请求,可以考虑使用批量API
  2. 合理设置超时:根据网络状况设置合适的超时时间
  3. 使用连接池:对于高频调用,使用requests.Session来复用连接
  4. 监控资源使用:定期检查GPU内存使用情况,避免过载
# 使用Session优化性能
session = requests.Session()

def optimized_api_call(messages):
    payload = {
        "model": "/root/.cache/huggingface/ZhipuAI/GLM-4.7-Flash",
        "messages": messages,
        "temperature": 0.7,
        "max_tokens": 1024
    }
    
    try:
        # 使用Session而不是单独的requests.post
        response = session.post(api_url, json=payload, timeout=10)
        return response.json() if response.status_code == 200 else None
    except:
        return None

7. 总结

通过本文的实战示例,你应该已经掌握了如何使用GLM-4.7-Flash的OpenAI兼容API进行各种调用。无论是简单的单次问答,还是复杂的多轮对话,或者是实时的流式响应,GLM-4.7-Flash都能提供出色的性能表现。

关键要点总结:

  • GLM-4.7-Flash提供完全兼容OpenAI的API接口,迁移成本低
  • 流式响应能够显著提升用户体验,实现实时交互效果
  • 多轮对话需要妥善管理对话历史,保持上下文连贯
  • 在实际应用中要添加完善的错误处理和重试机制
  • 合理调整temperature参数可以控制生成内容的创造性程度

GLM-4.7-Flash作为一个300亿参数的大模型,在保持强大能力的同时还能提供快速的响应速度,非常适合各种实时应用场景。无论是智能客服、内容创作、还是知识问答,都能发挥出色的表现。


获取更多AI镜像

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

┌─────────────────────────────────────┐
│     桦漫AIGC集成开发                 │
│     微信: henryhan1117              │
├─────────────────────────────────────┤
│  技术支持 · 定制开发 · 模型部署      │
└─────────────────────────────────────┘
Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐