从RAG到Agent，大模型推理框架SGLang如何让LLM应用快10倍？
在当今大语言模型（LLM）应用蓬勃发展的时代，开发者们面临着前所未有的挑战：如何高效地构建复杂的LLM应用？如何让模型调用更加快速、稳定且易于管理？当传统的LangChain方案显得过于笨重时，一个新兴的框架正在悄然改变游戏规则。
今天要介绍的项目是 SGLang（Structured Generation Language），这是一个由SGL项目组开发的创新性框架，专门用于简化大语言模型的推理调用和复杂应用构建。如果你正在寻找一个既能提升开发效率，又能显著优化推理性能的解决方案，那么这篇文章将带你深入了解SGLang的一切。
为什么值得关注：SGLang的核心价值
在深入学习SGLang之前，我们需要理解为什么这个项目值得你投入时间。SGLang的出现并非偶然，而是针对当前LLM应用开发中的真实痛点。
传统开发方式的困境
大多数开发者在构建LLM应用时，都会遇到类似的挑战。首先是调用方式繁琐的问题：使用原始API调用需要处理大量的重复性代码，包括错误处理、并发控制、响应解析等。其次是性能瓶颈——单个请求的串行处理无法充分利用GPU的并行计算能力，导致资源浪费和响应延迟。最后，复杂的多步骤任务处理往往需要编写大量的胶水代码，使得项目变得难以维护和扩展。
LangChain和LlamaIndex等框架试图解决这些问题，但它们往往引入了过多的抽象层，导致学习曲线陡峭，且运行时开销不容忽视。开发者们迫切需要一个更轻量、更高效、更直观的解决方案。
SGLang的设计哲学
SGLang的设计理念可以用三个关键词概括：简洁、性能、结构化。它没有试图成为另一个“全能框架”，而是专注于做好一件事——让LLM调用变得更快、更简单、更可控。
与传统的链式调用方式不同，SGLang引入了结构化生成的概念。它允许开发者以声明式的方式定义prompt模板、输出约束和调用流程，从而大幅简化复杂应用的开发。这种设计不仅让代码更加清晰易读，还能让框架在运行时进行深度优化，实现显著的性能提升。
在实际测试中，SGLang相比传统调用方式能够实现3到10倍的性能提升，这一数字对于需要处理大量请求的生产环境来说意义重大。无论是构建RAG系统、构建Agent应用，还是开发复杂的对话系统，SGLang都能提供坚实的技术支撑。
环境搭建：快速启动SGLang
现在开始动手实践。首先需要准备好开发环境，SGLang对系统要求相对友好，但有一些依赖需要注意。
系统要求与前置条件
SGLang基于Python开发，官方推荐使用Python 3.9或更高版本。在开始之前，请确保你的系统中已经安装了以下软件：CUDA工具包（推荐11.8或更高版本以获得最佳性能）、PyTorch（2.0及以上版本）。如果你计划使用GPU进行推理，还需要一块支持CUDA的显卡。
对于深度学习爱好者来说，使用conda或venv创建独立的Python环境是一个好习惯，这可以避免依赖冲突问题。
安装SGLang
安装SGLang有多种方式，推荐使用pip进行快速安装：
pip install sglang
如果你需要安装包含所有可选依赖的完整版本，可以使用：
pip install "sglang[all]"
对于想要体验最新功能的开发者，也可以直接从源代码安装：
git clone https://github.com/sgl-project/sglang.git
cd sglang
pip install -e .
安装完成后，可以通过以下命令验证安装是否成功：
import sglang
print("SGLang版本:", sglang.__version__)
# 输出类似: SGLang版本: 0.1.0
后端引擎配置
SGLang支持多种后端引擎，其中与vLLM的集成是最常见的选择。vLLN是一个高性能的LLM推理引擎，能够显著提升吞吐量。以下是安装vLLM的步骤：
pip install vllm
安装完vLLM后，SGLang会自动检测并使用它作为默认后端。如果你想使用其他后端，可以查阅官方文档了解详细的配置方法。
快速验证示例
让我们通过一个简单的示例来验证整个环境是否正常工作：
import sglang as sg
# 创建SGLang运行时
runtime = sg.Runtime(model_path="microsoft/Phi-3-mini-128k-instruct")
# 定义一个简单的prompt
prompt = "请用一句话介绍人工智能的发展历史。"
# 执行推理
result = runtime.generate(prompt, max_tokens=200)
print("生成的文本:", result["text"])
如果这个示例能够正常运行并输出结果，说明你的环境已经准备就绪。接下来我们将深入学习SGLang的核心功能。
核心功能详解：SGLang的设计精髓
SGLang的核心功能围绕三个主要方面展开：结构化提示（Structured Prompting）、链式调用（Chained Generation）和约束解码（Constrained Decoding）。理解这三个概念是掌握SGLang的关键。
结构化提示：让Prompt工程变得优雅
在传统的LLM开发中，prompt往往以字符串形式存在，这种方式在简单场景下还算够用，但一旦prompt变得复杂——比如需要包含变量、条件判断、循环结构——代码就会变得难以维护。SGLang引入了结构化prompt的概念，允许开发者以更加直观的方式定义prompt模板。
考虑这样一个场景：我们需要根据用户的不同问题类型生成不同的回复。在传统方式下，你可能会这样写：
def generate_response(user_type, user_question):
if user_type == "technical":
prompt = f"作为技术专家，请回答以下问题：{user_question}"
elif user_type == "business":
prompt = f"作为商业顾问，请分析这个问题：{user_question}"
else:
prompt = f"请回答：{user_question}"
response = model.generate(prompt)
return response
这种方式虽然能工作，但随着条件增多，代码会变得臃肿。SGLang提供了更加优雅的解决方案：
import sglang as sg
# 使用SGLang的结构化prompt
@sglang.function
def qa_system(user_type: str, user_question: str) -> str:
# 通过结构化的方式定义prompt逻辑
if user_type == "technical":
return sg.user(f"作为技术专家，请详细解答：{user_question}")
elif user_type == "business":
return sg.user(f"作为商业顾问，请提供专业分析：{user_question}")
else:
return sg.user(user_question)
return sg.assistant(sg.gen("answer", max_tokens=500))
这种声明式的写法不仅让代码更加清晰，还能让SGLang在运行时进行优化，减少不必要的token处理。
链式调用：构建复杂的工作流
SGLang的另一个核心特性是支持链式调用。在实际应用中，很少有任务只需要一次LLM调用就能完成。常见的场景包括：先让模型分析问题，再根据分析结果查询相关信息，最后生成最终答案。
传统的实现方式会将这些步骤串联起来，每个步骤都需要独立的API调用，导致延迟累加。SGLang允许你将这些步骤融合在一个prompt图中，实现更高效的批处理。
以下是一个RAG（检索增强生成）系统的示例：
import sglang as sg
@sglang.function
def rag_system(question: str, context_docs: list[str]):
# 步骤一：理解问题并提取关键实体
understand_prompt = sg.user(
f"问题：{question}\n请提取问题中的关键实体和意图。"
)
understanding = sg.assistant(sg.gen("understanding", max_tokens=100))
# 步骤二：根据理解的问题筛选相关文档
relevant_docs = []
for doc in context_docs:
relevance_check = sg.assistant(
sg.gen(f"relevance_{doc[:10]}", max_tokens=20)
)
# 这里简化了相关性判断逻辑
relevant_docs.append(doc)
# 步骤三：基于相关文档生成答案
answer_prompt = sg.user(
f"基于以下文档回答问题：\n" + "\n".join(relevant_docs) +
f"\n问题：{question}"
)
final_answer = sg.assistant(sg.gen("answer", max_tokens=300))
return final_answer
在上面的代码中，sg.gen()函数是SGLang的核心语法之一，它告诉框架需要生成一个标记为特定名称的文本块。这些生成操作可以被并行优化，从而显著提升整体性能。
约束解码：精准控制输出格式
在构建需要结构化输出的应用时（如JSON、代码等），约束解码是一个不可或缺的特性。SGLang提供了强大的约束解码功能，让你能精确控制模型的输出格式和内容。
import sglang as sg
from pydantic import BaseModel
class WeatherResponse(BaseModel):
city: str
temperature: float
condition: str
humidity: int
recommendation: str
@sglang.function
def weather_agent(user_query: str):
prompt = sg.user(
f"用户查询：{user_query}\n"
"请以上述格式返回天气信息。"
)
# 使用constrained参数指定输出约束
result = sg.assistant(
sg.gen(
"weather_info",
max_tokens=300,
format=WeatherResponse # 指定输出格式
)
)
return result
SGLang支持多种约束方式，包括正则表达式约束、JSON Schema约束、上下文无关文法约束等。这种灵活性让你能够构建几乎任何需要结构化输出的应用。
实战教程：从入门到精通
现在我们已经了解了SGLang的核心概念，接下来通过一系列实战案例来加深理解。这些案例涵盖从简单到复杂的各种场景，每个案例都提供了完整的代码示例。
案例一：构建一个智能问答助手
让我们从最简单的场景开始——构建一个能够回答各类问题的智能助手。这个案例将帮助你熟悉SGLang的基本用法。
import sglang as sg
# 初始化SGLang运行时
runtime = sg.Runtime(
model_name="Qwen/Qwen2-7B-Instruct",
tensor_parallel_size=1, # GPU数量
dtype="half", # 使用半精度浮点数节省显存
)
# 定义问答函数
@runtime.function
def qa_assistant(question: str, domain: str = "general"):
"""
智能问答助手的主函数
参数:
question: 用户的问题
domain: 问题领域（general/tech/business）
"""
# 根据领域选择系统提示
if domain == "tech":
system_prompt = "你是一位经验丰富的技术专家，回答要专业、精准、深入。"
elif domain == "business":
system_prompt = "你是一位资深的商业顾问，回答要注重实操性和商业价值。"
else:
system_prompt = "你是一位博学的助手，回答要全面、易懂、有帮助。"
# 构建完整的prompt
full_prompt = sg.user(f"{system_prompt}\n\n问题：{question}")
# 生成回答
answer = sg.assistant(sg.gen("response", max_tokens=500))
return answer
# 测试问答助手
if __name__ == "__main__":
# 简单测试
result = qa_assistant(
question="什么是大语言模型？",
domain="tech"
)
print("=" * 60)
print("问答助手测试结果")
print("=" * 60)
print(f"问题：什么是大语言模型？")
print(f"领域：技术")
print(f"回答：{result['response']}")
print("=" * 60)
运行这个脚本后，你应该能看到模型生成的回答。尝试修改问题类型和领域参数，观察输出风格的变化。
案例二：实现多文档摘要系统
接下来我们实现一个更复杂的系统——多文档摘要。这个案例将展示如何使用SGLang处理多个输入源，并进行批处理优化。
import sglang as sg
from typing import List, Dict
class MultiDocSummarizer:
"""
多文档摘要生成器
这个类封装了文档摘要的完整逻辑，包括：
- 文档预处理
- 分块处理
- 摘要生成
- 结果整合
"""
def __init__(self, model_name: str = "Qwen/Qwen2-7B-Instruct"):
self.runtime = sg.Runtime(model_name=model_name)
self.max_chunk_tokens = 2000 # 每个文档块的最大token数
self.summary_tokens = 300 # 摘要的最大token数
def chunk_document(self, text: str) -> List[str]:
"""将长文档分割成小块"""
# 简单按段落分割，实际应用中可能需要更复杂的策略
paragraphs = text.split("\n\n")
chunks = []
current_chunk = ""
for para in paragraphs:
if len(current_chunk) + len(para) < self.max_chunk_tokens:
current_chunk += para + "\n\n"
else:
if current_chunk:
chunks.append(current_chunk.strip())
current_chunk = para + "\n\n"
if current_chunk:
chunks.append(current_chunk.strip())
return chunks
def summarize_chunk(self, chunk: str, doc_title: str) -> str:
"""对单个文档块生成摘要"""
@self.runtime.function
def chunk_summary(text: str, title: str):
prompt = sg.user(
f"请为以下文档片段生成简洁的摘要。\n\n"
f"文档标题：{title}\n"
f"内容：{text}\n\n"
f"摘要要求："
f"1. 提取核心观点"
f"2. 保留关键数据和结论"
f"3. 长度控制在100字以内"
)
return sg.assistant(sg.gen("summary", max_tokens=150))
result = chunk_summary(chunk, doc_title)
return result["summary"]
def generate_final_summary(self, partial_summaries: List[str]) -> str:
"""从多个部分摘要生成最终摘要"""
@self.runtime.function
def final_summary(summaries: List[str]):
combined = "\n---\n".join(summaries)
prompt = sg.user(
f"以下是文档各部分的摘要：\n{combined}\n\n"
f"请整合这些摘要，生成一篇连贯、完整的最终摘要，"
f"突出文档的核心内容和价值。"
)
return sg.assistant(sg.gen("final", max_tokens=500))
result = final_summary(partial_summaries)
return result["final"]
def summarize(self, documents: Dict[str, str]) -> str:
"""
对多篇文档生成综合摘要
参数:
documents: 字典，键为文档标题，值为文档内容
返回:
str: 综合摘要文本
"""
all_partial_summaries = []
print("开始处理文档...")
for idx, (title, content) in enumerate(documents.items(), 1):
print(f" 处理文档 {idx}/{len(documents)}: {title}")
# 分块
chunks = self.chunk_document(content)
print(f" 分割为 {len(chunks)} 个块")
# 各块摘要
chunk_summaries = []
for chunk_idx, chunk in enumerate(chunks, 1):
print(f" 生成块 {chunk_idx} 的摘要...")
summary = self.summarize_chunk(chunk, title)
chunk_summaries.append(summary)
# 合并该文档的摘要
doc_summary = " ".join(chunk_summaries)
all_partial_summaries.append(f"【{title}】{doc_summary}")
print("生成最终综合摘要...")
final = self.generate_final_summary(all_partial_summaries)
return final
# 使用示例
if __name__ == "__main__":
# 准备测试数据
sample_docs = {
"AI发展趋势报告": """
人工智能技术在2024年继续保持快速发展态势。根据最新研究数据，
全球AI市场规模预计将在2025年突破5000亿美元。在技术层面，
大语言模型、多模态学习、强化学习等领域取得了重大突破。
企业应用方面，AI正在从辅助工具向核心业务组件转型，
预计到2026年，超过80%的大型企业将采用AI解决方案。
""",
"机器学习技术白皮书": """
机器学习作为AI的核心技术，近年来发展迅速。深度学习模型
的规模持续增长，从最初的百万参数发展到现在的万亿参数。
新的训练方法如提示学习、小样本学习等降低了AI应用门槛。
在应用层面，计算机视觉、自然语言处理、推荐系统等领域
的算法性能显著提升，已在多个场景实现商用落地。
边缘计算与机器学习的结合也成为重要趋势。
"""
}
# 创建摘要器实例
summarizer = MultiDocSummarizer()
# 执行摘要
result = summarizer.summarize(sample_docs)
print("\n" + "=" * 60)
print("多文档摘要结果")
print("=" * 60)
print(result)
print("=" * 60)
这个案例展示了SGLang在实际生产环境中的应用方式。通过将复杂的处理流程分解为可管理的步骤，并利用SGLang的优化能力，我们可以构建高效的多文档处理系统。
案例三：构建具备工具调用能力的Agent
SGLang的另一个强大功能是支持工具调用（Function Calling）。这使得构建能够使用外部工具的Agent成为可能。接下来的案例将展示如何构建一个具备搜索、计算能力的智能Agent。
import sglang as sg
import json
from typing import Any, Dict, List, Optional
from datetime import datetime
class Tool:
"""工具基类"""
name: str = ""
description: str = ""
parameters: Dict[str, Any] = {}
def execute(self, **kwargs) -> str:
raise NotImplementedError
class CalculatorTool(Tool):
"""计算器工具"""
name = "calculator"
description = "执行数学计算，支持加减乘除和复杂表达式"
parameters = {
"expression": {
"type": "string",
"description": "要计算的数学表达式，如 '2 + 3 * 4'"
}
}
def execute(self, expression: str) -> str:
try:
# 安全评估数学表达式
allowed_ops = {'+': lambda a, b: a + b, '-': lambda a, b: a - b,
'*': lambda a, b: a * b, '/': lambda a, b: a / b,
'**': lambda a, b: a ** b}
result = eval(expression, {"__builtins__": {}}, allowed_ops)
return f"计算结果：{expression} = {result}"
except Exception as e:
return f"计算错误：{str(e)}"
class DateTimeTool(Tool):
"""日期时间工具"""
name = "get_datetime"
description = "获取当前日期时间信息"
parameters = {}
def execute(self) -> str:
now = datetime.now()
return f"当前时间：{now.strftime('%Y年%m月%d日 %H:%M:%S')}"
class SearchTool(Tool):
"""搜索工具（模拟）"""
name = "search"
description = "搜索相关信息"
parameters = {
"query": {
"type": "string",
"description": "搜索关键词"
}
}
def execute(self, query: str) -> str:
# 在实际应用中，这里会调用真实的搜索API
# 这里使用模拟数据演示
return f"搜索'{query}'的结果：\n" \
f"1. 相关信息来源A\n" \
f"2. 相关信息来源B\n" \
f"3. 相关信息来源C"
class Agent:
"""
基于SGLang的工具调用Agent
这个Agent能够理解用户意图，选择合适的工具，
执行工具调用，并根据结果生成最终回答。
"""
def __init__(self, model_name: str = "Qwen/Qwen2-7B-Instruct"):
self.runtime = sg.Runtime(model_name=model_name)
self.tools: List[Tool] = []
self._register_default_tools()
def _register_default_tools(self):
"""注册默认工具集"""
self.register_tool(CalculatorTool())
self.register_tool(DateTimeTool())
self.register_tool(SearchTool())
def register_tool(self, tool: Tool):
"""注册新工具"""
self.tools.append(tool)
def _build_tool_schema(self) -> List[Dict]:
"""构建工具Schema用于prompt"""
schemas = []
for tool in self.tools:
schemas.append({
"name": tool.name,
"description": tool.description,
"parameters": tool.parameters
})
return schemas
def _parse_tool_calls(self, response: str) -> Optional[Dict]:
"""从模型响应中解析工具调用"""
# 尝试解析JSON格式的工具调用
try:
# 查找JSON块
if "```json" in response:
start = response.find("```json") + 7
end = response.find("```", start)
json_str = response[start:end].strip()
else:
# 尝试直接解析
json_str = response.strip()
call = json.loads(json_str)
if "tool" in call and "parameters" in call:
return call
except:
pass
# 尝试从文本中提取
import re
pattern = r'(\w+)\((.*?)\)'
matches = re.findall(pattern, response)
if matches:
tool_name, params = matches[0]
return {
"tool": tool_name,
"parameters": {"query": params}
}
return None
def run(self, user_input: str) -> str:
"""
运行Agent处理用户输入
参数:
user_input: 用户的自然语言输入
返回:
Agent的最终响应
"""
tool_schemas = self._build_tool_schema()
@self.runtime.function
def agent_think(input_text: str, tools: List[Dict]):
system_prompt = f"""你是一个智能助手，可以通过工具来帮助用户。
可用的工具：
{json.dumps(tools, ensure_ascii=False, indent=2)}
当用户的问题需要使用工具时，请按以下格式返回：
```json
{{
"tool": "工具名称",
"parameters": {{"参数名": "参数值"}}
}}
如果不需要工具或已完成所有工具调用，直接回答用户问题。
“””
full_prompt = sg.user(system_prompt + f”\n\n用户问题：{input_text}”)
# 模型决定是否调用工具
decision = sg.assistant(sg.gen("decision", max_tokens=500))
return decision
# 首次推理，获取模型决策
decision = agent_think(user_input, tool_schemas)
print(f"模型决策：{decision['decision'][:200]}...")
# 解析工具调用
tool_call = self._parse_tool_calls(decision["decision"])
if tool_call:
tool_name = tool_call["tool"]
params = tool_call["parameters"]
# 查找并执行工具
for tool in self.tools:
if tool.name == tool_name:
print(f"执行工具：{tool_name}")
print(f"参数：{params}")
tool_result = tool.execute(**params)
print(f"工具结果：{tool_result}")
# 使用工具结果生成最终回答
@self.runtime.function
def final_response(user_q: str, tool_r: str):
prompt = sg.user(
f"用户问题：{user_q}\n\n"
f"工具执行结果：{tool_r}\n\n"
f"请根据工具结果回答用户问题。"
)
return sg.assistant(sg.gen("answer", max_tokens=500))
final = final_response(user_input, tool_result)
return final["answer"]
# 无需工具，直接返回决策作为回答
return decision["decision"]
def run_with_tools(self, user_input: str, max_turns: int = 3) -> str:
"""
支持多轮工具调用的Agent
参数:
user_input: 用户输入
max_turns: 最大工具调用轮数
"""
tool_schemas = self._build_tool_schema()
conversation_history = []
@self.runtime.function
def agent_loop(history: List[str], tools: List[Dict]):
system_prompt = f"""你是一个智能助手，可以通过工具来帮助用户。
可用的工具：
{json.dumps(tools, ensure_ascii=False, indent=2)}
规则：
1. 当需要工具时，返回：{{“tool”: “工具名”, “parameters”: {{“参数”: “值”}}}}
2. 当完成所有必要工具调用后，返回最终答案
3. 历史对话中的[TOOL_RESULT]是工具执行结果，请结合它回答
“””
history_text = “\n”.join(history)
full_prompt = sg.user(system_prompt + f”\n\n{history_text}”)
return sg.assistant(sg.gen("response", max_tokens=500))
conversation_history.append(f"用户：{user_input}")
for turn in range(max_turns):
print(f"\n--- 第 {turn + 1} 轮 ---")
result = agent_loop(conversation_history, tool_schemas)
response = result["response"]
print(f"模型响应：{response[:300]}...")
# 检查是否需要调用工具
tool_call = self._parse_tool_calls(response)
if tool_call:
tool_name = tool_call["tool"]
params = tool_call["parameters"]
# 执行工具
for tool in self.tools:
if tool.name == tool_name:
tool_result = tool.execute(**params)
print(f"工具 {tool_name} 结果：{tool_result}")
conversation_history.append(f"助手：{{\"tool\": \"{tool_name}\"}}")
conversation_history.append(f"[TOOL_RESULT] {tool_result}")
break
else:
# 没有工具调用，返回最终回答
conversation_history.append(f"助手：{response}")
return response
return "抱歉，我无法在有限的轮次内完成这个任务。"
使用示例
if name == “main“:
# 创建Agent
agent = Agent()
print("工具调用Agent演示")
print("=" * 60)
# 测试不同类型的查询
test_queries = [
"请计算 (15 + 25) * 3 的结果",
"现在是什么时间？",
"帮我搜索关于量子计算的最新进展"
]
for query in test_queries:
print(f"\n查询：{query}")
print("-" * 40)
response = agent.run(query)
print(f"回答：{response}")
print("=" * 60)
这个案例完整展示了一个功能完备的Agent实现。通过SGLang的链式调用能力，我们可以轻松实现多轮对话和工具调用的复杂逻辑。
**案例四：构建企业级RAG系统**
检索增强生成（RAG）是当前最流行的LLM应用架构之一。在这个案例中，我们将使用SGLang构建一个完整的企业级RAG系统。
```python
import sglang as sg
from typing import List, Dict, Tuple, Optional
import numpy as np
class SimpleVectorStore:
"""
简化的向量存储实现
在实际生产环境中，你应该使用专门的向量数据库，
如Milvus、ChromaDB、Qdrant等。
"""
def __init__(self):
self.documents: List[str] = []
self.embeddings: List[List[float]] = []
def add_documents(self, docs: List[str], embeddings: List[List[float]]):
"""添加文档和对应的embeddings"""
self.documents.extend(docs)
self.embeddings.extend(embeddings)
def search(self, query_embedding: List[float], top_k: int = 5) -> List[Tuple[str, float]]:
"""
搜索最相关的文档
返回:
List of (document, similarity_score) tuples
"""
if not self.embeddings:
return []
# 计算余弦相似度
scores = []
for emb in self.embeddings:
score = np.dot(query_embedding, emb) / (
np.linalg.norm(query_embedding) * np.linalg.norm(emb) + 1e-8
)
scores.append(score)
# 获取top_k
top_indices = np.argsort(scores)[-top_k:][::-1]
return [(self.documents[i], scores[i]) for i in top_indices]
class RAGSystem:
"""
基于SGLang的企业级RAG系统
功能特性：
- 文档检索与重排序
- 多轮上下文管理
- 答案生成与验证
- 源追溯能力
"""
def __init__(
self,
model_name: str = "Qwen/Qwen2-7B-Instruct",
embedding_model: str = "sentence-transformers/all-MiniLM-L6-v2"
):
self.runtime = sg.Runtime(model_name=model_name)
self.vector_store = SimpleVectorStore()
self.embedding_model = embedding_model
self.conversation_history: List[Dict] = []
self.max_history_length = 10
def _get_embedding(self, text: str) -> List[float]:
"""
获取文本的embedding向量
在实际应用中，这里应该调用真正的embedding服务
"""
# 简化实现：使用随机向量
# 真实场景应使用: from sentence_transformers import SentenceTransformer
np.random.seed(hash(text) % (2**32))
embedding = np.random.randn(384).tolist()
# 归一化
embedding = np.array(embedding) / np.linalg.norm(embedding)
return embedding.tolist()
def index_documents(self, documents: List[Dict[str, str]]):
"""
为文档建立索引
参数:
documents: List of dicts with 'id', 'content', 'metadata'
"""
print(f"开始索引 {len(documents)} 个文档...")
contents = [doc["content"] for doc in documents]
embeddings = [self._get_embedding(doc["content"]) for doc in documents]
self.vector_store.add_documents(contents, embeddings)
print(f"索引完成，共 {len(self.vector_store.documents)} 个文档块")
def _rewrite_query(self, query: str) -> str:
"""
查询改写
将口语化查询转换为更适合检索的形式
"""
@self.runtime.function
def query_rewrite(original_query: str):
prompt = sg.user(
f"将以下口语化查询改写为更适合信息检索的形式，"
f"保持原意但使用更精确的关键词。\n\n"
f"原始查询：{original_query}\n\n"
f"改写后的查询："
)
return sg.assistant(sg.gen("rewritten", max_tokens=100))
result = query_rewrite(query)
return result["rewritten"].strip()
def _retrieve(self, query: str, top_k: int = 5) -> List[Tuple[str, float, Dict]]:
"""
检索相关文档
"""
query_embedding = self._get_embedding(query)
results = self.vector_store.search(query_embedding, top_k)
# 添加元数据（实际应用中应该从原始数据获取）
enriched_results = []
for doc, score in results:
enriched_results.append((doc, score, {"source": "knowledge_base"}))
return enriched_results
def _rerank(self, query: str, retrieved_docs: List[Tuple]) -> List[Tuple]:
"""
对检索结果进行重排序
"""
if len(retrieved_docs) <= 1:
return retrieved_docs
@self.runtime.function
def rerank_documents(query: str, docs: List[str]):
doc_list = "\n\n".join([f"[文档{i+1}]\n{doc}" for i, doc in enumerate(docs)])
prompt = sg.user(
f"查询：{query}\n\n"
f"以下是与该查询相关的文档，请按相关性从高到低排序，"
f"输出排序后的文档编号，用逗号分隔。\n\n"
f"{doc_list}\n\n"
f"排序结果（只用编号）："
)
return sg.assistant(sg.gen("ranking", max_tokens=50))
docs_only = [doc for doc, _, _ in retrieved_docs]
rerank_result = rerank_documents(query, docs_only)
# 解析排序结果
ranking_text = rerank_result["ranking"]
print(f"重排序结果：{ranking_text}")
# 按新顺序返回结果
try:
# 尝试解析编号
numbers = [int(x.strip()) for x in ranking_text.split(",") if x.strip().isdigit()]
if numbers and max(numbers) <= len(retrieved_docs):
return [retrieved_docs[i-1] for i in numbers if i <= len(retrieved_docs)]
except:
pass
return retrieved_docs
def _generate_answer(
self,
query: str,
context_docs: List[Tuple],
use_history: bool = True
) -> Dict:
"""
基于检索结果生成答案
"""
# 准备上下文
context_parts = []
for idx, (doc, score, meta) in enumerate(context_docs, 1):
context_parts.append(
f"【参考文档 {idx}】(相关性: {score:.2f})\n{doc}\n"
)
context = "\n".join(context_parts)
# 准备对话历史
history_context = ""
if use_history and self.conversation_history:
recent = self.conversation_history[-3:] # 最近3轮
history_context = "\n\n--- 对话历史 ---\n"
for item in recent:
history_context += f"用户：{item['user']}\n"
history_context += f"助手：{item['assistant']}\n"
@self.runtime.function
def answer_generation(
user_query: str,
context: str,
history: str
):
system_prompt = """你是一个专业的知识助手。你的任务是：
1. 基于提供的参考文档回答用户问题
2. 如果参考文档中有相关信息，务必引用
3. 如果信息不足以回答，可以基于常识推理，但请明确说明
4. 回答要清晰、有条理、专业"""
prompt = sg.user(
f"{system_prompt}\n\n"
f"{history}\n\n"
f"--- 参考文档 ---\n{context}\n\n"
f"--- 用户问题 ---\n{user_query}\n\n"
f"请基于参考文档回答，如果信息不足请说明："
)
answer = sg.assistant(sg.gen("answer", max_tokens=500))
# 同时生成引用信息
citation_prompt = sg.user(
f"用户问题：{user_query}\n\n"
f"回答：{answer['answer']}\n\n"
f"参考文档：{context}\n\n"
f"请列出回答中引用的文档编号（如：参考文档1）。"
)
citation = sg.assistant(sg.gen("citation", max_tokens=100))
return {
"answer": answer["answer"],
"citation": citation["citation"]
}
result = answer_generation(query, context, history_context)
return {
"answer": result["answer"],
"sources": [doc for doc, _, _ in context_docs],
"scores": [score for _, score, _ in context_docs],
"citation": result["citation"]
}
def query(
self,
user_query: str,
top_k: int = 5,
use_rerank: bool = True,
use_history: bool = True
) -> Dict:
"""
处理用户查询的完整流程
参数:
user_query: 用户的问题
top_k: 检索的文档数量
use_rerank: 是否使用重排序
use_history: 是否考虑对话历史
"""
print(f"\n处理查询：{user_query}")
print("-" * 50)
# 步骤1: 查询改写
print("步骤1: 改写查询...")
rewritten_query = self._rewrite_query(user_query)
print(f" 改写后：{rewritten_query}")
# 步骤2: 检索
print("步骤2: 检索相关文档...")
retrieved = self._retrieve(rewritten_query, top_k * 2) # 多检索一些用于重排序
print(f" 初步检索到 {len(retrieved)} 个文档")
# 步骤3: 重排序
if use_rerank and len(retrieved) > 1:
print("步骤3: 重排序...")
reranked = self._rerank(rewritten_query, retrieved)
final_docs = reranked[:top_k]
else:
final_docs = retrieved[:top_k]
print(f" 最终选择 {len(final_docs)} 个文档")
# 步骤4: 生成答案
print("步骤4: 生成答案...")
result = self._generate_answer(user_query, final_docs, use_history)
# 记录对话历史
self.conversation_history.append({
"user": user_query,
"assistant": result["answer"]
})
# 限制历史长度
if len(self.conversation_history) > self.max_history_length:
self.conversation_history = self.conversation_history[-self.max_history_length:]
return result
def reset_history(self):
"""重置对话历史"""
self.conversation_history = []
# 使用示例
if __name__ == "__main__":
# 创建RAG系统
rag = RAGSystem()
# 准备知识库文档
knowledge_base = [
{
"id": "doc_001",
"content": "人工智能（AI）是计算机科学的一个分支，致力于开发能够模拟人类智能的系统。"
"AI技术包括机器学习、深度学习、自然语言处理等多个子领域。"
"近年来，大语言模型的发展使得AI在理解和生成文本方面取得了突破性进展。",
"metadata": {"category": "AI基础", "date": "2024-01"}
},
{
"id": "doc_002",
"content": "机器学习是AI的一个子领域，专注于开发能够从数据中学习的算法。"
"监督学习、无监督学习和强化学习是机器学习的三种主要类型。"
"深度学习使用多层神经网络，在图像识别、自然语言处理等领域表现出色。",
"metadata": {"category": "机器学习", "date": "2024-02"}
},
{
"id": "doc_003",
"content": "RAG（检索增强生成）是一种结合检索系统和生成模型的AI架构。"
"RAG系统首先从外部知识库检索相关信息，然后基于这些信息生成回答。"
"这种架构可以有效缓解大模型的幻觉问题，提高回答的准确性和可追溯性。",
"metadata": {"category": "AI应用", "date": "2024-03"}
},
{
"id": "doc_004",
"content": "向量数据库是RAG系统中的关键组件，用于存储和检索文档的向量表示。"
"常见的向量数据库包括Milvus、Pinecone、Chroma等。"
"向量检索通过计算向量之间的相似度来找到相关内容。",
"metadata": {"category": "AI基础设施", "date": "2024-04"}
}
]
# 建立索引
rag.index_documents(knowledge_base)
# 测试查询
print("\n" + "=" * 60)
print("RAG系统演示")
print("=" * 60)
queries = [
"什么是人工智能？",
"RAG系统是如何工作的？",
"深度学习和机器学习有什么关系？"
]
for q in queries:
result = rag.query(q)
print("\n" + "-" * 50)
print(f"问题：{q}")
print(f"\n回答：{result['answer']}")
print(f"\n引用：{result['citation']}")
print("-" * 50)
这个RAG系统展示了SGLang在实际生产环境中的应用。通过将检索、重排序和生成流程有机结合，我们可以构建功能强大的知识问答系统。
常见使用场景与最佳实践
通过前面的学习，你应该对SGLang有了较为全面的理解。接下来让我们总结一些常见的应用场景，并分享在实际使用中的最佳实践。
场景一：实时对话系统
SGLang非常适合构建需要快速响应的对话系统。通过其优化后的推理引擎，可以实现毫秒级的首token响应。关键配置包括使用流式输出（Streaming）和适当的预填充策略。
场景二：批量文档处理
当需要处理大量文档时，SGLang的批处理能力尤为重要。合理设置batch大小和并发数，可以在保证质量的同时最大化吞吐量。
场景三：复杂Agent架构
对于需要多工具协作的复杂Agent系统，SGLang的链式调用特性让实现变得简单。通过良好的任务分解和清晰的接口设计，可以构建出功能强大的自主Agent。
性能优化技巧
在实际应用中，以下几个优化技巧往往能带来显著的性能提升。首先是合理选择模型大小——在满足需求的前提下，使用更小的模型可以大幅提升推理速度。其次是优化Prompt长度——不必要的system prompt和few-shot示例可以适当精简。再次是利用缓存——对于重复的prefix，可以使用KV cache来避免重复计算。最后是批量处理——当有多个独立请求时，批量提交可以更高效地利用GPU资源。
总结与资源链接
SGLang是一个专注于LLM推理优化的创新框架，它通过结构化生成、链式调用和约束解码等核心特性，让构建复杂LLM应用变得简单高效。无论是构建智能问答系统、RAG应用还是功能强大的Agent，SGLang都能提供坚实的技术支撑。
在实际项目中，SGLang与LangChain、LlamaIndex等框架并非互斥关系，而是可以很好地互补使用。你可以根据具体需求选择合适的工具，构建最佳的解决方案。
相关资源链接
如果你想进一步学习SGLang，以下资源值得关注：
- 官方GitHub仓库：https://github.com/sgl-project/sglang
- 官方文档：https://docs.sglang.ai/
- 模型列表：SGLang支持的主流开源模型包括Qwen系列、LLaMA系列、Mistral系列等
- vLLM集成：作为SGLang的默认后端，vLLM项目也值得关注：https://github.com/vllm-project/vllm
相关AI项目推荐
如果你对LLM应用开发感兴趣，以下项目也值得了解：
- LangChain：成熟的LLM应用开发框架，适合快速原型开发
- LlamaIndex：专注于知识检索增强的框架，与RAG系统天然契合
- vLLM：高效的大模型推理引擎，SGLang的重要依赖
- Transformers：Hugging Face的核心库，提供了丰富的预训练模型
- LiteLLM：统一调用多种LLM API的库，简化多模型切换
下一步学习建议
建议的学习路径是：先通过本文的示例代码进行实践，熟悉SGLang的基本用法；然后尝试将SGLang集成到你自己的项目中，解决实际遇到的问题；最后深入研究官方文档和源码，理解底层的优化原理。
大语言模型领域发展迅速，新的工具和方法层出不穷。保持学习的态度，持续关注社区动态，你一定能在这个领域找到属于自己的价值。祝你学习愉快，coding顺利！
“`
评论区