gyc-llm-connect

配置驱动的 Python 模型调用库。当前版本为 0.2.0。已完成离线协议测试；用户已反馈 DeepSeek 文本的普通调用与流式调用测试通过，其他供应商及媒体能力仍需按实际模型验证。

安装

需要 Python 3.11 或更新版本：

python -m pip install gyc-llm-connect

安装后即可 from gyc_llm_connect import Client, AsyncClient。如需 YAML 配置，安装 "gyc-llm-connect[yaml]"。已有安装可执行 python -m pip install --upgrade gyc-llm-connect；固定版本可安装 "gyc-llm-connect==0.2.0"。

本地安装

如果已经获取本项目源码，可以使用虚拟环境以可编辑方式安装：

cd gyc_llm_connect
python3 -m venv .venv
source .venv/bin/activate
python -m pip install -e .
python examples/local_demo.py

local_demo.py 完全离线，不需要 API Key，演示普通调用、流式调用、图片结果和视频任务。

在其他项目的虚拟环境中，也可以安装本地源码：

python -m pip install -e /path/to/gyc_llm_connect

之后直接 from gyc_llm_connect import Client。编辑源码后无需重新安装。

最小调用

import os
from gyc_llm_connect import Client

with Client(
    provider="openai_compatible",
    base_url=os.environ["LLM_BASE_URL"],  # 例如网关的 /v1 地址
    api_key=os.environ["LLM_API_KEY"],
    model=os.environ["LLM_CHAT_MODEL"],
) as client:
    response = client.chat.create(input="你好，请介绍自己")
    print(response.text)

本地无鉴权服务使用 auth="none"，例如：

with Client(provider="ollama", auth="none", model="你的本地模型名") as client:
    print(client.chat.create(input="你好").text)

配置文件

将 examples/llm.example.toml 中需要的配置复制到 llm.local.toml。示例中的其他服务连接不要一起复制，以免要求无关的环境变量。

最小配置：

schema_version = 1

[defaults]
model = "chat"
request_timeout_s = 60
max_retries = 2

[connections.main]
provider = "openai_compatible"
base_url_env = "LLM_BASE_URL"
api_key_env = "LLM_API_KEY"

[models.chat]
connection = "main"
model_env = "LLM_CHAT_MODEL"

export LLM_BASE_URL="你的完整 API 基础地址"
export LLM_API_KEY="你的密钥"
export LLM_CHAT_MODEL="你的模型名称"
python examples/try_llm.py --config llm.local.toml --prompt "你好" --stream

配置支持 TOML、JSON，以及安装 gyc-llm-connect[yaml] 后的 YAML。也可以 Client(config_dict)。

连接与模型可直接配置 base_url、api_key、model，或者使用对应的 _env 引用；每对只能使用一种形式。包不会自动读取 .env。

参数优先级为：单次调用 > 模型 defaults > 连接 defaults > 全局 defaults。环境变量在配置加载时解析；缺失时立即报错，不产生付费请求。

同步与异步流式

from gyc_llm_connect import Client

with Client.from_file("llm.local.toml") as client:
    with client.chat.stream(input="写一段介绍") as stream:
        for text in stream.text_deltas:
            print(text, end="", flush=True)
        response = stream.get_final_response()
        print(response.usage)

完整事件直接 for event in stream 获取，可读取 type、delta、候选索引、内容块索引、工具 ID 和用量。text_deltas 只读取第一个候选。流是单次消费的；消费完文本仍可以取得完整结果。退出上下文会关闭连接。

import asyncio
from gyc_llm_connect import AsyncClient


async def main():
    async with AsyncClient.from_file("llm.local.toml") as client:
        response = await client.chat.create(input="你好")
        print(response.text)
        async with client.chat.stream(input="写一段介绍") as stream:
            async for text in stream.text_deltas:
                print(text, end="", flush=True)
            response = await stream.get_final_response()


asyncio.run(main())

多轮消息、工具与结构化输出

from gyc_llm_connect import Message, Tool, ToolResult

response = client.chat.create(
    messages=[Message("system", "简洁回答"), Message("user", "北京天气如何？")],
    tools=[
        Tool(
            "weather",
            "查询天气",
            {
                "type": "object",
                "properties": {"city": {"type": "string"}},
                "required": ["city"],
            },
        )
    ],
)

if response.tool_calls:
    call = response.tool_calls[0]
    print(call.name, call.parsed_arguments)
    # 应用自行执行工具；SDK 不自动执行任何工具。
    followup = client.chat.create(
        messages=[
            Message("user", "北京天气如何？"),
            Message("assistant", response.output),
            Message("user", [ToolResult(call.id, {"temperature": 20}, name=call.name)]),
        ]
    )

也支持常用的字典消息格式。input 与 messages 互斥。多候选结果保存在 response.candidates；response.text 只读取第一个候选，不拼接候选。

OpenAI、Responses、Gemini、Ollama 适配器支持对应的 response_format 转换；供应商与模型是否接受指定 Schema，以实际接口为准。Anthropic 的原生特定格式可通过 provider_options 传递，本版没有将其声明为统一的 Schema 保证。

图片和视频输入

from gyc_llm_connect import Image, Text

response = client.chat.create(
    model="vision",
    input=[Text("说明图片内容"), Image.from_path("photo.png")],
)

支持 Image.from_url(...)、Image.from_bytes(data, "image/png")。Ollama 图片输入需要本地文件或 bytes。

Gemini 支持视频输入。小文件可以使用 Video.from_path；大文件先上传并等待就绪：

from gyc_llm_connect import Text

file = client.files.upload("clip.mp4", connection="google")
response = client.chat.create(
    model="video_understanding",
    input=[Text("总结这个视频"), file.as_video()],
)
client.files.delete(file)

文件 ID 带连接作用域，跨连接使用会报错。上传文件不会自动删除。默认内嵌上限为 20 MiB，可通过连接的 max_inline_bytes 调整；不会自动抽帧或压缩视频。

图片生成与编辑

result = client.images.generate(
    model="image_generation",
    prompt="一只橘猫坐在窗边",
)
result.images[0].save("cat.png")

edited = client.images.edit(
    model="image_generation",
    image="cat.png",
    prompt="把背景改成花园",
)
edited.images[0].save("garden.png")

同步图片生成、编辑使用 OpenAI 媒体协议；DashScope 旧版 text2image 异步协议提供 images.submit 和 images.generate。不同图片模型支持的参数有差异，不会自动猜测模型能力。

视频任务与恢复

import json
from pathlib import Path

job = client.videos.submit(
    model="video_generation",
    prompt="海边日出，固定镜头",
)
Path("job.json").write_text(json.dumps(job.to_dict()), encoding="utf-8")
print(job.id, job.status)
result = job.wait(timeout=600, poll_interval=2)
result.videos[0].save("sunrise.mp4")

# 重启进程后，使用相同连接名与供应商配置恢复：
job = client.videos.get(json.loads(Path("job.json").read_text()))

OpenAI 视频协议支持本地图片 input_reference="reference.png"；DashScope 图生视频使用 input_reference=Image.from_url(...)。具体尺寸、时长、参考图约束由模型服务验证。

异步接口对应 await client.images.generate(...)、await client.videos.submit(...)、await job.wait(...)，下载使用 await asset.asave(...)。job.result() 只解析已经完成的结果，不执行网络调用。

任务序列化不包含密钥。超时不会取消远端任务；当前没有实现远端取消。下载延迟到 read/save/aread/asave 时执行，必须在 client 关闭前完成；临时 URL 可能过期。

当前实现范围

以下表示代码与离线协议测试覆盖。DeepSeek 文本的普通调用与流式调用已有用户实测反馈；其他供应商及媒体能力尚未完成真实凭据验证。

协议 / profile	对话	流式	图片输入	视频输入	媒体生成
openai / openai_chat	有	SSE	有	未实现	图片生成/编辑，视频任务
openai_compatible	有	SSE	有	未实现	实现 OpenAI 媒体协议的服务才可用
openai_responses 适配器	有	SSE	有，另支持文件	未实现	使用独立媒体接口
anthropic	有	SSE	有	未实现	未实现
gemini	有	SSE	有	有	未实现
ollama	有	NDJSON	本地图片	未实现	未实现
deepseek、qwen	复用兼容协议	SSE	由模型决定	未实现	未声明
dashscope	对话请使用 qwen profile	—	—	—	原生异步图片/视频任务

OpenAI 兼容适配器将 max_output_tokens 转为 max_tokens；openai_chat 转为 max_completion_tokens。选择对应协议可适配旧版兼容服务与新版 OpenAI 接口。

使用 Responses 可配置：

[connections.main.adapters]
chat = "openai_responses"

兼容服务不支持流末尾用量时可传 provider_options={"include_usage": False}。

当前未实现：音频生成/转写、WebSocket 实时会话、Azure/Vertex/Bedrock 原生鉴权、自动故障切换、模型能力自动探测。Audio 已有内容类型，可用于 Gemini 接受的多模态输入。未知模型能力由服务验证；配置 models.<alias>.capabilities 可以明确禁止某些操作。

扩展与错误

• provider_options 传递厂商专属字段；与已规范化字段冲突时本地报错。原生内置工具可通过 provider_options={"tools": [...]} 传入，不能与统一的 tools 参数同时使用。
• connections.<name>.headers/query/endpoints 可覆盖连接头、查询参数和操作路径；获取任务路径可使用 {id} 占位符。
• register_protocol、register_provider 注册自定义协议和供应商。协议接口位于 gyc_llm_connect.protocols.Protocol。调用 load_plugins() 才加载 entry points 插件。
• 公共异常继承 LLMError，包含 status_code/code/request_id/retryable；StreamInterruptedError.partial_response 可读取部分输出。
• 不支持的参数在本地报错。开始输出后不自动重放流；媒体提交不自动重试。网络故障不能保证远端任务没有创建。
• 日志默认不输出请求、响应或凭据；自定义 transport/hooks 的日志由应用自行管理。
• 外部传入 http_client 由调用方关闭；SDK 自建的 HTTP 客户端由上下文管理器关闭。

详细架构见 设计方案。

开发检查

python -m pip install -e '.[dev]'
python -m pytest
ruff check .
ruff format --check .
python -m build

默认测试完全离线，无需凭据。真实连通性测试可以运行 examples/try_llm.py，会产生实际模型调用。