前回（第1回）は Claude API の初期セットアップと、messages.create() で最初のレスポンスを返すところまで紹介しました。今回はその続きとして、実際のアプリ開発で必ず使うことになる3つの機能——システムプロンプト・会話履歴の管理・ストリーミング——をまとめて扱います。
この3つ、どれも「知ってるけど微妙に挙動がわかってない」状態になりがちなんですよね。自分もそうでした。特に会話履歴は最初ちゃんと理解せずに実装して、なぜか会話がかみ合わない謎のチャットボットを生み出した苦い記憶があります。
システムプロンプトの渡し方
Claude API では、システムプロンプトは messages リストに含めるのではなく、専用の system パラメータで渡します。OpenAI API に慣れてる人だと {"role": "system", ...} を messages に入れようとしてエラーになることがあるので注意です。
import anthropic
client = anthropic.Anthropic()
response = client.messages.create(
model="claude-opus-4-20250514",
max_tokens=1024,
system="あなたは簡潔に答えることを好むアシスタントです。回答は3文以内にしてください。",
messages=[
{"role": "user", "content": "Pythonのリスト内包表記を教えてください"}
]
)
print(response.content[0].text)
system パラメータに文字列を渡すだけでOKです。シンプルですが、ここでどれだけ丁寧に書くかがそのままアウトプットの質に直結するので、地味に大事な部分だったりします。
複数のシステム指示を分けて書く
システムプロンプトが長くなってくると、一つの文字列に詰め込むのがつらくなります。そういうときは system に配列を渡す書き方もできます。
response = client.messages.create(
model="claude-opus-4-20250514",
max_tokens=1024,
system=[
{
"type": "text",
"text": "あなたはPythonの専門家です。"
},
{
"type": "text",
"text": "回答はコードと説明をセットで提供してください。",
"cache_control": {"type": "ephemeral"} # Prompt Cachingを使うときはここに書く
}
],
messages=[
{"role": "user", "content": "デコレータの使い方を教えてください"}
]
)
この書き方は Prompt Caching（長いシステムプロンプトをキャッシュしてコストを削減する機能）と組み合わせるときにも使う形式のようです。今回は深追いしませんが、頭の片隅に置いておくと後で役立つかもしれません。
会話履歴の管理
Claude API はステートレスな設計になっています。つまり、APIを呼び出すたびに過去の会話コンテキストは何も覚えていない。会話を続けたければ、自分でやりとりの履歴を積み上げて毎回渡す必要があります。
最初にこれを知ったとき「え、毎回全部渡すの？」と思いましたが、そういう仕様です。その分、コンテキストをどこまで渡すかを自分でコントロールできるので、慣れると悪くないです。
import anthropic
client = anthropic.Anthropic()
conversation_history = []
def chat(user_input: str) -> str:
conversation_history.append({
"role": "user",
"content": user_input
})
response = client.messages.create(
model="claude-opus-4-20250514",
max_tokens=1024,
system="あなたは親切なアシスタントです。",
messages=conversation_history
)
assistant_message = response.content[0].text
conversation_history.append({
"role": "assistant",
"content": assistant_message
})
return assistant_message
print(chat("Pythonでファイルを読み込む方法は？"))
print(chat("その方法でCSVも読めますか？"))
print(chat("pandasを使う方法と比べるとどちらがいいですか？"))
ポイントは、レスポンスを受け取ったあとに assistant ロールのメッセージとして履歴に追加することです。ユーザー側だけ追加してアシスタント側を追加し忘れると、会話の流れがおかしくなります。自分は最初ここでミスりました。
履歴が長くなりすぎる問題
会話を続けているとトークン数がどんどん増えていきます。コンテキストウィンドウの上限に近づいたり、コストが増えたりするので、実用上は何らかのトリミングが必要になります。
MAX_HISTORY = 20 # 直近N件だけ保持
def chat_with_trim(user_input: str) -> str:
conversation_history.append({
"role": "user",
"content": user_input
})
# 直近N件に絞る（奇数にならないよう偶数でトリム）
trimmed = conversation_history[-MAX_HISTORY:]
response = client.messages.create(
model="claude-opus-4-20250514",
max_tokens=1024,
system="あなたは親切なアシスタントです。",
messages=trimmed
)
assistant_message = response.content[0].text
conversation_history.append({
"role": "assistant",
"content": assistant_message
})
return assistant_message
もっとスマートなやり方（要約してコンテキストを圧縮するとか）もあるらしいですが、まずはシンプルにスライスするだけで十分な場面も多いです。設計はユースケース次第かなと思います。
ストリーミングで体験が変わる
ストリーミングは、レスポンスが全部生成されるのを待たずに、生成されたテキストをリアルタイムで受け取る機能です。チャットUIを作るときは実質必須と言っていいです。待機時間の体感が全然違う。
Anthropic Python SDK では client.messages.stream() を使います。
import anthropic
client = anthropic.Anthropic()
with client.messages.stream(
model="claude-opus-4-20250514",
max_tokens=1024,
system="あなたは丁寧な技術解説者です。",
messages=[
{"role": "user", "content": "Pythonのジェネレータについて説明してください"}
]
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
print() # 最後に改行
stream.text_stream はテキストチャンクのイテレータで、これをループするだけでリアルタイムに出力できます。flush=True を忘れるとバッファリングされて結局まとめて表示されるので注意です。
ストリーミング後にレスポンス全体を取得する
ストリーミングしつつ、最終的なレスポンスオブジェクト（使用トークン数など）も取りたい場合は、stream.get_final_message() が使えます。
with client.messages.stream(
model="claude-opus-4-20250514",
max_tokens=1024,
messages=[
{"role": "user", "content": "Dockerの基本的な使い方を教えてください"}
]
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
final = stream.get_final_message()
print()
print(f"\n--- 使用トークン ---")
print(f"入力: {final.usage.input_tokens}")
print(f"出力: {final.usage.output_tokens}")
コスト管理をちゃんとやりたいならトークン数のログは取っておいたほうがいいです。Lambda で動かしてるとふとした瞬間に請求が膨らんでたりするので、自戒を込めて。
ストリーミング + 会話履歴を組み合わせる
ここが実際のチャットアプリに一番近い形です。ストリーミングで表示しながら、受け取ったテキストを最後に履歴へ追加する流れになります。
import anthropic
client = anthropic.Anthropic()
history = []
def stream_chat(user_input: str):
history.append({"role": "user", "content": user_input})
full_response = ""
with client.messages.stream(
model="claude-opus-4-20250514",
max_tokens=1024,
system="あなたは親切なアシスタントです。",
messages=history
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
full_response += text
print()
# ストリーム完了後に履歴へ追加
history.append({"role": "assistant", "content": full_response})
stream_chat("Pythonの非同期処理について教えてください")
stream_chat("asyncioとThreadingはどちらが速いですか？")
ストリーム中に full_response に結合して、ループが終わったあとで履歴に追加する、という流れです。シンプルですがこれで大体のユースケースはカバーできます。
非同期（async）版を使うケース
FastAPI やその他の非同期フレームワークで使う場合は、AsyncAnthropic クライアントに切り替える必要があります。
import asyncio
import anthropic
async def main():
client = anthropic.AsyncAnthropic()
async with client.messages.stream(
model="claude-opus-4-20250514",
max_tokens=1024,
messages=[
{"role": "user", "content": "Rustの特徴を教えてください"}
]
) as stream:
async for text in stream.text_stream:
print(text, end="", flush=True)
print()
asyncio.run(main())
FastAPI で Server-Sent Events（SSE）を使ってフロントにストリーミングを流す構成を作りたいときは、この非同期版が前提になります。そのあたりは別の機会に触れようと思います。
3つをまとめて整理すると
システムプロンプト・会話履歴・ストリーミング、それぞれは難しくないんですが、組み合わせたときに「どこで何をやってるか」がわかりにくくなりがちです。最終的なコードの構造としては、
systemパラメータ → 毎回同じ値を渡す（変えない）messagesパラメータ → ユーザーとアシスタントのやりとりを蓄積して渡すstream()→ テキストを受け取りながら表示し、完了後に履歴へ追加
この3レイヤーの役割を分けて考えると、コードが自然と整理されていく感じがあります。
※この記事にはプロモーションが含まれます
ちなみに、Aiarty Image Enhancer（AI画像高画質化ツール。ノイズ除去・8倍拡大に対応）も気になっています。Aiarty Image Enhancer
📚 シリーズ「Claude API Python 実践入門 Season2」（第2回 / 全4回）
← 前回の記事: 前回の記事はこちら
→ 次回の記事: 公開後にリンクが追加されます
参考になったらクリックしてもらえると嬉しいです!
