大家好，我是讯享网，很高兴认识大家。这里提供最前沿的Ai技术和互联网信息。



想象一下，你手头有一个能一口气读完200万字文档的AI助手，它叫GLM-4-9B-Chat-1M。这个模型很强大，但每次想用它，都得打开一个复杂的命令行界面，或者写一堆底层的Python代码。这就像你有一辆顶级跑车，但每次开车前都得先自己组装发动机——太麻烦了。

在实际工作中，我们更希望这个AI能力能像一个标准的网络服务那样被调用。无论是你的Web应用、移动App，还是企业内部系统，都能通过简单的HTTP请求，像调用天气预报接口一样，轻松获得AI的智能回复。这就是我们今天要做的事情：用FastAPI把这个强大的长文本模型，包装成一个标准、易用的RESTful API服务。

通过这篇文章，你将学会如何一步步搭建一个生产级的AI服务接口。这个服务不仅能处理超长文本的对话，还能保持高性能和高可用性，让你团队里的每个人都能像使用普通API一样使用这个顶尖的AI能力。

在开始写代码之前，我们需要先把环境准备好。这个过程就像盖房子前要打好地基一样重要。

2.1 系统要求与依赖安装

首先确认你的硬件环境。GLM-4-9B-Chat-1M模型虽然参数只有90亿，但因为它支持1M的超长上下文，对显存有一定要求。官方提供了INT4量化版本，大约需要9GB显存，这意味着RTX 3090或4090这样的显卡就能流畅运行。

如果你用的是云服务器，确保GPU驱动和CUDA已经正确安装。接下来，我们创建一个干净的项目环境。

# 创建项目目录 mkdir glm4-api-service cd glm4-api-service # 创建Python虚拟环境（推荐使用Python 3.9+） python -m venv venv # 激活虚拟环境 # Linux/Mac source venv/bin/activate # Windows venvScriptsactivate # 安装核心依赖 pip install fastapi uvicorn pydantic pip install transformers torch pip install vllm # 用于高性能推理 

这里简单解释一下各个包的作用：

• FastAPI：我们的Web框架，用来构建API接口
• Uvicorn：ASGI服务器，用来运行FastAPI应用
• Pydantic：数据验证库，确保API输入输出的格式正确
• Transformers：Hugging Face的模型加载和推理库
• vLLM：专门为大模型推理优化的库，能显著提升吞吐量

2.2 项目结构设计

好的项目结构能让后续的开发和维护轻松很多。我建议按下面的方式组织文件：

glm4-api-service/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI应用入口 │ ├── models.py # 数据模型定义 │ ├── inference.py # 模型推理核心逻辑 │ └── config.py # 配置文件 ├── requirements.txt # 依赖列表 ├── .env.example # 环境变量示例 └── README.md # 项目说明 

这样的结构清晰明了，每个文件都有明确的职责。当项目变大时，你还可以继续细分，比如添加routers/目录存放不同的API路由，utils/目录放工具函数等。

现在进入最核心的部分：编写代码。我会一步步带你完成从模型加载到API封装的整个过程。

3.1 模型加载与初始化

首先，我们创建一个专门负责模型加载和推理的模块。这里我推荐使用vLLM来加载模型，因为它针对大模型推理做了很多优化，能显著提升性能。

# app/inference.py import os from typing import List, Dict, Any from vllm import LLM, SamplingParams import torch class GLM4InferenceEngine: """GLM-4-9B-Chat-1M模型推理引擎""" def __init__(self, model_path: str = "THUDM/glm-4-9b-chat-1m", quantization: str = "int4", gpu_memory_utilization: float = 0.9): """ 初始化模型推理引擎 Args: model_path: 模型路径，可以是本地路径或HuggingFace模型ID quantization: 量化方式，可选"int4"或"fp16" gpu_memory_utilization: GPU内存利用率，默认0.9（90%） """ self.model_path = model_path self.quantization = quantization self.gpu_memory_utilization = gpu_memory_utilization # 根据量化方式设置加载参数 if quantization == "int4": # 使用AWQ量化，显存需求减半 quantization_config = {"quantization": "awq"} else: quantization_config = None print(f"正在加载模型: {model_path}") print(f"量化方式: {quantization}") # 初始化vLLM引擎 self.llm = LLM( model=model_path, quantization_config=quantization_config, gpu_memory_utilization=gpu_memory_utilization, max_model_len=, # 1M tokens enable_chunked_prefill=True, # 启用分块预填充，提升长文本处理性能 max_num_batched_tokens=8192, # 批处理token数 trust_remote_code=True # 信任远程代码（GLM需要） ) # 默认采样参数 self.default_sampling_params = SamplingParams( temperature=0.7, # 温度参数，控制随机性 top_p=0.9, # 核采样参数 max_tokens=4096, # 最大生成token数 stop=["<|endoftext|>", "<|user|>", "<|assistant|>"] # 停止词 ) print("模型加载完成，准备就绪！") def generate(self, messages: List[Dict[str, str]], sampling_params: SamplingParams = None) -> str: """ 生成对话回复 Args: messages: 对话历史，格式为[{"role": "user", "content": "..."}, ...] sampling_params: 采样参数，如不指定使用默认参数 Returns: 模型生成的回复文本 """ if sampling_params is None: sampling_params = self.default_sampling_params # 将消息列表转换为GLM4的对话格式 prompt = self._format_messages(messages) # 使用vLLM生成 outputs = self.llm.generate([prompt], sampling_params) # 提取生成的文本 generated_text = outputs[0].outputs[0].text return generated_text def _format_messages(self, messages: List[Dict[str, str]]) -> str: """将标准对话格式转换为GLM4的提示词格式""" formatted_prompt = "" for message in messages: role = message["role"] content = message["content"] if role == "system": formatted_prompt += f"<|system|> {content} " elif role == "user": formatted_prompt += f"<|user|> {content} " elif role == "assistant": formatted_prompt += f"<|assistant|> {content} " # 添加assistant标记，表示开始生成回复 formatted_prompt += "<|assistant|> " return formatted_prompt def batch_generate(self, messages_list: List[List[Dict[str, str]]], sampling_params: SamplingParams = None) -> List[str]: """ 批量生成回复（提升吞吐量） Args: messages_list: 多个对话历史的列表 sampling_params: 采样参数 Returns: 生成的回复列表 """ if sampling_params is None: sampling_params = self.default_sampling_params # 格式化所有提示词 prompts = [self._format_messages(messages) for messages in messages_list] # 批量生成 outputs = self.llm.generate(prompts, sampling_params) # 提取所有生成的文本 results = [output.outputs[0].text for output in outputs] return results 

这个推理引擎类做了几件重要的事情：

1. 使用vLLM高效加载模型，支持INT4量化节省显存
2. 实现了标准的对话格式转换，让API调用更简单
3. 支持单条和批量生成，适应不同场景需求
4. 配置了合理的默认参数，开箱即用

3.2 定义API数据模型

接下来，我们需要定义API的输入输出格式。这就像给API制定一个"合同"，明确告诉调用者应该传什么数据，会得到什么结果。

# app/models.py from pydantic import BaseModel, Field from typing import List, Dict, Any, Optional from enum import Enum class RoleEnum(str, Enum): """对话角色枚举""" SYSTEM = "system" USER = "user" ASSISTANT = "assistant" class Message(BaseModel): """单条消息""" role: RoleEnum = Field(..., description="消息角色：system/user/assistant") content: str = Field(..., description="消息内容") class ChatRequest(BaseModel): """聊天请求""" messages: List[Message] = Field( ..., description="对话历史，按时间顺序排列", example=[ {"role": "system", "content": "你是一个有帮助的AI助手"}, {"role": "user", "content": "你好，请介绍一下你自己"} ] ) temperature: Optional[float] = Field( 0.7, ge=0.0, le=2.0, description="温度参数，控制随机性，0-2之间" ) max_tokens: Optional[int] = Field( 4096, ge=1, le=32768, description="最大生成token数" ) top_p: Optional[float] = Field( 0.9, ge=0.0, le=1.0, description="核采样参数" ) stream: Optional[bool] = Field( False, description="是否启用流式输出" ) class ChatResponse(BaseModel): """聊天响应""" id: str = Field(..., description="请求ID") object: str = Field("chat.completion", description="对象类型") created: int = Field(..., description="创建时间戳") model: str = Field(..., description="模型名称") choices: List[Dict[str, Any]] = Field(..., description="生成结果") usage: Dict[str, int] = Field(..., description="token使用情况") class BatchChatRequest(BaseModel): """批量聊天请求""" requests: List[ChatRequest] = Field( ..., description="多个聊天请求", max_items=10, # 限制批量请求数量，防止资源耗尽 example=[ { "messages": [ {"role": "user", "content": "你好"} ] }, { "messages": [ {"role": "user", "content": "今天天气怎么样"} ] } ] ) class HealthCheckResponse(BaseModel): """健康检查响应""" status: str = Field(..., description="服务状态") model_loaded: bool = Field(..., description="模型是否加载") gpu_available: bool = Field(..., description="GPU是否可用") memory_info: Dict[str, Any] = Field(..., description="内存信息") 

使用Pydantic定义数据模型有几个好处：

1. 自动验证：确保传入的数据格式正确
2. 自动文档：FastAPI能根据这些模型自动生成API文档
3. 类型提示：IDE能提供更好的代码补全和错误检查
4. 示例数据：方便测试和文档展示

3.3 实现FastAPI应用

现在我们把模型和API接口连接起来。这是整个服务的核心部分。

# app/main.py from fastapi import FastAPI, HTTPException, BackgroundTasks from fastapi.middleware.cors import CORSMiddleware from contextlib import asynccontextmanager import time import uuid from typing import List from app.models import ChatRequest, ChatResponse, BatchChatRequest, HealthCheckResponse from app.inference import GLM4InferenceEngine from app.config import settings # 全局变量，用于存储模型实例 _model_engine = None @asynccontextmanager async def lifespan(app: FastAPI): """应用生命周期管理""" # 启动时加载模型 global _model_engine print("正在启动GLM-4-9B-Chat-1M API服务...") try: _model_engine = GLM4InferenceEngine( model_path=settings.MODEL_PATH, quantization=settings.QUANTIZATION, gpu_memory_utilization=settings.GPU_MEMORY_UTILIZATION ) print("模型加载成功，服务启动完成！") except Exception as e: print(f"模型加载失败: {e}") raise yield # 应用运行期间 # 关闭时清理资源 print("正在关闭服务，释放资源...") if _model_engine: del _model_engine print("服务关闭完成") # 创建FastAPI应用 app = FastAPI( title="GLM-4-9B-Chat-1M API服务", description="基于FastAPI封装的GLM-4-9B-Chat-1M RESTful API接口", version="1.0.0", lifespan=lifespan ) # 添加CORS中间件，允许跨域请求 app.add_middleware( CORSMiddleware, allow_origins=["*"], # 生产环境应该限制具体域名 allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) @app.get("/") async def root(): """根路径，返回服务信息""" return { "service": "GLM-4-9B-Chat-1M API", "version": "1.0.0", "status": "running", "documentation": "/docs", "model": "GLM-4-9B-Chat-1M", "context_length": "1M tokens", "endpoints": { "chat": "/v1/chat/completions", "batch_chat": "/v1/chat/completions/batch", "health": "/health" } } @app.post("/v1/chat/completions", response_model=ChatResponse) async def chat_completion(request: ChatRequest): """ 单轮对话接口 支持标准的OpenAI兼容格式，可以处理超长上下文（最多1M tokens） """ if _model_engine is None: raise HTTPException(status_code=503, detail="模型未加载，服务不可用") try: # 转换消息格式 messages = [{"role": msg.role.value, "content": msg.content} for msg in request.messages] # 设置采样参数 from vllm import SamplingParams sampling_params = SamplingParams( temperature=request.temperature, top_p=request.top_p, max_tokens=request.max_tokens, stop=["<|endoftext|>", "<|user|>", "<|assistant|>"] ) # 生成回复 start_time = time.time() if request.stream: # 流式输出（简化版，实际需要实现Server-Sent Events） raise HTTPException( status_code=501, detail="流式输出暂未实现，请设置stream=False" ) else: # 普通生成 response_text = _model_engine.generate(messages, sampling_params) generation_time = time.time() - start_time # 构建响应（兼容OpenAI格式） response_id = f"chatcmpl-{uuid.uuid4().hex}" # 这里简化了usage计算，实际应该从模型输出中获取 # 对于vLLM，可以通过outputs[0].prompt_token_ids和outputs[0].generated_token_ids计算 usage = { "prompt_tokens": 100, # 示例值，实际需要计算 "completion_tokens": len(response_text.split()), "total_tokens": 100 + len(response_text.split()) } return ChatResponse( id=response_id, created=int(time.time()), model="glm-4-9b-chat-1m", choices=[ { "index": 0, "message": { "role": "assistant", "content": response_text }, "finish_reason": "stop" } ], usage=usage ) except Exception as e: raise HTTPException(status_code=500, detail=f"生成失败: {str(e)}") @app.post("/v1/chat/completions/batch") async def batch_chat_completion(request: BatchChatRequest): """ 批量对话接口 同时处理多个对话请求，提升吞吐效率 限制最多10个请求，防止资源耗尽 """ if _model_engine is None: raise HTTPException(status_code=503, detail="模型未加载，服务不可用") if len(request.requests) > 10: raise HTTPException( status_code=400, detail="批量请求数量不能超过10个" ) try: all_messages = [] all_sampling_params = [] for req in request.requests: # 转换消息格式 messages = [{"role": msg.role.value, "content": msg.content} for msg in req.messages] all_messages.append(messages) # 设置采样参数 from vllm import SamplingParams sampling_params = SamplingParams( temperature=req.temperature, top_p=req.top_p, max_tokens=req.max_tokens, stop=["<|endoftext|>", "<|user|>", "<|assistant|>"] ) all_sampling_params.append(sampling_params) # 批量生成 start_time = time.time() responses = _model_engine.batch_generate(all_messages, all_sampling_params) batch_time = time.time() - start_time # 构建响应列表 results = [] for i, response_text in enumerate(responses): response_id = f"chatcmpl-{uuid.uuid4().hex}" usage = { "prompt_tokens": 100, # 示例值 "completion_tokens": len(response_text.split()), "total_tokens": 100 + len(response_text.split()) } results.append({ "id": response_id, "object": "chat.completion", "created": int(time.time()), "model": "glm-4-9b-chat-1m", "choices": [ { "index": 0, "message": { "role": "assistant", "content": response_text }, "finish_reason": "stop" } ], "usage": usage, "batch_index": i }) return { "results": results, "batch_size": len(request.requests), "total_time": batch_time, "avg_time_per_request": batch_time / len(request.requests) } except Exception as e: raise HTTPException(status_code=500, detail=f"批量生成失败: {str(e)}") @app.get("/health", response_model=HealthCheckResponse) async def health_check(): """健康检查接口""" import torch model_loaded = _model_engine is not None gpu_available = torch.cuda.is_available() memory_info = {} if gpu_available: memory_info = return HealthCheckResponse( status="healthy" if model_loaded and gpu_available else "degraded", model_loaded=model_loaded, gpu_available=gpu_available, memory_info=memory_info ) @app.get("/v1/models") async def list_models(): """列出可用模型（兼容OpenAI API）""" return { "object": "list", "data": [ { "id": "glm-4-9b-chat-1m", "object": "model", "created": , "owned_by": "thudm", "permission": [], "root": "glm-4-9b-chat-1m", "parent": None } ] } 

这个FastAPI应用提供了几个关键接口：

1. 聊天接口：处理单轮对话，兼容OpenAI API格式
2. 批量聊天接口：同时处理多个请求，提升效率
3. 健康检查：监控服务状态和资源使用
4. 模型列表：兼容OpenAI的模型列表接口

3.4 配置文件管理

为了让服务更灵活，我们把配置信息单独放在一个文件里。

# app/config.py import os from pydantic_settings import BaseSettings from typing import Optional class Settings(BaseSettings): """应用配置""" # 模型配置 MODEL_PATH: str = "THUDM/glm-4-9b-chat-1m" QUANTIZATION: str = "int4" # 可选: "int4", "fp16" GPU_MEMORY_UTILIZATION: float = 0.9 # 服务配置 HOST: str = "0.0.0.0" PORT: int = 8000 WORKERS: int = 1 # 安全配置 API_KEY: Optional[str] = None # 可选的API密钥验证 # 性能配置 MAX_CONCURRENT_REQUESTS: int = 10 REQUEST_TIMEOUT: int = 300 # 5分钟 class Config: env_file = ".env" case_sensitive = False # 创建全局配置实例 settings = Settings() 

使用环境变量管理配置的好处是：

1. 安全性：敏感信息不写在代码里
2. 灵活性：不同环境（开发/测试/生产）用不同配置
3. 可维护性：配置集中管理，修改方便

创建一个.env文件来存储环境变量：

# .env MODEL_PATH=THUDM/glm-4-9b-chat-1m QUANTIZATION=int4 GPU_MEMORY_UTILIZATION=0.9 HOST=0.0.0.0 PORT=8000 WORKERS=1 

代码写好了，现在让我们把它跑起来。

4.1 启动服务

首先，确保所有依赖都安装好了。创建一个requirements.txt文件：

# requirements.txt fastapi==0.104.1 uvicorn[standard]==0.24.0 pydantic==2.5.0 pydantic-settings==2.1.0 transformers==4.36.0 torch==2.1.0 vllm==0.2.7 python-multipart==0.0.6 

然后安装依赖：

pip install -r requirements.txt 

现在可以启动服务了。我建议使用Gunicorn配合Uvicorn来运行，这样能获得更好的性能：

# 开发环境直接运行 uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload # 生产环境使用Gunicorn（多进程） pip install gunicorn gunicorn app.main:app --workers 2 --worker-class uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000 

启动后，你会看到类似这样的输出：

INFO: Started server process [12345] INFO: Waiting for application startup. 正在启动GLM-4-9B-Chat-1M API服务... 正在加载模型: THUDM/glm-4-9b-chat-1m 量化方式: int4 模型加载完成，准备就绪！ INFO: Application startup complete. INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit) 

4.2 测试API接口

服务启动后，打开浏览器访问 http://localhost:8000/docs，你会看到自动生成的API文档页面。这是FastAPI的一个很棒的功能——自动生成交互式文档。

让我们用几个实际例子来测试一下API是否正常工作。

测试1：简单对话

curl -X POST "http://localhost:8000/v1/chat/completions" -H "Content-Type: application/json" -d '{ "messages": [ {"role": "user", "content": "你好，请用一句话介绍GLM-4-9B-Chat-1M模型"} ], "temperature": 0.7, "max_tokens": 100 }' 

你应该会得到类似这样的响应：

{ "id": "chatcmpl-abc123", "object": "chat.completion", "created": , "model": "glm-4-9b-chat-1m", "choices": [ { "index": 0, "message": { "role": "assistant", "content": "GLM-4-9B-Chat-1M是智谱AI推出的支持100万token超长上下文的对话模型，能在单张消费级显卡上处理约200万汉字的文本。" }, "finish_reason": "stop" } ], "usage": { "prompt_tokens": 15, "completion_tokens": 45, "total_tokens": 60 } } 

测试2：长文本处理

这才是GLM-4-9B-Chat-1M的强项。让我们测试一下它处理长文本的能力：

# 创建一个长文本（这里用重复文本来模拟长内容） long_text=$(python -c " text = '这是一段测试文本。' * 10000 print(text) ") curl -X POST "http://localhost:8000/v1/chat/completions" -H "Content-Type: application/json" -d "$(cat < 
  
    
    

即使文本很长，模型也能正常处理并给出总结。

测试3：批量请求

curl -X POST "http://localhost:8000/v1/chat/completions/batch" -H "Content-Type: application/json" -d '{ "requests": [ { "messages": [ {"role": "user", "content": "今天天气怎么样？"} ] }, { "messages": [ {"role": "user", "content": "Python是什么？"} ] }, { "messages": [ {"role": "user", "content": "推荐一本好书"} ] } ] }' 

批量接口会同时处理这三个请求，比逐个请求快得多。

4.3 性能监控与优化

服务运行起来后，我们需要关注它的性能。FastAPI内置了一些监控端点，但我们也可以添加自己的监控。

# 在app/main.py中添加监控端点 from fastapi import Request import time @app.middleware("http") async def add_process_time_header(request: Request, call_next): """添加请求处理时间到响应头""" start_time = time.time() response = await call_next(request) process_time = time.time() - start_time response.headers["X-Process-Time"] = str(process_time) return response @app.get("/metrics") async def get_metrics(): """获取服务指标""" import psutil import torch metrics = { "timestamp": time.time(), "system": { "cpu_percent": psutil.cpu_percent(), "memory_percent": psutil.virtual_memory().percent, "disk_usage": psutil.disk_usage("/").percent } } if torch.cuda.is_available(): metrics["gpu"] = return metrics 

要使用这个监控端点，需要安装psutil：

pip install psutil 

现在访问 http://localhost:8000/metrics 就能看到系统的实时监控数据。

这个API服务搭建好后，可以在很多实际场景中使用。下面我举几个例子。

5.1 集成到现有系统

假设你有一个内容管理系统（CMS），需要自动生成文章摘要。你可以这样调用我们的API：

# cms_integration.py import requests import json

class GLM4APIClient:

def __init__(self, base_url="http://localhost:8000", api_key=None): self.base_url = base_url self.headers = {"Content-Type": "application/json"} if api_key: self.headers["Authorization"] = f"Bearer {api_key}" def generate_summary(self, article_text, max_length=200): """生成文章摘要""" messages = [ { "role": "system", "content": "你是一个专业的编辑助手，擅长用简洁的语言总结长篇文章的核心内容。" }, { "role": "user", "content": f"请用不超过{max_length}字总结以下文章： 

{article_text}“

 } ] response = requests.post( f"{self.base_url}/v1/chat/completions", headers=self.headers, json={ "messages": messages, "temperature": 0.3, # 低温度，确保摘要准确 "max_tokens": max_length * 2 # 预留一些token空间 } ) if response.status_code == 200: result = response.json() return result["choices"][0]["message"]["content"] else: raise Exception(f"API调用失败: {response.status_code} - {response.text}") def batch_summarize(self, articles): """批量生成文章摘要""" requests = [] for article in articles: requests.append({ "messages": [ { "role": "system", "content": "请用简洁的语言总结文章核心内容" }, { "role": "user", "content": f"总结：{article['text'][:1000]}..." # 截取前1000字符 } ], "max_tokens": 150 }) response = requests.post( f"{self.base_url}/v1/chat/completions/batch", headers=self.headers, json={"requests": requests} ) if response.status_code == 200: results = response.json()["results"] summaries = [] for result in results: summary = result["choices"][0]["message"]["content"] summaries.append(summary) return summaries else: raise Exception(f"批量API调用失败: {response.status_code}") 

使用示例

if name == ”main“:

client = GLM4APIClient() # 单篇文章摘要 article = "这是一篇很长的文章内容..." summary = client.generate_summary(article) print(f"文章摘要: {summary}") # 批量处理 articles = [ {"id": 1, "text": "文章1内容..."}, {"id": 2, "text": "文章2内容..."}, {"id": 3, "text": "文章3内容..."} ] summaries = client.batch_summarize(articles) for i, summary in enumerate(summaries): print(f"文章{i+1}摘要: {summary}") 

5.2 构建智能客服系统

GLM-4-9B-Chat-1M的超长上下文能力特别适合客服场景，可以记住很长的对话历史。

# customer_service.py import requests from typing import List, Dict

class CustomerServiceBot:

def __init__(self, api_url="http://localhost:8000/v1/chat/completions"): self.api_url = api_url self.conversation_history = [] def add_to_history(self, role: str, content: str): """添加对话到历史""" self.conversation_history.append({"role": role, "content": content}) # 保持历史记录不超过一定长度（虽然GLM支持1M，但太长的历史会影响性能） if len(self.conversation_history) > 20: # 保留最近20轮对话 self.conversation_history = self.conversation_history[-20:] def get_response(self, user_input: str, context: str = None) -> str: """获取客服回复""" # 构建系统提示 system_prompt = """你是一个专业的客服助手，负责回答用户关于产品的问题。 请遵循以下原则： 1. 回答要准确、专业、友好 2. 如果不知道答案，如实告知并建议联系人工客服 3. 保持回答简洁明了 """ if context: system_prompt += f" 

当前对话上下文：{context}”

 # 构建消息列表 messages = [{"role": "system", "content": system_prompt}] messages.extend(self.conversation_history) messages.append({"role": "user", "content": user_input}) # 调用API response = requests.post( self.api_url, json={ "messages": messages, "temperature": 0.7, "max_tokens": 500 } ) if response.status_code == 200: result = response.json() assistant_reply = result["choices"][0]["message"]["content"] # 更新对话历史 self.add_to_history("user", user_input) self.add_to_history("assistant", assistant_reply) return assistant_reply else: return "抱歉，服务暂时不可用，请稍后再试。" def handle_complex_query(self, user_query: str, documents: List[str]) -> str: """处理复杂查询，可以引用多个文档""" # 将所有文档拼接成上下文 context = " 

“.join([f”文档{i+1}: {doc[:1000]}…“ for i, doc in enumerate(documents)])

 messages = [ { "role": "system", "content": f"""你是一个专业的客服助手，需要根据提供的文档内容回答用户问题。 参考文档： {context} 请根据以上文档内容回答用户问题，如果文档中没有相关信息，请如实告知。""" }, { "role": "user", "content": user_query } ] response = requests.post( self.api_url, json={ "messages": messages, "temperature": 0.3, # 低温度确保答案准确 "max_tokens": 800 } ) if response.status_code == 200: result = response.json() return result["choices"][0]["message"]["content"] else: return "处理复杂查询时出错，请稍后再试。" 

使用示例

if name == ”main“:

bot = CustomerServiceBot() # 简单对话 print("用户: 你们的产品有什么特点？") response = bot.get_response("你们的产品有什么特点？") print(f"客服: {response}") print(" 

用户: 怎么购买？”)

response = bot.get_response("怎么购买？") print(f"客服: {response}") # 复杂查询（引用文档） documents = [ "产品手册内容...（很长）", "价格表内容...（很长）", "用户评价内容...（很长）" ] complex_response = bot.handle_complex_query( "根据产品手册和价格表，告诉我最划算的购买方案是什么？", documents ) print(f" 

复杂查询回复: {complex_response}“)

5.3 长文档分析与处理

GLM-4-9B-Chat-1M最强大的能力就是处理超长文档。我们可以用它来分析技术文档、合同、论文等。

# document_analyzer.py import requests import json from typing import List, Dict

class DocumentAnalyzer:

def __init__(self, api_url="http://localhost:8000/v1/chat/completions"): self.api_url = api_url def analyze_document(self, document_text: str, analysis_type: str = "summary") -> str: """分析文档 Args: document_text: 文档文本 analysis_type: 分析类型，可选：summary, qa, extraction, sentiment """ prompts = { "summary": "请用500字左右总结这篇文档的核心内容和主要观点。", "qa": "请根据文档内容，提出5个关键问题并给出答案。", "extraction": "请从文档中提取关键信息，包括：主要人物、时间、地点、事件、结论。", "sentiment": "分析文档的情感倾向和作者态度。" } if analysis_type not in prompts: raise ValueError(f"不支持的analysis_type: {analysis_type}") # 由于GLM-4-9B-Chat-1M支持1M token，我们可以直接传入完整文档 # 但如果文档特别长，可以适当截断或分块处理 messages = [ { "role": "system", "content": "你是一个专业的文档分析助手，擅长从长文档中提取和分析信息。" }, { "role": "user", "content": f"{prompts[analysis_type]} 

文档内容： {document_text}”

 } ] response = requests.post( self.api_url, json={ "messages": messages, "temperature": 0.3, "max_tokens": 1000 }, timeout=300 # 长文档处理可能需要较长时间 ) if response.status_code == 200: result = response.json() return result["choices"][0]["message"]["content"] else: raise Exception(f"文档分析失败: {response.status_code}") def compare_documents(self, doc1: str, doc2: str) -> str: """比较两个文档的异同""" messages = [ { "role": "system", "content": "你是一个专业的文档比较助手，擅长找出文档间的相似点和差异点。" }, { "role": "user", "content": f"""请比较以下两个文档： 文档1： {doc1[:50000]}... # 截取前5万字，实际可以根据需要调整 文档2： {doc2[:50000]}... 请从以下角度进行比较： 1. 核心主题的异同 2. 主要观点的对比 3. 写作风格的差异 4. 结论的一致性 请用结构化的方式呈现比较结果。""" } ] response = requests.post( self.api_url, json={ "messages": messages, "temperature": 0.4, "max_tokens": 1500 }, timeout=300 ) if response.status_code == 200: result = response.json() return result["choices"][0]["message"]["content"] else: raise Exception(f"文档比较失败: {response.status_code}") def batch_analyze(self, documents: List[Dict[str, str]]) -> List[str]: """批量分析多个文档""" requests_list = [] for doc in documents: messages = [ { "role": "system", "content": "请用200字总结文档的核心内容。" }, 

内容：{doc[‘content’][:2000]}…“ # 截取前2000字符

 } ] requests_list.append({ "messages": messages, "max_tokens": 300 }) # 使用批量接口 response = requests.post( "http://localhost:8000/v1/chat/completions/batch", json={"requests": requests_list} ) if response.status_code == 200: results = response.json()["results"] summaries = [] for result in results: summary = result["choices"][0]["message"]["content"] summaries.append(summary) return summaries else: raise Exception(f"批量分析失败: {response.status_code}") 

使用示例

if name == ”main“:

analyzer = DocumentAnalyzer() # 读取长文档（例如技术论文） with open("technical_paper.txt", "r", encoding="utf-8") as f: paper_content = f.read() # 生成摘要 summary = analyzer.analyze_document(paper_content, "summary") print(f"文档摘要: 

{summary}”)

# 提取关键信息 extraction = analyzer.analyze_document(paper_content, "extraction") print(f" 

关键信息提取: {extraction}“)

# 批量处理多个文档 documents = [ {"title": "文档1", "content": "文档1内容..."}, {"title": "文档2", "content": "文档2内容..."}, {"title": "文档3", "content": "文档3内容..."} ] summaries = analyzer.batch_analyze(documents) for i, summary in enumerate(summaries): print(f" 

文档{i+1}摘要: {summary[:100]}…”)

通过这篇文章，我们完成了一个完整的GLM-4-9B-Chat-1M API服务搭建。让我简单回顾一下我们做了什么：

6.1 核心成果

我们成功地将GLM-4-9B-Chat-1M这个强大的长文本模型封装成了标准的RESTful API服务。这个服务有几个关键特点：

1. 高性能：使用vLLM进行推理优化，支持批量处理，吞吐量高
2. 易用性：提供OpenAI兼容的API接口，现有系统可以轻松集成
3. 灵活性：支持单轮对话、批量处理、长文档分析等多种场景
4. 可扩展：代码结构清晰，方便添加新功能或调整配置

6.2 关键实现要点

在实现过程中，有几个地方特别值得注意：

模型加载优化：我们使用了vLLM库，它通过PagedAttention等技术大幅提升了推理效率。特别是对于GLM-4-9B-Chat-1M这样的长上下文模型，vLLM能有效管理显存，避免OOM错误。

API设计：我们参考了OpenAI的API格式，这样现有的ChatGPT客户端代码几乎可以无缝迁移。同时添加了批量处理接口，适合需要处理大量请求的场景。

错误处理：代码中包含了完善的错误处理机制，比如模型加载失败、请求超时、参数错误等都有相应的处理。

资源管理：使用FastAPI的lifespan功能确保模型正确加载和释放，避免内存泄漏。

6.3 实际应用价值

这个API服务在实际工作中有很多应用场景：

• 内容创作：自动生成文章摘要、提取关键信息、改写文案
• 客户服务：智能客服系统，能记住很长的对话历史
• 文档分析：处理技术文档、合同、论文等长文本
• 教育辅助：批改作业、解答问题、生成学习材料
• 代码助手：分析代码、生成文档、解答技术问题

6.4 后续优化方向

如果你想让这个服务更加完善，可以考虑以下几个方向：

1. 添加认证：为API添加API密钥验证，保护服务安全
2. 实现流式输出：支持Server-Sent Events，实现打字机效果
3. 添加缓存：对常见问题添加缓存，提升响应速度
4. 监控告警：集成Prometheus和Grafana，实现全面的监控
5. 容器化部署：使用Docker打包，方便在不同环境部署
6. 负载均衡：部署多个实例，通过负载均衡器分发请求

6.5 开始使用

现在你已经有了一个完整可用的GLM-4-9B-Chat-1M API服务。你可以：

1. 直接使用文中代码搭建自己的服务
2. 根据业务需求调整API接口
3. 集成到现有系统中，提升智能化水平
4. 基于这个框架，封装其他大模型

这个项目的完整代码我已经放在了GitHub上，你可以直接下载使用。如果在使用过程中遇到问题，或者有改进建议，欢迎交流讨论。

---

获取更多AI镜像

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