Loading...
Loading...
inference.sh 的 Python SDK - 运行AI应用、构建Agent,并集成150+种模型。包名称:inferencesh(可通过 pip install inferencesh 安装)。支持同步/异步、流式传输、文件上传功能。可通过模板或自定义模式构建Agent,具备工具构建器API、Skills以及人工审批机制。适用场景:Python集成、AI应用开发、Agent开发、RAG流水线、自动化。相关关键词:python sdk, inferencesh, pip install, python api, python client, async inference, python agent, tool builder python, programmatic ai, python integration, sdk python
npx skill4agent add inf-sh/skills python-sdk
pip install inferenceshfrom inferencesh import inference
client = inference(api_key="inf_your_key")
# 运行AI应用
result = client.run({
"app": "infsh/flux-schnell",
"input": {"prompt": "A sunset over mountains"}
})
print(result["output"])# 标准安装
pip install inferencesh
# 带异步支持的安装
pip install inferencesh[async]import os
from inferencesh import inference
# 直接传入API密钥
client = inference(api_key="inf_your_key")
# 从环境变量获取(推荐方式)
client = inference(api_key=os.environ["INFERENCE_API_KEY"])result = client.run({
"app": "infsh/flux-schnell",
"input": {"prompt": "A cat astronaut"}
})
print(result["status"]) # "completed"
print(result["output"]) # 输出数据task = client.run({
"app": "google/veo-3-1-fast",
"input": {"prompt": "Drone flying over mountains"}
}, wait=False)
print(f"任务ID: {task['id']}")
# 后续可通过 client.get_task(task['id']) 检查状态for update in client.run({
"app": "google/veo-3-1-fast",
"input": {"prompt": "Ocean waves at sunset"}
}, stream=True):
print(f"状态: {update['status']}")
if update.get("logs"):
print(update["logs"][-1])| 参数 | 类型 | 描述 |
|---|---|---|
| string | 应用ID(命名空间/名称@版本) |
| dict | 匹配应用架构的输入数据 |
| dict | 隐藏的配置设置 |
| string | 'cloud'(云端)或 'private'(私有部署) |
| string | 用于有状态执行的会话ID |
| int | 空闲超时时间(1-3600秒) |
result = client.run({
"app": "image-processor",
"input": {
"image": "/path/to/image.png" # 自动上传
}
})from inferencesh import UploadFileOptions
# 基础上传
file = client.upload_file("/path/to/image.png")
# 带选项的上传
file = client.upload_file(
"/path/to/image.png",
UploadFileOptions(
filename="custom_name.png",
content_type="image/png",
public=True
)
)
result = client.run({
"app": "image-processor",
"input": {"image": file["uri"]}
})# 启动新会话
result = client.run({
"app": "my-app",
"input": {"action": "init"},
"session": "new",
"session_timeout": 300 # 5分钟
})
session_id = result["session_id"]
# 在同一会话中继续执行
result = client.run({
"app": "my-app",
"input": {"action": "process"},
"session": session_id
})agent = client.agent("my-team/support-agent@latest")
# 发送消息
response = agent.send_message("Hello!")
print(response.text)
# 多轮对话
response = agent.send_message("Tell me more")
# 重置对话
agent.reset()
# 获取聊天历史
chat = agent.get_chat()from inferencesh import tool, string, number, app_tool
# 定义工具
calculator = (
tool("calculate")
.describe("执行计算操作")
.param("expression", string("数学表达式"))
.build()
)
image_gen = (
app_tool("generate_image", "infsh/flux-schnell@latest")
.describe("生成图片")
.param("prompt", string("图片描述"))
.build()
)
# 创建Agent
agent = client.agent({
"core_app": {"ref": "infsh/claude-sonnet-4@latest"},
"system_prompt": "You are a helpful assistant.",
"tools": [calculator, image_gen],
"temperature": 0.7,
"max_tokens": 4096
})
response = agent.send_message("What is 25 * 4?")| 模型 | 应用引用 |
|---|---|
| Claude Sonnet 4 | |
| Claude 3.5 Haiku | |
| GPT-4o | |
| GPT-4o Mini | |
from inferencesh import (
string, number, integer, boolean,
enum_of, array, obj, optional
)
name = string("用户姓名")
age = integer("年龄(岁)")
score = number("分数 0-1")
active = boolean("是否活跃")
priority = enum_of(["low", "medium", "high"], "优先级")
tags = array(string("标签"), "标签列表")
address = obj({
"street": string("街道"),
"city": string("城市"),
"zip": optional(string("邮政编码"))
}, "地址")greet = (
tool("greet")
.display("问候用户")
.describe("按姓名问候用户")
.param("name", string("要问候的姓名"))
.require_approval()
.build()
)generate = (
app_tool("generate_image", "infsh/flux-schnell@latest")
.describe("根据文本生成图片")
.param("prompt", string("图片描述"))
.setup({"model": "schnell"})
.input({"steps": 20})
.require_approval()
.build()
)from inferencesh import agent_tool
researcher = (
agent_tool("research", "my-org/researcher@v1")
.describe("研究指定主题")
.param("topic", string("研究主题"))
.build()
)from inferencesh import webhook_tool
notify = (
webhook_tool("slack", "https://hooks.slack.com/...")
.describe("发送Slack通知")
.secret("SLACK_SECRET")
.param("channel", string("频道"))
.param("message", string("消息内容"))
.build()
)from inferencesh import internal_tools
config = (
internal_tools()
.plan()
.memory()
.web_search(True)
.code_execution(True)
.image_generation({
"enabled": True,
"app_ref": "infsh/flux@latest"
})
.build()
)
agent = client.agent({
"core_app": {"ref": "infsh/claude-sonnet-4@latest"},
"internal_tools": config
})def handle_message(msg):
if msg.get("content"):
print(msg["content"], end="", flush=True)
def handle_tool(call):
print(f"\n[工具: {call.name}]")
result = execute_tool(call.name, call.args)
agent.submit_tool_result(call.id, result)
response = agent.send_message(
"Explain quantum computing",
on_message=handle_message,
on_tool_call=handle_tool
)# 从文件路径读取
with open("image.png", "rb") as f:
response = agent.send_message(
"What's in this image?",
files=[f.read()]
)
# 从base64格式读取
response = agent.send_message(
"Analyze this",
files=["data:image/png;base64,iVBORw0KGgo..."]
)agent = client.agent({
"core_app": {"ref": "infsh/claude-sonnet-4@latest"},
"skills": [
{
"name": "code-review",
"description": "代码评审指南",
"content": "# 代码评审\n\n1. 检查安全性\n2. 检查性能..."
},
{
"name": "api-docs",
"description": "API文档",
"url": "https://example.com/skills/api-docs.md"
}
]
})from inferencesh import async_inference
import asyncio
async def main():
client = async_inference(api_key="inf_...")
# 异步执行应用
result = await client.run({
"app": "infsh/flux-schnell",
"input": {"prompt": "A galaxy"}
})
# 异步Agent
agent = client.agent("my-org/assistant@latest")
response = await agent.send_message("Hello!")
# 异步流式传输
async for msg in agent.stream_messages():
print(msg)
asyncio.run(main())from inferencesh import RequirementsNotMetException
try:
result = client.run({"app": "my-app", "input": {...}})
except RequirementsNotMetException as e:
print(f"缺少依赖项:")
for err in e.errors:
print(f" - {err['type']}: {err['key']}")
except RuntimeError as e:
print(f"错误: {e}")def handle_tool(call):
if call.requires_approval:
# 展示给用户,获取确认
approved = prompt_user(f"是否允许执行{call.name}?")
if approved:
result = execute_tool(call.name, call.args)
agent.submit_tool_result(call.id, result)
else:
agent.submit_tool_result(call.id, {"error": "用户已拒绝"})
response = agent.send_message(
"Delete all temp files",
on_tool_call=handle_tool
)# JavaScript SDK
npx skills add inference-sh/skills@javascript-sdk
# 完整平台Skill(通过CLI访问所有150+应用)
npx skills add inference-sh/skills@inference-sh
# LLM模型
npx skills add inference-sh/skills@llm-models
# 图片生成
npx skills add inference-sh/skills@ai-image-generation