cURL:

# 基本提示，保存到文件
curl -o sunset.jpg "https://image.pollinations.ai/prompt/A%20beautiful%20sunset%20over%20the%20ocean"

# 带参数
curl -o sunset_large.jpg "https://image.pollinations.ai/prompt/A%20beautiful%20sunset%20over%20the%20ocean?width=1280&height=720&seed=42&model=flux"


# 使用 kontext 模型进行图像到图像的生成
curl -o logo_cake.png "https://image.pollinations.ai/prompt/bake_a_cake_from_this_logo?model=kontext&image=https://avatars.githubusercontent.com/u/86964862"

Python (requests):

import requests
import urllib.parse

prompt = "A beautiful sunset over the ocean"
params = {
    "width": 1280,
    "height": 720,
    "seed": 42,
    "model": "flux",
    # "nologo": "true", # 可选，为注册的 referrer/token 设置为 "true"
    # "image": "https://example.com/input-image.jpg", # 可选 - 用于图像到图像的生成 (kontext model)
    # "referrer": "MyPythonApp" # 可选，用于基于 referrer 的认证
}
encoded_prompt = urllib.parse.quote(prompt)
url = f"https://image.pollinations.ai/prompt/{encoded_prompt}"

try:
    # 增加超时时间以适应图像生成
    response = requests.get(url, params=params, timeout=300) 
    response.raise_for_status() # 如果状态码不佳则抛出异常

    with open('generated_image.jpg', 'wb') as f:
        f.write(response.content)
    print("图像已保存为 generated_image.jpg")

except requests.exceptions.RequestException as e:
    print(f"获取图像时出错: {e}")
    # 可以检查 response.text 获取 API 的错误信息
    # if response is not None: print(response.text)

cURL:

curl https://image.pollinations.ai/models

Python (requests):

import requests

url = "https://image.pollinations.ai/models"

try:
    response = requests.get(url)
    response.raise_for_status()
    models = response.json()
    print("可用的图像模型:")
    for model in models:
        print(f"- {model}")
except requests.exceptions.RequestException as e:
    print(f"获取模型列表时出错: {e}")

CURL:

# 基本提示
curl "https://text.pollinations.ai/What%20is%20the%20capital%20of%20France%3F"

# 带参数 (模型, 种子, 系统提示)
curl "https://text.pollinations.ai/Write%20a%20short%20poem%20about%20robots?model=mistral&seed=123&system=You%20are%20a%20poet"

# 获取 JSON 响应
curl "https://text.pollinations.ai/What%20is%20AI?json=true"

# 流式响应 (原始 SSE 输出)
curl -N "https://text.pollinations.ai/Tell%20me%20a%20very%20long%20story?stream=true"

Python (requests):

import requests
import urllib.parse
import json

prompt = "Explain the theory of relativity simply"
params = {
    "model": "openai",
    "seed": 42,
    # "json": "true", # 可选: 获取 JSON 字符串格式的响应
    # "system": "Explain things like I'm five.", # 可选
    # "referrer": "MyPythonApp" # 可选，用于基于 referrer 的认证
}
encoded_prompt = urllib.parse.quote(prompt)
encoded_system = urllib.parse.quote(params.get("system", "")) if "system" in params else None

url = f"https://text.pollinations.ai/{encoded_prompt}"
query_params = {k: v for k, v in params.items() if k != "system"} # 如果存在 system，则从查询参数中移除
if encoded_system:
    query_params["system"] = encoded_system

try:
    response = requests.get(url, params=query_params)
    response.raise_for_status()

    if params.get("json") == "true":
        # 响应是一个 JSON *字符串*，需要解析它
        try:
             data = json.loads(response.text)
             print("响应 (JSON 解析后):", data)
        except json.JSONDecodeError:
             print("错误: API 返回了无效的 JSON 字符串。")
             print("原始响应:", response.text)
    else:
        print("响应 (纯文本):")
        print(response.text)

except requests.exceptions.RequestException as e:
    print(f"获取文本时出错: {e}")
    # if response is not None: print(response.text)

cURL:

curl https://text.pollinations.ai/models

Python (requests):

import requests
import json

url = "https://text.pollinations.ai/models"

try:
    response = requests.get(url)
    response.raise_for_status()
    models_data = response.json() 
    print("可用的文本模型和音色:")
    print(json.dumps(models_data, indent=2, ensure_ascii=False))
  
    # 示例：如何根据预期结构解析特定部分：
    # 如果 `models_data` 是一个字典列表，你可以提取模型 ID：
    # if isinstance(models_data, list):
    #    model_ids = [m.get('id') for m in models_data if m.get('id')]
    #    print("\n模型 ID:", model_ids)
  
    # 如果 `models_data` 是一个字典，其中键是模型 ID，值是详细信息：
    # if isinstance(models_data, dict):
    #     print("\n可用音色 (来自 openai-audio 模型详情):")
    #     openai_audio_details = models_data.get('openai-audio', {})
    #     if 'voices' in openai_audio_details:
    #         print(openai_audio_details['voices'])
    #     else:
    #         print("openai-audio 模型未列出特定音色，或结构不同。")

except requests.exceptions.RequestException as e:
    print(f"获取文本模型列表时出错: {e}")

CURL:

curl https://text.pollinations.ai/openai \
  -H "Content-Type: application/json" \
  -d '{
    "model": "openai",
    "messages": [{"role": "system", "content": "你是一个有用的助手。"}, {"role": "user", "content": "今天巴黎天气怎么样？"}],
    "seed": 42
  }'

Python (requests):

import requests
import json

url = "https://text.pollinations.ai/openai"
payload = {
    "model": "openai", # 或 "mistral" 等
    "messages": [
        {"role": "system", "content": "你是一位有帮助的历史学家。"},
        {"role": "user", "content": "法国大革命是什么时候开始的？"}
    ],
    "seed": 101,
    # "private": True, # 可选
    # "referrer": "MyPythonApp" # 可选，用于基于 referrer 的认证
}
headers = {
    "Content-Type": "application/json"
}

try:
    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()
    result = response.json()
    print("助手:", result['choices'][0]['message']['content'])
    # print(json.dumps(result, indent=2)) # 打印完整响应
except requests.exceptions.RequestException as e:
    print(f"执行 POST 请求时出错: {e}")
    # if response is not None: print(response.text)

CURL:

# 使用 -N 进行流式传输
curl -N https://text.pollinations.ai/openai \
  -H "Content-Type: application/json" \
  -d '{
    "model": "openai",
    "messages": [
      {"role": "user", "content": "写一首关于大海的长诗。"}
    ],
    "stream": true
  }'

Python (requests 与 SSE):

import requests
import json
import sseclient # pip install sseclient-py

url = "https://text.pollinations.ai/openai"
payload = {
    "model": "openai",
    "messages": [
        {"role": "user", "content": "给我讲一个慢慢展开的故事。"}
    ],
    "stream": True
}
headers = {
    "Content-Type": "application/json",
    "Accept": "text/event-stream"
}

try:
    response = requests.post(url, headers=headers, json=payload, stream=True)
    response.raise_for_status()

    client = sseclient.SSEClient(response)
    full_response = ""
    print("流式响应:")
    for event in client.events():
        if event.data:
            try:
                # 处理可能的 '[DONE]' 标记
                if event.data.strip() == '[DONE]':
                     print("\n流已结束。")
                     break
                chunk = json.loads(event.data)
                content = chunk.get('choices', [{}])[0].get('delta', {}).get('content')
                if content:
                    print(content, end='', flush=True)
                    full_response += content
            except json.JSONDecodeError:
                 print(f"\n收到非 JSON 数据 (或非 [DONE] 的标记): {event.data}")

    print("\n--- 流结束 ---")
    # print("完整的流式响应:", full_response)

except requests.exceptions.RequestException as e:
    print(f"\n流式请求期间出错: {e}")
except Exception as e:
    print(f"\n处理流时出错: {e}")

cURL:

# 基本的 TTS GET 请求，保存到文件
curl -o hello_audio.mp3 "https://text.pollinations.ai/Hello%20world?model=openai-audio&voice=nova"

# 不同的音色
curl -o welcome_audio.mp3 "https://text.pollinations.ai/Welcome%20to%20Pollinations?model=openai-audio&voice=fable"

Python (requests):

import requests
import urllib.parse

text = "使用 GET 方法生成音频对于短文本来说很简单。"
voice = "echo" # 可选：alloy, echo, fable, onyx, nova, shimmer
output_filename = "generated_audio_get.mp3"

encoded_text = urllib.parse.quote(text)
url = f"https://text.pollinations.ai/{encoded_text}"
params = {
    "model": "openai-audio",
    "voice": voice
}

try:
    response = requests.get(url, params=params)
    response.raise_for_status()

    # 检查响应内容类型是否表示音频文件
    if 'audio/mpeg' in response.headers.get('Content-Type', ''):
        with open(output_filename, 'wb') as f:
            f.write(response.content)
        print(f"音频成功保存为 {output_filename}")
      
    else:
        print("错误：期望得到音频响应，但收到了意外的内容类型或数据。")
        print(f"Content-Type: {response.headers.get('Content-Type')}")
        print("响应体预览（前 200 个字符）:", response.text[:200])

except requests.exceptions.RequestException as e:
    print(f"执行 TTS GET 请求时出错: {e}")
    # if response is not None: print(response.text) # 打印 API 错误以供调试

Python (requests):

import requests
import base64
import json

url = "https://text.pollinations.ai/openai"
headers = {"Content-Type": "application/json"}

def encode_audio_base64(audio_path):
    try:
        with open(audio_path, "rb") as audio_file:
            return base64.b64encode(audio_file.read()).decode('utf-8')
    except FileNotFoundError:
        print(f"错误：在 {audio_path} 未找到音频文件")
        return None

def transcribe_audio(audio_path, question="转录此音频"):
    base64_audio = encode_audio_base64(audio_path)
    if not base64_audio:
        return None

    # 确定音频格式（通过扩展名简单检查）。目前仅支持 WAV 和 MP3。
    audio_format = audio_path.split('.')[-1].lower()
    supported_formats = ['mp3', 'wav'] 
    if audio_format not in supported_formats:
         print(f"警告：可能不支持的音频格式 '{audio_format}'。官方仅支持 {', '.join(supported_formats)}。")
         return None # 或者如果需要严格检查则抛出错误

    payload = {
        "model": "openai-audio",
        "messages": [
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": question},
                    {
                        "type": "input_audio",
                        "input_audio": {
                           "data": base64_audio,
                           "format": audio_format
                        }
                    }
                ]
            }
        ]
        # 可选：如果模型支持，可以添加 'language' (ISO-639-1) 等参数
    }
    try:
        response = requests.post(url, headers=headers, json=payload)
        response.raise_for_status()
        result = response.json()
        transcription = result.get('choices', [{}])[0].get('message', {}).get('content')
        return transcription
    except requests.exceptions.RequestException as e:
        print(f"转录音频时出错: {e}")
        # if response is not None: print(response.text) # 显示 API 错误以供调试
        return None

# --- 使用示例 (取消注释以运行) ---
# # 将 'path/to/your/audio.wav' 替换为实际的音频文件路径 (例如 'sample.wav' 或 'sample.mp3')
# transcript = transcribe_audio('path/to/your/audio.wav') 
# if transcript:
#     print("转录结果:", transcript)
# else:
#     print("转录失败。")

CURL (使用 URL):

# 获取带图像分析的 JSON 响应
curl https://text.pollinations.ai/openai \
  -H "Content-Type: application/json" \
  -d '{
    "model": "openai",
    "messages": [
      {
        "role": "user",
        "content": [
          {"type": "text", "text": "这张图片里有什么？"},
          {"type": "image_url", "image_url": {"url": "https://upload.wikimedia.org/wikipedia/commons/thumb/d/dd/Gfp-wisconsin-madison-the-nature-boardwalk.jpg/1024px-Gfp-wisconsin-madison-the-nature-boardwalk.jpg"}}
        ]
      }
    ],
    "max_tokens": 300
  }'

Python (requests, 使用 URL 和本地文件/base64):

import requests
import base64
import json

url = "https://text.pollinations.ai/openai"
headers = {"Content-Type": "application/json"}

# 辅助函数，将本地图像编码为 base64
def encode_image_base64(image_path):
    try:
        with open(image_path, "rb") as image_file:
            return base64.b64encode(image_file.read()).decode('utf-8')
    except FileNotFoundError:
        print(f"错误：在 {image_path} 未找到图像文件")
        return None

# --- 方案 1: 分析来自 URL 的图像 ---
def analyze_image_url(image_url, question="这张图片里有什么？"):
    payload = {
        "model": "openai", # 确保此模型支持视觉功能
        "messages": [
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": question},
                    {"type": "image_url", "image_url": {"url": image_url}}
                ]
            }
        ],
        "max_tokens": 500 # 可选：限制响应长度
    }
    try:
        response = requests.post(url, headers=headers, json=payload)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"分析 URL 图像时出错: {e}")
        return None

# --- 方案 2: 分析本地图像文件 ---
def analyze_local_image(image_path, question="这张图片里有什么？"):
    base64_image = encode_image_base64(image_path)
    if not base64_image:
        return None

    # 确定图像格式（通过扩展名简单检查）
    image_format = image_path.split('.')[-1].lower()
    if image_format not in ['jpeg', 'jpg', 'png', 'gif', 'webp']:
         print(f"警告：可能不支持的图像格式 '{image_format}'。假设为 jpeg。")
         image_format = 'jpeg' # 对于未知格式，设为默认值或进行更稳健的处理

    payload = {
        "model": "openai", # 确保此模型支持视觉功能
        "messages": [
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": question},
                    {
                        "type": "image_url",
                        "image_url": {
                           "url": f"data:image/{image_format};base64,{base64_image}"
                        }
                    }
                ]
            }
        ],
        "max_tokens": 500
    }
    try:
        response = requests.post(url, headers=headers, json=payload)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"分析本地图像时出错: {e}")
        # if response is not None: print(response.text) # 显示 API 的错误信息
        return None

# --- 使用示例 (取消注释以运行) ---
# result_url = analyze_image_url("https://upload.wikimedia.org/wikipedia/commons/thumb/d/dd/Gfp-wisconsin-madison-the-nature-boardwalk.jpg/1024px-Gfp-wisconsin-madison-the-nature-boardwalk.jpg")
# if result_url:
#     print("URL 图像分析:", result_url['choices'][0]['message']['content'])

# # 将 'path/to/your/image.jpg' 替换为实际的图像文件路径
# result_local = analyze_local_image('path/to/your/image.jpg', question="描述一下主要主体。")
# if result_local:
#     print("本地图像分析:", result_local['choices'][0]['message']['content'])

注意： 这些示例演示了如何定义工具以及如何解释模型调用函数的请求。您需要在自己的应用程序逻辑中实现实际的函数执行（例如本例中的 get_current_weather）。

cURL (定义工具):

curl https://text.pollinations.ai/openai \
  -H "Content-Type: application/json" \
  -d '{
    "model": "openai",
    "messages": [{"role": "user", "content": "波士顿的天气怎么样？"}],
    "tools": [
      {
        "type": "function",
        "function": {
          "name": "get_current_weather",
          "description": "获取给定位置的当前天气",
          "parameters": {
            "type": "object",
            "properties": {
              "location": {
                "type": "string",
                "description": "城市和州，例如 San Francisco, CA"
              },
              "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
            },
            "required": ["location"]
          }
        }
      }
    ],
    "tool_choice": "auto"
  }'
# 预期响应 (如果模型选择调用该工具) 可能包含：
# ... "choices": [ { "message": { "role": "assistant", "tool_calls": [ { "id": "call_abc123", "type": "function", "function": { "name": "get_current_weather", "arguments": "{\"location\": \"Boston, MA\"}" } } ] } } ] ...

Python (requests - 设置和响应处理):

import requests
import json

url = "https://text.pollinations.ai/openai"
headers = {"Content-Type": "application/json"}

# 对话的初始消息
messages = [{"role": "user", "content": "东京的天气怎么样？"}]

# 定义您的应用程序向 AI 模型公开的工具
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "获取给定位置的当前天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string", "description": "城市和州，例如 San Francisco, CA"},
                    "unit": {"type": "string", "enum": ["celsius", "fahrenheit"], "default": "celsius"}
                },
                "required": ["location"]
            }
        }
    }
]

# 初始 API 调用的负载
payload = {
    "model": "openai", # 模型必须支持函数调用
    "messages": messages,
    "tools": tools,
    "tool_choice": "auto" # 允许模型决定是调用工具还是直接响应
                         # 也可以强制指定工具：{"type": "function", "function": {"name": "get_current_weather"}}
}

# --- 你的函数实现 ---
# 这个函数模拟获取天气数据。在实际应用中，
# 它会调用一个真实的天气服务 API。
def execute_get_current_weather(location, unit="celsius"):
    print(f"\n--- 正在执行 get_current_weather(location='{location}', unit='{unit}') ---")
    # 基于位置的虚拟响应
    if "tokyo" in location.lower():
        return json.dumps({"location": location, "temperature": "15", "unit": unit, "description": "多云"})
    else:
        return json.dumps({"location": location, "temperature": "unknown"})
# --- 函数实现结束 ---

try:
    print("--- 第一次 API 调用 (用户请求) ---")
    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()

    # 解析第一次 API 调用的 JSON 响应
    response_data = response.json()

    # 检查模型是否决定调用工具
    if response_data.get("choices", [{}])[0].get("message", {}).get("tool_calls"):
        print("\n--- 模型请求调用工具 ---")
        # 为简单起见，假设只有一个工具调用；如有多个，请遍历 tool_calls
        tool_call = response_data["choices"][0]["message"]["tool_calls"][0] 
        function_name = tool_call["function"]["name"]
        function_args = json.loads(tool_call["function"]["arguments"])

        if function_name == "get_current_weather":
            # 使用模型提供的参数调用你的实际后端函数
            function_response_content = execute_get_current_weather(
                location=function_args.get("location"),
                unit=function_args.get("unit", "celsius") # 处理默认值
            )

            # 将助手的请求（带有 tool_calls）附加到消息历史中
            messages.append(response_data["choices"][0]["message"]) 
            # 将工具的响应附加到消息历史中
            messages.append(
                {
                    "tool_call_id": tool_call["id"], # 将工具调用与其结果关联起来至关重要
                    "role": "tool",
                    "name": function_name,
                    "content": function_response_content, # 你执行的函数的实际结果
                }
            )

            # --- 第二次 API 调用 (带有函数结果) ---
            print("\n--- 第二次 API 调用 (将函数结果发回给模型) ---")
            second_payload = {
                 "model": "openai",
                 "messages": messages # 发送更新后的消息历史，包括工具的输出
            }
            second_response = requests.post(url, headers=headers, json=second_payload)
            second_response.raise_for_status()
            final_result = second_response.json()
            print("\n--- 模型的最终响应 ---")
            print(json.dumps(final_result, indent=2, ensure_ascii=False))
            print("\n最终助手消息:", final_result['choices'][0]['message']['content'])

        else:
            print(f"错误：模型请求了一个未知的函数 '{function_name}'")

    else:
        print("\n--- 模型直接响应 (未调用工具) ---")
        print("助手:", response_data['choices'][0]['message']['content'])

except requests.exceptions.RequestException as e:
    print(f"函数调用请求期间出错: {e}")
    # if response is not None: print(response.text) # 打印 API 错误以供调试
except Exception as e:
     print(f"处理过程中发生意外错误: {e}")

cURL:

# 显示原始 SSE 流
curl -N https://image.pollinations.ai/feed

Python (sseclient-py):

import sseclient # pip install sseclient-py
import requests
import json
import time

feed_url = "https://image.pollinations.ai/feed"

def connect_image_feed():
     while True: # 循环以在出错时重连
        try:
            print(f"正在连接到图像 Feed: {feed_url}")
            # 使用 stream=True 让 requests 处理 SSE
            response = requests.get(feed_url, stream=True, headers={'Accept': 'text/event-stream'})
            response.raise_for_status() # 对 HTTP 错误抛出异常
            client = sseclient.SSEClient(response)

            print("连接已建立。等待新图像...")
            for event in client.events():
                 if event.data:
                     try:
                         image_data = json.loads(event.data)
                         print("\n--- 新图像 ---")
                         print(f"  提示: {image_data.get('prompt', 'N/A')}")
                         print(f"  URL: {image_data.get('imageURL', 'N/A')}")
                         print(f"  模型: {image_data.get('model', 'N/A')}, 种子: {image_data.get('seed', 'N/A')}")
                         # 你可以在这里进一步处理 image_data，例如在 UI 中显示、记录到数据库等。
                     except json.JSONDecodeError:
                         print(f"\n从图像 Feed 收到非 JSON 数据: {event.data}")
                       
        except requests.exceptions.RequestException as e:
            print(f"\n连接图像 Feed 时出错: {e}。10 秒后重连...")
            time.sleep(10) # 等待一段时间再尝试重连
        except KeyboardInterrupt:
             print("\n图像 Feed 被用户中断。正在退出。")
             break # 用户手动中断时退出循环
        except Exception as e:
             print(f"\n图像 Feed 发生意外错误: {e}。10 秒后重连...")
             time.sleep(10)

# --- 使用 (取消注释以运行) ---
# connect_image_feed()

cURL:

# 显示原始 SSE 流
curl -N https://text.pollinations.ai/feed

Python (sseclient-py):

import sseclient # pip install sseclient-py
import requests
import json
import time

feed_url = "https://text.pollinations.ai/feed"

def connect_text_feed():
     while True: # 循环以在出错时重连
        try:
            print(f"正在连接到文本 Feed: {feed_url}")
            response = requests.get(feed_url, stream=True, headers={'Accept': 'text/event-stream'})
            response.raise_for_status() # 对 HTTP 错误抛出异常
            client = sseclient.SSEClient(response)

            print("连接已建立。等待新文本响应...")
            for event in client.events():
                 if event.data:
                     try:
                         text_data = json.loads(event.data)
                         print("\n--- 新文本响应 ---")
                         print(f"  模型: {text_data.get('model', 'N/A')}")
                         # 获取用户提示，如果 messages 中有的话
                         user_prompt = "N/A"
                         if text_data.get('messages') and isinstance(text_data['messages'], list):
                             for msg in text_data['messages']:
                                 if msg.get('role') == 'user' and msg.get('content'):
                                     user_prompt = (msg['content'] or "")[:100] + ("..." if len(msg['content']) > 100 else "")
                                     break
                         print(f"  用户提示: {user_prompt}")

                         # 截断长响应以保持日志整洁
                         response_preview = (text_data.get('response', 'N/A') or "")[:200]
                         if len(text_data.get('response', '')) > 200: response_preview += "..."
                         print(f"  响应: {response_preview}")
                         # 你可以在这里进一步处理 text_data，例如分析内容、显示等。
                     except json.JSONDecodeError:
                         print(f"\n从文本 Feed 收到非 JSON 数据: {event.data}")

        except requests.exceptions.RequestException as e:
            print(f"\n连接文本 Feed 时出错: {e}。10 秒后重连...")
            time.sleep(10) # 等待一段时间再尝试重连
        except KeyboardInterrupt:
             print("\n文本 Feed 被用户中断。正在退出。")
             break # 用户手动中断时退出循环
        except Exception as e:
             print(f"\n文本 Feed 发生意外错误: {e}。10 秒后重连...")
             time.sleep(10)

# --- 使用 (取消注释以运行) ---
# connect_text_feed()