思考控制、能力查询及配置见 Reasoning。新接口导出 Reasoning、
ReasoningCapabilities、TokenBudget、ModelCapabilities。
统一的 LLM 客户端,自动选择底层实现。
from flexllm import LLMClient
client = LLMClient(
model: str, # 模型名称
base_url: str = None, # API 地址
api_key: str = "EMPTY", # API 密钥
provider: str = "auto", # "auto", "openai", "gemini"
cache: ResponseCacheConfig = None, # 缓存配置
concurrency_limit: int = 10, # 最大并发数
max_qps: float = None, # QPS 限制
retry_times: int = 3, # 最大尝试次数(含首次调用)
retry_delay: float = 1.0, # 退避基数(秒),指数退避+抖动
timeout: int = 120, # 请求超时(秒);非流式=单请求总时长,流式=相邻 chunk 最长间隔
)方法:
result = await client.chat(messages, model=None, **gen_kwargs)
results = await client.chat_batch(messages_list, model=None, **kwargs)单条 chat() 总是返回 ChatCompletionResult,失败抛结构化异常:
| 字段 | 说明 |
|---|---|
content |
文本内容 |
usage |
{"prompt_tokens", "completion_tokens", "total_tokens"} |
reasoning_content |
思考内容(DeepSeek-R1 / Qwen3 / Claude / Gemini) |
tool_calls |
list[ToolCall] |
finish_reason |
"stop" / "length" / "tool_calls" …(各 provider 统一到这套值) |
assistant_message |
下一轮需原样回传的 assistant 消息(带签名的 thinking / tool_use 时非 None) |
extra |
响应里标准信封之外的顶层字段(网关挂的带外信息) |
raw_response |
provider 原始响应 |
cached / queue_time |
是否缓存命中 / 客户端排队耗时 |
error / ok |
仅批量失败项使用,见下 |
失败抛 LLMRequestError 及其子类(LLMHTTPError / LLMConnectionError /
LLMTimeoutError / LLMResponseError),带 status_code、response_data、
request_id、retryable。
批量 chat_batch() 不抛异常:批量里单条失败是预期结果的一种,不是控制流事件。
返回等长同构的 BatchResult,失败项是 content=None 且带 .error 的同一种结果对象:
results = await client.chat_batch(messages_list, output_jsonl="out.jsonl")
for r in results: # BatchResult 可迭代、可索引、可 len()
if r.ok:
use(r.content)
elif r.error.retryable:
retry_later(r)
results.errors # {index: LLMRequestError},按原始 index 对齐
results.ok # 成功项
results.failed # 失败项
results.success_count / results.failed_count / results.cached_count
results.cost # CostReport 或 None
results.elapsed # 秒
results.raise_for_errors() # 需要 fail-fast 时显式调用只有整批开不了工(参数非法等)才抛异常。chat_sync / chat_batch_sync
是同步版本。这两个入口不接受 return_raw / return_usage / return_summary /
return_cost_report / raise_on_error——返回形状是固定的,传了会直接报错。
批量常用参数:output_jsonl(断点续传)、show_progress、track_cost、
metadata_list、params_list、save_input、flush_interval、save_raw。
其中 save_raw 默认 False:checkpoint 的 result 字段不保存 raw_response
(content/usage 已在记录顶层),百万级批量时这份冗余是实打实的磁盘开销;
需要事后按原始响应分析时再打开。
async for index, result in client.chat_batch_iter(messages_list, output_jsonl="out.jsonl"):
handle(index, result) # result 是 ChatCompletionResult,失败项带 .error与 chat_batch() 是同一份执行(缓存 / 断点续传 / 成本 / 多 endpoint 故障转移都一样),
只是按完成顺序逐条交出,不在内存里攒整批——适合下游流水线处理或超大批量。
- 每个 index 恰好产出一次:checkpoint 恢复项、缓存命中排在最前,然后是本轮请求按完成
顺序,最后是没跑到的项(预算中止时,带 error)。把结果收集起来按 index 排好就等于
chat_batch()。 - 失败项与
chat_batch()一致:content=None且带.error,不抛异常。 - 提前停止会取消在途请求并关闭 checkpoint 文件。直接
break时关闭由事件循环稍后调度; 需要在 break 后立刻读 checkpoint,用contextlib.aclosing包住。 - 参数与
chat_batch()相同;不返回整批的cost/elapsed,需要成本时读客户端的 cost tracker。
以下旧入口在 0.18.0 移除,现在调用会发出 LegacyResponseWarning:
| 旧入口 | 替代 |
|---|---|
chat_completions / chat_completions_sync / chat_completions_or_raise |
chat() / chat_sync() |
chat_completions_batch / chat_completions_batch_sync |
chat_batch() / chat_batch_sync() |
iter_chat_completions_batch |
chat_batch_iter() |
chat_completions_stream |
chat_stream() |
async def chat_completions(
messages: List[dict],
model: str = None,
return_raw: bool = False,
return_usage: bool = False,
raise_on_error: bool = False,
skip_cache: bool = False,
**kwargs
) -> str | ChatCompletionResult | RequestResult单条异步请求。返回形状由 return_raw / return_usage 决定;默认失败返回
RequestResult 而不抛异常,raise_on_error=True 才抛结构化异常。
使用旧返回形状时发出 LegacyResponseWarning。新代码用 chat()。
chat_completions_sync、chat_completions_or_raise 同样已弃用。
async def chat_completions_batch(
messages_list: List[List[dict]],
output_jsonl: str = None,
show_progress: bool = True,
return_summary: bool = False,
return_cost_report: bool = False,
**kwargs
) -> List[str] | Tuple[List[str], dict]批量异步请求,支持断点续传。失败项为 None——想知道每条为何失败,用 chat_batch()。
上表旧入口的返回形状、参数顺序、失败时的返回值与 0.16.x 逐字一致,只在末尾
追加了默认 False 的 raise_on_error。用 v0.16.6 的代码跑同一组调用逐行比对过,
包括这些容易被忽略的角落:
- 单条失败返回
RequestResult而不抛异常;批量失败项为None return_summary在单 endpoint 返回统计字符串、在 pool 返回 dict(两者本就不同)return_cost_report只在启用了cost_tracker时才多返回一项;多 endpoint 分布式批量下不返回(0.16.x 的分布式路径直接丢弃该参数,这个形状一并保留, 但会打一条 warning 指向chat_batch().cost)- checkpoint JSONL 的字段集合不变;旧文件能被新版续跑,新文件也能被旧版读
(
resume_from_jsonl只挑它认识的键) transcribe/speech及其 batch 版本的失败项仍是RequestResult
行为差异只有两处:
chat_completions_or_raise失败时抛的是LLMHTTPError等子类而非LLMRequestError基类,except LLMRequestError照常捕获。- 批量预算中止(0.17.8 起):超预算后不再发新请求,已在途的请求收完后照常返回;
此前这些已被服务端处理、已计费的结果会变成
None,成本报告也少计了它们。返回形状 不变(等长 list,失败项None),只是部分None变成了有效结果;哪些位置是None本来就取决于请求完成的时序。
async for event in client.chat_stream(messages, model=None, **kwargs):
...事件恒为 dict,不受任何开关影响(思考内容只走 thinking 事件,不会以 <think> 混进正文):
| type | 字段 | 说明 |
|---|---|---|
thinking |
content |
思考片段 |
content |
content |
正文片段 |
tool_call_delta |
tool_calls |
工具调用增量(OpenAI 形态,按 index 合并) |
extra |
extra |
网关带外字段 |
result |
result |
最后一条,成功时恰好一次:ChatCompletionResult,与 chat() 同构 |
result 里 content / reasoning_content / tool_calls 已累加好,finish_reason / usage /
assistant_message(下一轮需原样回传的续接状态,如 Claude 带签名的 thinking block)也都在上面,
调用方不需要自己拼。失败抛 typed error,与 chat() 相同。
与 chat() 的差异(均为有意):
content只含正文。OpenAI 兼容端点在thinking=True时,chat()的content带<think>…</think>前缀,这里思考只在reasoning_content;没有正文时为None。- 响应里不带 functionCall id 时(老版本 Gemini 模型),tool call id 是本地合成的:流式按 functionCall 顺序编号,非流式按 part 下标编号,二者不保证相同。Gemini 3 返回真实 id,两边一致。
async for event in client.chat_stream(messages, tools=tools):
if event["type"] == "content":
print(event["content"], end="", flush=True)
elif event["type"] == "result":
result = event["result"] # result.tool_calls / result.assistant_message ...多 endpoint(pool)时,故障转移只发生在首个事件之前:流已开始输出后失败直接抛异常—— 已送出的内容收不回来,换 endpoint 从头再流只会产生重复输出。
流式的 timeout 是空闲超时(两个 chunk 之间的最长间隔),不限制整条流的总时长——
长思考模型一轮可能持续数分钟,只要还在吐 token 就不算卡死。
async def chat_completions_stream(
messages: List[dict],
return_usage: bool = False,
timeout: int = None,
**kwargs
) -> AsyncIterator[str | dict]流式响应。默认逐段 yield str;return_usage=True 时 yield 事件 dict,按序为:
| type | 字段 | 说明 |
|---|---|---|
thinking |
content |
思考片段(reasoning 模型) |
content |
content |
正文片段 |
tool_call_delta |
tool_calls |
工具调用增量 |
assistant_message |
message |
工具回合结束时可原样放入下一轮的 provider 续接消息;仅在存在思考或工具状态时发送 |
finish |
reason |
模型停止原因,OpenAI 语义:stop / length(被 max_tokens 截断)/ tool_calls / content_filter / …;provider 不给时为 None。流末尾必发一次 |
usage |
usage |
token 用量,最后一条(provider 给了才有) |
新代码用 chat_stream():它事件形状固定,并在结尾给出汇总好的结果。
OpenAI 兼容 API 客户端。
from flexllm import OpenAIClient
client = OpenAIClient(
base_url: str,
api_key: str = "EMPTY",
model: str = None,
# ... 其他参数同 LLMClient
)额外参数:
thinking: 思考模式控制False: 禁用思考(官方 OpenAI 与 Ollama 默认端口:11434发reasoning_effort="none", 官方端点上关不掉推理的型号会 400;vLLM 发chat_template_kwargs.enable_thinking=False。 Ollama 跑在其他端口时显式传reasoning_effort="none"——它的/v1不认其他关闭字段)True: 启用思考str: 推理强度,原样作为reasoning_effort发送。可用档位由模型与部署决定 (如 vLLM 上的 Qwen3.8 只收low/medium/xhigh),不支持时后端报错并列出可选值dict: 透传 provider 原生thinking配置None: 使用模型默认行为
静态方法:
@staticmethod
def parse_thoughts(response_data: dict) -> dict解析思考内容,返回 {"thought": str, "answer": str}。
Anthropic Claude 客户端。
from flexllm import ClaudeClient
client = ClaudeClient(
api_key: str,
model: str = "claude-sonnet-4-20250514",
base_url: str = "https://api.anthropic.com/v1",
api_version: str = "2023-06-01",
)thinking 参数:
False: 禁用扩展思考;不支持 thinking 的旧模型会省略参数True: Claude 4.6+ 使用 adaptive thinking;Claude 3.7/4.0-4.5 使用默认 budget_tokensstr: 统一强度minimal/low/medium/high/xhigh/max/ultraint: 为 Claude 3.7/4.0-4.6 启用并指定 budget_tokens;4.7+ 会明确报错dict: 直接传入 Anthropic 原生 thinking 配置None: 使用模型默认行为
也可传 reasoning_effort 作为统一强度入口。Claude 4.6+ 会生成
thinking={"type": "adaptive"} 与 output_config.effort;Claude 3.7/4.0-4.5 自动换算为
token 预算;Claude 3.5 及更早版本不支持该参数。
Fable/Mythos 5 的 adaptive thinking 始终开启,传 thinking=False 会明确报错。
静态方法:
@staticmethod
def parse_thoughts(response_data: dict) -> dict解析思考内容,返回 {"thought": str, "answer": str}。
Google Gemini 客户端。
from flexllm import GeminiClient
# Developer API 模式
client = GeminiClient(
api_key: str,
model: str = "gemini-3-flash-preview",
)
# Vertex AI 模式
client = GeminiClient(
project_id: str,
location: str = "us-central1",
model: str = "gemini-3-flash-preview",
use_vertex_ai: bool = True,
)thinking 参数:
False: 禁用(thinkingBudget=0;Pro 系列不能关思考,API 会拒绝)True: 启用并返回思考内容"minimal","low","medium","high": 思考级别("xhigh"/"max"/"ultra"取最高档"high")。 Gemini 3 发thinkingLevel;Gemini 2.x 不支持级别,换算成thinkingBudget(512/2048/8192/24576)int: 思考 token 预算(thinkingBudget,Gemini 2.5 与 3 均支持)
response_format 的 json_schema 走 responseJsonSchema,OpenAI strict 模式的 schema(带
additionalProperties: false)可直接用。
工具调用(与 OpenAI 用法一致):
tools传 OpenAI 格式即可,自动转成functionDeclarations(schema 走parametersJsonSchema, 完整 JSON Schema 可用);Gemini 原生格式(含googleSearch等)原样透传。tool_choice的"auto"/"none"/"required"/ 指定函数映射到toolConfig。- 多轮时把
result.assistant_message原样追加进 messages。Gemini 3 要求回传 functionCall 上的thoughtSignature,缺了直接 400;assistant_message.content保存的就是带签名的原生 parts。 只保留了 OpenAI 格式历史(content+tool_calls)也能继续,flexllm 会补官方的跳过校验占位签名, 代价是模型看不到自己之前的推理。 role="tool"消息按tool_call_id找回函数名,转成functionResponse;同一步的多条结果并进 一条消息。结果可以带图([text, image_url]块列表):Gemini 3 放进functionResponse.parts; Gemini 2.x 不支持多模态函数结果,图片挪到紧随其后的一条 user 消息(与 functionResponse 同消息时模型看不见,实测)。- Gemini 3 的纯文本回答也带签名,所以
assistant_message通常不为None,不要拿它判断 "有没有工具调用",用tool_calls。只有思考摘要、没有签名的回答(Gemini 2.x)不产出它。 - 工具调用时
finish_reason为"tool_calls"(Gemini 原始值是STOP)。没有对应 OpenAI 值的 停止原因转小写透出,如模型生成了非法调用时为"malformed_function_call"(此时没有 tool_calls)。 usage.completion_tokens包含思考 token(completion_tokens_details.reasoning_tokens), Gemini 的candidatesTokenCount不含它,但它按输出计费。- 外部图片/音视频 URL 会自动下载转 base64(Gemini 不拉取外链),无需手动
preprocess_msg=True。 - 以上协议行为均在 Gemini Developer API(gemini-3-flash-preview / gemini-2.5-flash)上实测;
Vertex AI 路径(
use_vertex_ai=True)共用同一套转换,但未经真实调用验证。
多 Endpoint 客户端池,支持容量感知的负载均衡和故障转移。
from flexllm import LLMClientPool
pool = LLMClientPool(
endpoints: List[dict] = None, # Endpoint 配置列表
clients: List[LLMClient] = None, # 或直接传入客户端
fallback: bool = True, # 故障转移
failure_threshold: int = 3, # 失败阈值
recovery_time: float = 60.0, # 恢复时间(秒)
)方法: 与 LLMClient 完全一致。
@dataclass
class ChatCompletionResult:
content: str # 响应内容
usage: Optional[dict] = None # Token 使用情况
reasoning_content: Optional[str] = None # 思考内容
tool_calls: Optional[list[ToolCall]] = None
queue_time: Optional[float] = None # 客户端排队耗时(semaphore + QPS 漏桶);缓存命中为 None
finish_reason: Optional[str] = None # 停止原因(OpenAI 语义,见 chat_stream 的 result 事件);缓存命中为 None
assistant_message: Optional[dict] = None # 下一轮应原样回传的 provider 续接消息
latency: Optional[float] = None # 端到端耗时;缓存命中为 Nonelatency = queue_time + service_time,只覆盖真实请求,不含消息预处理(图片下载转
base64)。想知道某个 endpoint 自己有多快,用 latency - queue_time,多 endpoint
选路正是据此判断快慢(见高级用法 · 负载均衡策略)。
finish_reason == "length" 表示输出被 max_tokens 截断。注意 reasoning 模型的思考
tokens 也计入 max_tokens:预算过小时可能 content 为空而 finish_reason == "length"。
当工具调用伴随独立思考状态时,下一轮应回传 assistant_message,不要只从正文和
tool_calls 重建 assistant 消息。
@dataclass
class BatchResultItem:
index: int # 请求索引
content: Optional[str] # 响应内容
usage: Optional[dict] # Token 使用
status: str # "success", "error", "cached"
error: Optional[str] # 错误信息
latency: float # 延迟(秒)@dataclass
class TranscriptionResult:
text: str # 转录文本
language: Optional[str] # 识别出的语言
duration: Optional[float] # 音频时长(秒)
segments: list[dict] # 分段(含 start/end/text)
raw: Optional[dict] # 服务端原始响应
def to_srt(self) -> str: ... # 渲染 SRT 字幕
def to_vtt(self) -> str: ... # 渲染 WebVTT 字幕OpenAI 兼容客户端(含 LLMClient)提供转录与合成,详见 语音能力。
# 转录 /audio/transcriptions
text = client.transcribe_sync("a.wav", model="glm-asr")
result = client.transcribe_sync("a.wav", model="glm-asr", return_details=True)
texts = client.transcribe_batch_sync(["a.wav", "b.wav"], model="glm-asr")
# 合成 /audio/speech
audio: bytes = client.speech_sync("你好", model="glm-tts", voice="tongtong")
path = client.speech_sync("你好", model="glm-tts", output="hello.wav")
paths = client.speech_batch_sync(["一", "二"], model="glm-tts", outputs=["1.wav", "2.wav"])去掉 _sync 后缀即为异步版本。并发与限流沿用客户端的 concurrency_limit / max_qps。
from flexllm import ResponseCacheConfig
config = ResponseCacheConfig(
enabled: bool = False,
ttl: int = 86400, # TTL(秒),0 表示永不过期
cache_dir: str = "~/.flexllm/cache/response",
)
# 快捷方法
ResponseCacheConfig.with_ttl(3600) # 1 小时
ResponseCacheConfig.persistent() # 永久from flexllm import (
count_tokens,
count_messages_tokens,
estimate_cost,
estimate_batch_cost,
messages_hash,
MODEL_PRICING,
)
# 计数
tokens = count_tokens("Hello world", model="gpt-4")
tokens = count_messages_tokens(messages, model="gpt-4")
# 成本估算
cost = estimate_cost(tokens, model="gpt-4", is_input=True)
total = estimate_batch_cost(messages_list, model="gpt-4")
# 消息哈希(用于缓存 key)
hash_str = messages_hash(messages)支持的模型定价:
MODEL_PRICING = {
"gpt-4o": {"input": 2.5/1e6, "output": 10/1e6},
"gpt-4": {"input": 30/1e6, "output": 60/1e6},
"gpt-3.5-turbo": {"input": 0.5/1e6, "output": 1.5/1e6},
"claude-3-5-sonnet": {"input": 3/1e6, "output": 15/1e6},
"deepseek-chat": {"input": 0.14/1e6, "output": 0.28/1e6},
"qwen-max": {"input": 2/1e6, "output": 6/1e6},
# ...
}from flexllm import extract_code_snippets, parse_to_obj, parse_to_code
# 提取代码片段
snippets = extract_code_snippets(text)
# 返回: [{"language": "python", "code": "..."}, ...]
# 解析为 Python 对象
obj = parse_to_obj(text)
# 提取代码字符串
code = parse_to_code(text)用于测试和开发的轻量级 Mock LLM 服务器,支持 OpenAI / Claude / Gemini 三种 API 格式。
from flexllm.mock import MockLLMServer, MockServerConfig
config = MockServerConfig(
port=8001, # 端口号
delay_min=0.1, # 最小延迟(秒)
delay_max=0.1, # 最大延迟
model="mock-model", # 模型名称
response_min_len=10, # 响应最小长度(字符)
response_max_len=1000, # 响应最大长度
rps=0, # RPS 限制,0 不限制
token_rate=0, # 流式 token 速率,0 不限制
error_rate=0, # 错误率 (0-1)
thinking=False, # 是否返回思考内容
qa_path="qa.jsonl", # QA 数据集路径,匹配输入时返回确定性回复
)
# 上下文管理器方式(后台进程)
with MockLLMServer(config) as server:
# OpenAI / Claude: server.url -> "http://localhost:8001/v1"
# Gemini: server.gemini_url -> "http://localhost:8001"
pass| 格式 | 端点 | base_url |
|---|---|---|
| OpenAI | POST /v1/chat/completions |
server.url |
| Claude | POST /v1/messages |
server.url |
| Gemini | POST /models/{model}:generateContent |
server.gemini_url |
| Gemini 流式 | POST /models/{model}:streamGenerateContent |
server.gemini_url |
Gemini 端点按真实 API(Gemini 3,实测)还原协议,客户端的协议 bug 在 mock 上就会暴露:
- 响应:
functionCall带id,第一个 functionCall part 带thoughtSignature;文本回答的签名在 文本 part 上(流式是末尾带签名的空文本 chunk,附finishReason);开启思考时工具调用前也有 thought part;candidatesTokenCount不含思考,思考单列thoughtsTokenCount - 与真实 API 一样返回 400 的请求:
tools里有未知字段(如 OpenAI 格式的type/function)functionDeclarations[].parameters含additionalProperties(应改用parametersJsonSchema);generationConfig.responseSchema同理(应改用responseJsonSchema)functionResponse.response不是对象- 当前轮(最后一条用户文本之后)的 functionCall 缺
thoughtSignature,或签名不是合法 base64 (标准与 URL-safe 字母表都接受,官方占位签名skip_thought_signature_validator可通过)
通过 MockServerConfig(thinking=True) 全局启用,或通过请求参数动态触发:
| 格式 | 请求参数 |
|---|---|
| OpenAI | "think": true |
| Claude | "thinking": {"type": "enabled", "budget_tokens": 10000} |
| Gemini | "generationConfig": {"thinkingConfig": {"includeThoughts": true}} |
通过 --qa 指定 JSONL 文件,当用户输入精确匹配时返回预设回复,未匹配则回退随机生成:
{"input": "你好", "output": "你好!有什么可以帮你的?"}
{"input": "1+1等于几", "output": "2"}通过 --log 记录每个请求的输入输出到 JSONL 文件,base64 图片自动替换为占位符(如 <image:32.1KB>):
flexllm mock --log requests.jsonl日志格式:
{"timestamp": "...", "api_format": "openai", "request": {...}, "output": "回复文本", "prompt_tokens": 10, "completion_tokens": 50}flexllm mock # 默认配置
flexllm mock --thinking # 启用思考内容
flexllm mock -p 8080 -d 0.1-0.5 # 自定义端口和延迟
flexllm mock --error-rate 0.3 # 30% 错误率
flexllm mock --qa qa.jsonl # 使用 QA 数据集确定性回复
flexllm mock --log requests.jsonl # 记录请求日志from flexllm import ProviderRouter, ProviderConfig, create_router_from_urls
# 快速创建
router = create_router_from_urls(
urls=["http://host1:8000/v1", "http://host2:8000/v1"],
api_key="EMPTY",
)
# 获取下一个 provider
provider = router.get_next()
# 更新状态
router.mark_success(provider)
router.mark_failed(provider)