大模型推理框架#

1
# seq 初始值：Prefill后拼接了第一个新词的完整长序列
2
while True:
3
    # 1. 把当前完整长序列全部重新送入模型
4
    logits, _ = model(seq)
5
    # 2. 依旧只取最后一行logits，预测下一个新词
6
    next_token = torch.argmax(logits[-1, :], dim=0, keepdim=True)
7
    # 3. 新词追加到序列末尾
8
    seq = torch.cat([seq, next_token], dim=-1)
9
    # 4. 命中结束符则终止循环
10
    if next_token == token_eos:
11
        break

1
pip install vllm
2
pip show vllm

1
pip install modelscope
2
modelscope download --model model_name --local_dir local_path

1
vllm serve /root/models/Qwen3-8B --api-key abc123 --served-model-name Qwen/Qwen3-8B --max_model_len 4096 --port 7890
2

3
# 多卡部署的话（以两卡为例）
4
CUDA_VISIBLE_DEVICES=0,1 vllm serve local_path \
5
    --api-key abc123 \                # 设置api-key
6
    --served-model-name Qwen/Qwen3-8B \  # API服务的模型名称
7
    --max_model_len 4096 \            # 最大处理长度（输入prompt + 生成内容的 token 总数）
8
    --tensor-parallel-size 2 \        # 张量并行数量，和使用GPU卡数保持一致
9
    --port 7890

1
根节点
2
└── 节点A [3000token 固定Agent System] ref_count=100
3
     ├── 子节点B [用户query：今天天气]
4
     └── 子节点C [用户query：计算1+1]

1
import sgl
2
from pydantic import BaseModel
3

4
# 1. 连本地SGL服务（GPU推理，主流）
5
sgl.set_default_backend(sgl.RuntimeEndpoint("http://127.0.0.1:30000"))
6

7
# 2. 直连OpenAI云端API（无本地模型）
8
# sgl.set_default_backend(sgl.OpenAI(api_key="xxx"))

1
@sgl.function
2
def agent_chat(user_input: str):
3
    # 直接写prompt文本，{变量}插值
4
    yield f"""你是工具智能体，只能输出JSON工具调用
5
用户问题：{user_input}
6
"""
7
    # 生成输出，存到key="tool_call"
8
    res = sgl.gen("tool_call", max_tokens=256)
9
    return res

1
# 定义约束Schema
2
class ToolSchema(BaseModel):
3
    tool: str
4
    keyword: str
5

6
# 带JSON格式强制约束（自动触发Compressed FSM）
7
sgl.gen(
8
    "output_key",
9
    max_tokens=200,
10
    temperature=0,
11
    schema=ToolSchema  # 直接传Pydantic类
12
)
13

14
# 正则约束替代schema
15
# sgl.gen("res", regex=r"\{.*\}")

1
@sgl.function
2
def multi_search(q):
3
    yield f"问题：{q}"
4
    # 并行两条工具调用分支
5
    with sgl.fork():
6
        yield sgl.gen("search1", schema=ToolSchema)
7
    with sgl.fork():
8
        yield sgl.gen("search2", schema=ToolSchema)
9
    # join 收集所有分支结果
10
    results = sgl.join()
11
    return results

1
# 调用函数
2
output = agent_chat("成都天气")
3
# 取出生成的json字符串
4
json_str = output["tool_call"]
5

6
# 批量批量请求（自动复用Radix缓存）
7
prompts = [agent_chat("问题1"), agent_chat("问题2")]
8
batch_res = sgl.run_batch(prompts)

1
@sgl.function
2
def round_chat(history, new_q):
3
    yield f"历史对话：{history}"
4
    yield f"用户新问题：{new_q}"
5
    yield sgl.gen("reply")

1
import sgl
2
import json
3
from pydantic import BaseModel, Literal
4
from typing import Optional
5

6
# 初始化LLM
7
sgl.set_default_backend(sgl.RuntimeEndpoint("http://127.0.0.1:30000"))
8

9
# 工具输出结构（Compressed FSM）
10
class SingleTool(BaseModel):
11
    tool: Literal["search", "calculate"]
12
    keyword: str
13

14
# 支持一次返回多个工具调用数组
15
class MultiToolCall(BaseModel):
16
    calls: list[SingleTool]
17
    finish: Literal["yes", "no"]  # no=还要继续调用工具 yes=信息足够，直接回答
18

19
# 简单本地工具实现
20
def search(keyword: str) -> str:
21
    if "成都天气" in keyword:
22
        return "成都今日气温26℃，湿度65%，多云"
23
    elif "北京天气" in keyword:
24
        return "北京今日气温22℃，小雨"
25
    return f"搜索结果：{keyword}"
26

27
def calculate(expr: str) -> str:
28
    try:
29
        val = eval(expr)
30
        return f"计算结果 {expr} = {val}"
31
    except Exception as e:
32
        return f"计算失败：{str(e)}"
33

34
# 工具分发执行
35
def execute_tools(call_list: list[SingleTool]) -> list[str]:
36
    outputs = []
37
    for item in call_list:
38
        if item.tool == "search":
39
            outputs.append(search(item.keyword))
40
        elif item.tool == "calculate":
41
            outputs.append(calculate(item.keyword))
42
    return outputs
43

44
# DSL函数：生成工具调用/判断是否结束
45
@sgl.function
46
def agent_plan(history_context: str, user_question: str):
47
    prompt = """
48
你拥有两个工具：
49
1. search：查询城市天气、百科资讯
50
2. calculate：数学表达式四则运算
51

52
规则：
53
1. 需要查资料/计算就输出calls数组，finish填no；
54
2. 信息充足不需要工具时，finish填yes，calls留空；
55
3. 可以一次输出多个工具并行调用；
56
只输出JSON，禁止额外文字。
57

58
历史上下文（包含过往工具调用与返回结果）：
59
{history_context}
60
用户当前问题：{user_question}
61
"""
62
    yield prompt.format(history_context=history_context, user_question=user_question)
63
    res = sgl.gen("plan_out", max_tokens=300, temperature=0, schema=MultiToolCall)
64
    return res
65

66
# DSL函数：最终整理自然语言回答
67
@sgl.function
68
def gen_final_answer(full_context: str, user_question: str):
69
    prompt = """
70
基于下面完整上下文（用户问题、所有工具调用与工具返回数据），用通顺中文完整回答用户，不要调用工具、不要输出JSON。
71
完整上下文：
72
{full_context}
73
用户原始提问：{user_question}
74
"""
75
    yield prompt.format(full_context=full_context, user_question=user_question)
76
    ans = sgl.gen("answer", max_tokens=400)
77
    return ans
78

79
if __name__ == "__main__":
80
    user_q = "北京天气怎么样，123*456等于多少？顺便说下成都湿度"
81
    history = ""
82
    max_loop = 5  # 防止无限循环，限制最多5轮工具调用
83

84
    # ===================== 多轮工具循环核心逻辑 =====================
85
    for loop in range(max_loop):
86
        print(f"\n===== 第{loop+1}轮工具规划 =====")
87
        # 模型生成工具计划
88
        plan_res = agent_plan(history, user_q)
89
        plan_data = json.loads(plan_res["plan_out"])
90
        finish_flag = plan_data["finish"]
91
        tool_calls = plan_data["calls"]
92

93
        print(f"模型输出工具调用：{plan_data}")
94

95
        # 分支1：信息足够，停止工具调用，生成最终回答
96
        if finish_flag == "yes":
97
            print("模型判断无需更多工具，生成最终答案")
98
            final = gen_final_answer(history, user_q)
99
            print("\n【最终回答】", final["answer"])
100
            break
101

102
        # 分支2：存在多个工具，批量执行
103
        tool_returns = execute_tools(tool_calls)
104
        print(f"工具执行结果列表：{tool_returns}")
105

106
        # 把本轮工具调用+结果追加到上下文，下一轮模型可见
107
        round_log = f"\n【第{loop+1}轮工具调用记录】\n调用列表：{tool_calls}\n工具返回：{tool_returns}"
108
        history += round_log
109
    else:
110
        # 循环达到上限强制终止
111
        print("\n达到最大工具调用轮次限制，强制生成回答")
112
        final = gen_final_answer(history, user_q)
113
        print("【最终回答】", final["answer"])

1
# 比较稳定的一个版本
2
pip install "sglang[all]==0.5.10"
3
pip show sglang

1
pip install modelscope
2
modelscope download --model model_name --local_dir local_path

1
python -m sglang.launch_server /root/models/Qwen3-8B \
2
--api-key abc123 \                  # 设置接口鉴权密钥
3
--served-model-name Qwen/Qwen3-8B \ # API调用时使用的模型名称
4
--max-model-len 4096 \              # 最大处理长度（输入prompt + 生成内容token总数）
5
--port 30000                        # 服务监听端口
6
--speculative-algorithm EAGLE3 \                    # 开启EAGLE投机解码总开关
7
--speculative-draft-model-path lmsys/SGLang-EAGLE3-Qwen3-8B \ # 配套草稿模型权重路径
8
--speculative-num-draft-tokens 4                    # 草稿模型单次预生成token数量，推荐2~8
9

10
这个是开启了EAGLE投机解码的，不想开直接把最后三行删了就行了，RadixAttention是默认开启的，Compressed FSM也无需启动参数，调用接口时在extra_body传入json_schema自动生效。
11

12
## 多卡部署（以两卡为例）
13
CUDA_VISIBLE_DEVICES=0,1 python -m sglang.launch_server local_path \
14
--api-key abc123 \                      # 设置接口鉴权key
15
--served-model-name Qwen/Qwen3-8B \     # API调用时使用的模型名
16
--max-model-len 4096 \                  # 上下文最大总token长度（输入+输出）
17
--tp-size 2 \                           # 张量并行卡数，和指定GPU数量一致
18
--port 30000
19
--speculative-algorithm EAGLE3 \                    # 开启EAGLE投机解码总开关
20
--speculative-draft-model-path lmsys/SGLang-EAGLE3-Qwen3-8B \ # 配套草稿模型权重路径
21
--speculative-num-draft-tokens 4                    # 草稿模型单次预生成token数量，推荐2~8
22

23

24
常用附加参数（按需拼接）
25
--host 0.0.0.0 \                    # 允许外部机器访问服务
26
--gpu-memory-utilization 0.85 \     # GPU显存占用上限
27
--max-running-requests 32 \          # 最大并发请求数
28
--log-level info \                  # 日志输出级别
29
--enable-metrics \                 # 开启监控指标采集
30
--stream-interval 1 \              # 流式推送token间隔，单位token，1=逐token实时返回
31
--stream-output                    # 开启分片流式输出底层支持（默认已启用，显式写更稳妥）