Ajan Yanıtlarını Akışla İletme: Çok Adımlı İş Akışları İçin Gerçek Zamanlı Çıktı
Ajan Yanıtlarını Akışla İletme: Çok Adımlı İş Akışları İçin Gerçek Zamanlı Çıktı
Ajanınız bir soruyu araştırmak, üç araç çağırmak ve bir yanıt oluşturmak için 20 saniye harcıyor. Kullanıcı “Sor” düğmesine tıklıyor ve 20 saniye boyunca dönen bir simgeye bakıyor. Sistemin çalışıp çalışmadığından emin değil. Sayfayı yenilemeyi düşünüyor. Baştan başlaması gerekip gerekmediğini merak ediyor.
Şimdi şunu hayal edin: Kullanıcı “Sor” düğmesine tıklıyor ve 500 milisaniye içinde ajanın ilk kelimeleri beliriyor. Bilgi aradığını izliyor—”🔍 Sipariş veritabanı aranıyor…”—ve sonuçların gerçek zamanlı olarak geldiğini görüyor. Yanıtı yazılırken, cümle cümle okuyor. Aynı 20 saniye. Tamamen farklı bir deneyim.
Akış iletimi, kullanıcıya yönelik ajanlarda güzel bir ek özellik değil; bir UX zorunluluğudur. İlk tokena geçen süre, ajan arayüzlerindeki en önemli gecikme metriğidir. İlerlemeyi gören kullanıcılar sabırlıdır. Hiçbir şey görmeyen kullanıcılar ise vazgeçer. Web performansı üzerine yapılan çalışma çalışma sonra çalışma, algılanan gecikmenin gerçek gecikmeden daha önemli olduğunu göstermektedir; akış iletimi ise ikisi arasındaki uçurumu kapatmak için sahip olduğunuz en güçlü araçtır.
Bu makalede, çok adımlı AI ajanları için gerçek zamanlı akış iletimini nasıl uygulayacağınızı öğreneceksiniz; temel token bazlı iletimden araç çağrısı şeffaflığına, durum güncellemelerine, aşamalı açıklamaya, hata kurtarmaya ve taşıma katmanı seçimlerine kadar.
Bölüm 1: Claude Akış API’si Temelleri
Bir ajanı akışla iletmeden önce tek bir Claude yanıtını akışla iletmeniz gerekir. Temellerle başlayalım.
Akışlı ve Akışsız Karşılaştırması
Akışsız bir API çağrısı, yanıtın tamamı oluşturulana kadar bekler ve ardından hepsini bir anda döndürür. Akışlı bir çağrı ise yanıt üretilirken ilk tokendan itibaren bir dizi olay döndürür.
Koddaki fark minimumdur. Kullanıcı deneyimindeki fark ise çok büyüktür.
Olay Türleri
Claude akış API’si, yapılandırılmış bir sunucu tarafından gönderilen olay dizisi yayınlar:
message_start— Meta veriler içeren ilkMessagenesnesini barındırır (model, rol, kullanım).content_block_start— Bir içerik bloğunun başlangıcını bildirir (metin veya tool_use).content_block_delta— Artımlı içerik taşır: metin parçaları veya kısmi araç girdisi JSON’ı.content_block_stop— Bir içerik bloğunun sonunu bildirir.message_delta— Mesaja yapılan son güncellemeler (durdurma nedeni, son kullanım).message_stop— Akış tamamlanmıştır.
Metin yanıtları için, her biri küçük bir metin parçası taşıyan (genellikle birkaç token) çok sayıda content_block_delta olayı alırsınız.
Temel Akış Uygulaması
Anthropic Python SDK kullanarak senkron bir akış örneği:
import anthropic
client = anthropic.Anthropic()
def stream_basic_response(user_message: str): """Stream a basic Claude response token by token.""" with client.messages.stream( model="claude-sonnet-4-20250514", max_tokens=1024, messages=[{"role": "user", "content": user_message}], ) as stream: for text in stream.text_stream: print(text, end="", flush=True) print() # Newline after stream completes
stream_basic_response("Explain quantum entanglement in simple terms.")Üretim ortamındaki web sunucularında kullanacağınız asenkron sürüm ise şu şekildedir:
import anthropicimport asyncio
async_client = anthropic.AsyncAnthropic()
async def stream_basic_response_async(user_message: str): """Async streaming with the Anthropic SDK.""" async with async_client.messages.stream( model="claude-sonnet-4-20250514", max_tokens=1024, messages=[{"role": "user", "content": user_message}], ) as stream: async for text in stream.text_stream: print(text, end="", flush=True) print()
asyncio.run(stream_basic_response_async("Explain quantum entanglement in simple terms."))Kısmi Tokenları ve UTF-8 Sınırlarını Yönetme
SDK, UTF-8 kod çözmeyi sizin için halleder; ancak ham HTTP akışıyla çalışıyorsanız, çok baytlı karakterlerin parçalar arasında bölünebileceğini unutmayın. Ham baytları her zaman tamponlayın ve yalnızca tam UTF-8 dizileri elde ettiğinizde kod çözme işlemi yapın. anthropic SDK’sının text_stream yineleyicisi bunu otomatik olarak yönetir; bu da ham SSE akışını kendiniz ayrıştırmak yerine SDK kullanmanız için bir başka nedendir.
Bölüm 2: Araç Çağrılarını Akışla İletme
Temel metin akışı artık olmazsa olmaz bir özellik. Asıl zorluk—ve asıl değer—ajanınızın araçları kullandığında ortaya çıkar. Üç aracı sırayla çağıran çok adımlı bir ajan, ne olduğunu kullanıcıya göstermediğiniz sürece araç yürütme sırasında durmuş gibi hissettireibilir.
Araç Çağrıları Akışta Nasıl Görünür?
Claude bir araç kullanmaya karar verdiğinde, akış type: "tool_use" içeren bir content_block_start olayı yayınlar; ardından araç girdisi JSON’ının parçalarını içeren content_block_delta olayları gelir. Tam araç çağrısı bir araya geldikten sonra aracı çalıştırır, sonucu ekler ve konuşmaya devam edersiniz.
Önemli nokta: Araç adını, girdi JSON’ı tamamlanmadan önce content_block_start geldiği anda bilirsiniz. Bu, tam araç çağrısını beklemeden kullanıcıya anında ”🔍 Sipariş veritabanı aranıyor…” gibi bir şey gösterebileceğiniz anlamına gelir.
Tam Akışlı Ajan Döngüsü
Araç çağrılarını gerçek zamanlı olarak görüntüleyen tam bir akışlı ajan döngüsü:
import anthropicimport json
client = anthropic.Anthropic()
# Define toolstools = [ { "name": "search_orders", "description": "Search customer orders by order ID or customer email.", "input_schema": { "type": "object", "properties": { "query": {"type": "string", "description": "Order ID or email"}, }, "required": ["query"], }, }, { "name": "get_shipping_status", "description": "Get real-time shipping status for an order.", "input_schema": { "type": "object", "properties": { "order_id": {"type": "string", "description": "The order ID"}, }, "required": ["order_id"], }, },]
def execute_tool(tool_name: str, tool_input: dict) -> str: """Execute a tool and return the result as a string.""" if tool_name == "search_orders": # Simulated database lookup return json.dumps({ "order_id": "ORD-12345", "customer": "jane@example.com", "items": ["Blue Widget x2", "Red Gadget x1"], "total": "$47.99", "status": "shipped", }) elif tool_name == "get_shipping_status": return json.dumps({ "order_id": tool_input["order_id"], "carrier": "FedEx", "tracking": "7891011", "estimated_delivery": "2026-03-10", "current_location": "Memphis, TN", }) return json.dumps({"error": f"Unknown tool: {tool_name}"})
def stream_agent_response(user_message: str): """ Complete streaming agent loop with real-time tool call display. """ messages = [{"role": "user", "content": user_message}]
while True: # Stream the model response tool_calls = [] current_tool = None
with client.messages.stream( model="claude-sonnet-4-20250514", max_tokens=4096, tools=tools, messages=messages, ) as stream: response = None for event in stream: # The SDK exposes raw events via iteration pass
# Use the helper to collect text and tool use response = stream.get_final_message()
# Process the response content blocks for block in response.content: if block.type == "text": print(block.text, end="", flush=True) elif block.type == "tool_use": tool_name = block.name tool_input = block.input tool_id = block.id
# Show the user what's happening print(f"\n⚙️ Calling tool: {tool_name}({json.dumps(tool_input)})")
# Execute the tool result = execute_tool(tool_name, tool_input) print(f"✅ Result received from {tool_name
---
## İlgili Makaleler
- [Araç Kullanım Desenleri: Güvenilir Ajan-Araç Arayüzleri Oluşturma](/tr/blog/agent-tool-use-patterns/)- [Çok Ajanlı Desenler: Orkestratörler, İşçiler ve Boru Hatları](/tr/blog/multi-agent-patterns/)- [Ajan Hata Kurtarma: Üretim Güvenilirliği için 5 Desen](/tr/blog/agent-error-recovery-patterns/)- [Web Otomasyon Ajanları: Claude ve Computer Use ile Tarayıcı Kontrolü](/tr/blog/web-automation-agents-browser-control-with-claude-and-computer-use/)