API 統合では、結果を検証できる 1 つのタスクから始めてください。例えば、ドキュメントの要約、インシデントノートからのフィールド抽出、スクリーンショットのレビューなどです。ルートのサポート機能とレスポンス動作を確認したら、その作業中のリクエストを基に、より広いエージェントワークフローを構築できます。
Other Ways to Use It
Google AI Studio Playground: モデルを選択して対話的にプロンプトを試せます(Carbon には有料アクセスが必要)。
Gemini app / AI Mode / Antigravity: Google AI Ultra がある場合は利用可能(提供される箇所でモデルを選択)。
本ガイドでは、ネイティブルートを使って Google Gen AI SDK を CometAPI のキーとベース URL で再利用します。現在のデフォルトグループ価格は、Google の導入価格から 20% 低い設定です。モデル ID は設定で保持し、アプリを書き換えずに Carbon と Argon を比較できるようにしてください。ツール、ファイルアップロード、長い継続生成を追加する前に、ルート固有のサポートを確認しましょう。
import os
from google import genai
from google.genai import types
client = genai.Client(
api_key=os.environ["COMETAPI_KEY"],
http_options={
"api_version": "v1beta",
"base_url": "https://api.cometapi.com",
},
)
model = os.getenv("COMETAPI_MODEL", "gemini-4-carbon")
def completed_text(response):
if not response.candidates:
raise RuntimeError("No candidate returned; inspect prompt feedback.")
candidate = response.candidates[0]
reason = getattr(candidate.finish_reason, "value", candidate.finish_reason)
if reason != "STOP":
raise RuntimeError(f"Generation did not finish normally: {reason}")
parts = candidate.content.parts if candidate.content else []
text = "".join(
part.text or "" for part in (parts or []) if not part.thought
)
if not text.strip():
raise RuntimeError("The completed candidate contains no answer text.")
return text
Step 4: Send Your First Request and Check the Result
from pathlib import Path
from carbon_client import client, model, types, completed_text
response = client.models.generate_content(
model=model,
contents="Give me a five-point checklist for reviewing a Python API client.",
config=types.GenerateContentConfig(
max_output_tokens=4096,
thinking_config=types.ThinkingConfig(thinking_level="MEDIUM"),
),
)
answer = completed_text(response)
Path("answer.txt").write_text(answer, encoding="utf-8")
print(answer)
print("Saved answer.txt")
if response.usage_metadata:
print(response.usage_metadata.model_dump(exclude_none=True))
python first_request.py
## How Can Gemini 4 Carbon API Analyze Files?
以下の例は carbon_client.py と同じディレクトリで実行します。ローカルファイルをインラインコンテンツパーツに読み込み、別のファイルアップロードサービスに依存しない初期ワークフローを保ちます。
PDF の理解には視覚コンテンツが含まれます。Gemini はテキスト、表、チャート、画像を合わせて解釈できます。 [Google のドキュメント処理ガイド](https://ai.google.dev/gemini-api/docs/document-processing)は、ファイルサイズとコンテキスト制限の範囲で最大 1,000 ページまでの PDF をサポートすると説明しています。ページ数は上限であり、すべての文書がゲートウェイのリクエストに収まることを保証するものではありません。
小規模で一時的な PDF にはインライン入力を使用します。より大きなファイルや複数ターンの質問を繰り返す場合、Google は Files API を推奨し、ファイル全体を再送しないようにします。このワークフローを CometAPI で使う前に、アップロードエンドポイントとファイル参照のサポートを確認してください。以下のインライン例は実用的な出発点のままです。
### Summarize a local PDF
Google は GenerateContent における[インライン PDF 入力](https://ai.google.dev/gemini-api/docs/document-processing)を文書化しています。小さく、機密でない report.pdf をプロジェクトディレクトリに置き、以下を analyze_pdf.py として保存し、python analyze_pdf.py を実行します。
ローカル PDF を送信し、完了した要約を保存:
from pathlib import Path
from carbon_client import client, model, types, completed_text
response = client.models.generate_content(
model=model,
contents=[
types.Part.from_bytes(
data=Path("report.pdf").read_bytes(),
mime_type="application/pdf",
),
"Summarize this report in five bullets. Do not invent missing figures.",
],
config=types.GenerateContentConfig(max_output_tokens=4096),
)
from pathlib import Path
from carbon_client import client, model, types, completed_text
response = client.models.generate_content(
model=model,
contents=[
types.Part.from_bytes(
data=Path("screen.png").read_bytes(),
mime_type="image/png",
),
"Review this screenshot for unreadable text and overlapping elements. "
"Describe only visible issues and identify their locations.",
],
config=types.GenerateContentConfig(max_output_tokens=4096),
)
print(completed_text(response))
報告された各問題をスクリーンショットと比較してください。JPEG を送る場合は、ファイル名と MIME タイプの両方を一致させます。拡張子の変更だけでは画像は変換されません。
## How Do You Stream Gemini 4 Carbon Responses ?
### Display streamed text and check completion
ストリーミングはサーバー送信イベント(SSE)としてレスポンスチャンクを返し、生成が続いている間にアプリケーションで部分的な出力を表示できます。
SDK は generate_content_stream で[ストリーミングテキスト生成](https://ai.google.dev/gemini-api/docs/text-generation)を提供します。HTTP を直接呼ぶ場合、CometAPI は alt=sse を伴う streamGenerateContent を使用します。
以下を stream_response.py として保存し、python stream_response.py を実行します。到着したテキストを表示し、通常の終了理由でストリームが終わったことを検証します。例外が発生した場合、表示されたテキストは部分結果のままで、受理された回答にはなりません。
ストリーミングテキストを読み取り、完了を要求してから受け入れる:
from carbon_client import client, model, types
finish_reason = None
received_text = False
for chunk in client.models.generate_content_stream(
model=model,
contents="Write a concise checklist for investigating intermittent API timeouts.",
config=types.GenerateContentConfig(max_output_tokens=4096),
):
if not chunk.candidates:
continue
candidate = chunk.candidates[0]
if candidate.finish_reason is not None:
finish_reason = getattr(candidate.finish_reason, "value", candidate.finish_reason)
parts = candidate.content.parts if candidate.content else []
for part in parts or []:
if part.text and not part.thought:
received_text = True
print(part.text, end="", flush=True)
print()
if finish_reason != "STOP" or not received_text:
raise RuntimeError(f"Stream not accepted as complete: {finish_reason}")
import json
from carbon_client import client, model, types, completed_text
response = client.models.generate_content(
model=model,
contents=(
"Extract the service and impact from this incident note: "
"The checkout API returned errors for ten minutes during a database restart."
),
config=types.GenerateContentConfig(
max_output_tokens=4096,
response_mime_type="application/json",
response_schema={
"type": "OBJECT",
"properties": {
"service": {"type": "STRING"},
"impact": {"type": "STRING"},
},
"required": ["service", "impact"],
},
),
)
data = json.loads(completed_text(response))
if not isinstance(data, dict) or not all(
isinstance(data.get(field), str) for field in ("service", "impact")
):
raise ValueError("The response does not match the expected field types.")
print(json.dumps(data, indent=2))