Agent Server

Ajan sunucusu, ajan kodunuzu bir hizmete dönüştüren kütüphanedir. Ajan döngüsünü HTTP sunucusunda sarar, istemcilerin ajanı çalıştırmak için çağırdığı API'yi tanımlar, istemci bağlantılarını yönetir ve bir çalışma kesintiye uğradığında ne olacağını belirler. Ajan sunucusu, ajan çalışma zamanında çalışır. Katmanların nasıl birbirine uyduğunu öğrenmek için Azure Databricks'te aracıları dağıtma sayfasına bakınız.

Yeni ajanlar için Databricks, bu sayfada açıklanan ve AgentKit'in sağladığı ajan sunucusunu önerir. Eski ajan sunucularıyla karşılaştırmak için bkz. Azure Databricks'te ajan sunucularını karşılaştır.

Requirements

  • Python 3.10 ve üzeri.
  • AgentKit'i de içeren databricks-agentbricks paketi. agentbricks init ile oluşturduğunuz projeler onu bir bağımlılık olarak tanımlar.

AgentKit

AgentKit, Agent Runtime üzerinde çalışan ajanlar oluşturmak için Agent Bricks kütüphanesidir. Ajanınıza hizmet veren ajan sunucusunu ve ajan belleği, ajan oturumları, izleme ve kimlik doğrulama için istemci ve yardımcıları sağlar. Agent Bricks CLI ile oluşturduğunuz projeler varsayılan olarak AgentKit kullanır.

Agent sunucusu

AgentKit, ajan sunucusunu DurableAgentServer sınıfı aracılığıyla sağlar. Ajan sunucusu, ajan döngünüzü çağırma API'sine hizmet veren bir HTTP sunucusu içinde çalıştırır, her çalışmayı takip eder ve bir çökme veya yeniden başlatma sırasında kesilen çalışmaları kurtarır. Şunları sağlar:

  • Her istek modu için tek bir API: Senkron, akış ve arka plan çağrıları, ayrıca akışın yeniden bağlanması, hepsi aynı işleyici tarafından sunulur.
  • Idempotent çağrılar: İstemci tarafından oluşturulan bir çağırma kimliği, yeniden denenen bir isteğin yinelenen bir çalıştırma başlatmamasını sağlar.
  • Sıralı oturumlar: Aynı oturumdaki çağrılar birer birer sırayla çalıştırılır.
  • Kalıcı çalıştırma durumu: Dağıtıldığında, çalıştırma durumu, olaylar ve sonuçlar işçi yeniden başlatmalarından sonra da korunur.
  • Çökme kurtarma: Sunucu kesintiye uğramış çalıştırmaları algılar ve bir ikame denemesi başlatır.
  • İstek-kullanıcı yetkilendirmesi: Araçlar, talebi gönderen kullanıcının izinleriyle hareket edebilir.
  • Özel uç noktalar: Ajan sunucusu bir FastAPI uygulamasıdır, bu yüzden kendi rotalarınızı ekleyebilirsiniz.

AgentKit neler içeriyor

AgentKit, databricks_agentkit Python kütüphanesidir. Şunları dışa aktarır:

Export Description
DurableAgentServer, InvocationContext Ajan sunucusu ve invoke ile recovery işleyicilerinize ilettiği bağlam.
AgentKitClient Ajan belleği ve ajan oturum depoları için bir istemci. Mağazaları oluşturur ve alır, mağazaların belleklerini ve oturumlarını Memory, MemoryStore, MemorySearchResult, Session, SessionStore ve SessionItem nesneleri olarak sunar.
configure_tracing, start_trace Ajan için MLflow izleme kur ve bir iş birimi etrafında bir iz başlat.
workspace_client, workspace_headers Doğrulamalı bir Databricks SDK WorkspaceClientoluşturun veya ajanın ortamından doğrudan HTTP çağrıları için kimlik doğrulama başlıkları alın.
list_ai_gateway_model_services Ajanın Unity Gateway üzerinden çağırabileceği model hizmetlerini listeleyin.

Kütüphane ayrıca databricks_agentkit.openai ve databricks_agentkit.langgraph içinde çerçeve yardımcıları da içerir; oluşturulan şablonlar bunları her framework'ü oturum deposuna bağlamak için kullanır. Bellek ve oturum API'leri için bkz. Ajan belleği ve Ajan oturumları.

Temsilcinizi kaydedin

Bir proje oluşturduğunuzda agentbricks init, CLI bunu sizin için yapar. Oluşturulan runtime/main.py, sunucuyu oluşturur ve şablonun invoke ve recovery işleyicilerini kaydeder; böylece yalnızca agent/ içindeki aracı kodunu düzenlersiniz. Bu bölümdeki adımları izleyerek mevcut bir aracıyı içe aktarabilir veya kendi işleyicinizi yazabilirsiniz.

Bir DurableAgentServer oluşturun ve @app.invoke ile bir asenkron çağrı işleyicisi kaydedin. İşleyici isteğin input öğesini ve çağrı bağlamını alır ve JSON serileştirilebilir bir sonuç döndürür. İlerlemeleri etkinlik olarak yayınlayın.context.emit

from databricks_agentkit import DurableAgentServer, InvocationContext

app = DurableAgentServer()


@app.invoke
async def invoke(input, context: InvocationContext) -> dict:
    await context.emit({"type": "status", "message": "Looking that up"})
    answer = await run_my_agent(input, session_id=context.session_id)
    return {"answer": answer}

Bir çağırma işleyicisi kaydedebilirsiniz ve sunucu onsuz başlamaz. İşleyici her istek modunu sunar: istemci sonucu beklemek, olayları akış olarak iletmek veya arka planda çalıştırmak arasında seçim yapar.

Sunucuyu yerel olarak çalıştırmak için onu agentbricks dev ile başlatın. agentbricks init ile oluşturduğunuz projelerde, Uvicorn ile sunucuyu çalıştıran bir giriş noktası ve dağıtımdan sonra aynı giriş noktasını başlatan bir app.yaml dosyası bulunur.

Çağırma bağlamı

İşleyicinin ikinci argümanı şudur InvocationContext:

Öznitelik Description
invocation_id Müşterinin bu çağrı için gönderdiği ID.
session_id Çağrının ait olduğu oturum ya da istemci göndermediyse None
attempt Deneme numarası. İlk deneme 1 mükemmeldir.
is_recovery True kurtarma yöneticisi bir değiştirme girişimi çalıştırırken.
emit(event) Bir JSON etkinliğini saklar, akış istemcilerine sunar ve etkinliğin akıştaki konumunu döndürür.
request_auth Aracının istek-kullanıcı yetkilendirmesi gerektirdiği durumlarda kullanılan istek-kullanıcı kimlik bilgisi çözümleyicisi. Tersi durumda None.

Çağrı API'si

DurableAgentServer Çağırma API'sine şu adreste /api/invocationshizmet verir:

  • POST /api/invocations bir çağrı başlatır. Varsayılan olarak, istek sonucun gelmesini bekler. Olayları Server-Sent Olaylar olarak almak için stream olarak ayarlayın veya hemen bir durum URL'siyle döndürmek için background.
  • GET /api/invocations/<id> bir çağrının durumunu ve tamamlandıktan sonra çıktısını döndürür.
  • GET /api/invocations/<id>/events?after=<event-id>, kaydedilen olayları akış olarak iletir, böylece bir istemci bağlantı kesildikten sonra yeniden bağlanabilir.

İstek alanları, örnekler ve yanıt formatları için bkz. Azure Databricks'te konuşlandırılan ajanları sorgulama.

İdemponentlik

Müşteriler her çağrıda bir UUID id gönderir. Sunucu, ID'yi bir idempotency anahtarı olarak ele alırken çağrı kaydını korur: aynı isteği tekrar göndermek, ajanı yeniden çalıştırmak yerine mevcut invocation'ı döndürür. Farklı bir istek için bir kimliği yeniden kullanmak 409 hatası veriyor.

Sessions

Müşteriler, çağrıları tek bir konuşmada gruplamak için bir session_id gönderebilir. Sunucu, oturum kimliğini input öğesinden ayrı olarak saklar, bunu işleyicinize context.session_id olarak verir ve aynı oturum kimliğini paylaşan çağrıları sırayla, birer birer çalıştırır. Sunucu, çağrı kimliğinden veya girdiden oturum çıkarmaz. Oturum kimliği olmadan, bir çağrı oturumsuzdur.

Çalışma durumu

DurableAgentServer her çağrının isteğini, durumunu, kalp atışlarını, olaylarını ve sonucunu Runtime Store içinde depolar.

  • Yerel geliştirme: agentbricks dev süreç içi bir Runtime Store kullanır. Çağırma API'si de aynı şekilde davranır, ancak süreç durduğunda çalıştırma durumu kaybolur ve sunucu kesintiye uğramış işi yeniden başlatmaz.
  • Dağıtılan aracılar: agentbricks deploy, Azure Databricks tarafından yönetilen bir Lakebase projesinde her dağıtımın Runtime Store’u için özel bir veritabanı sağlar ve yeniden dağıttığınızda bunu yeniden kullanır. Kendi Lakebase projenizi Runtime Store için kullanamazsınız ve bunu kendiniz oluşturmaz veya bağlamazsınız. Sonuçlar ve olaylar işçi yeniden başlatmalarına dayanır ve ajanın herhangi bir örneği durum ve yeniden bağlantı taleplerini karşılayabilir. agentbricks deployments delete Dağıtımla birlikte Runtime Store'u kaldırır.

Çalışma Zamanı Deposu, sunucunun yürütme durumunu tutar. Bu, ajanınızın konuşma geçmişi ve uzun süreli hafıza için kullandığı oturum ve hafıza depolarından ayrıdır.

Çökme kurtarma

Bir worker çökmesi veya yeniden başlatmasının kesintiye uğrattığı çalıştırmaları kurtarmak için, @app.recover ile bir kurtarma işleyicisi kaydedin. Dağıtılan bir sunucu, bir çalışmanın kalp atışlarının durduğunu tespit ettiğinde, mevcut bir worker düğümünde bir değiştirme girişimi başlatır ve kurtarma işleyicisini orijinal girdiyi kullanarak çağırır.

@app.recover
async def recover(input, context: InvocationContext) -> dict:
    # Resume from the agent's last checkpoint in the session store,
    # or replay the input if that's safe for your agent.
    return await resume_my_agent(input, session_id=context.session_id)

Bir kurtarma işleyicisi kaydetmezseniz, otomatik kurtarma kapalı olur ve sunucu başlatıldığında günlüğe bir uyarı yazar.

Kurtarma şu şekilde çalışır:

  • İyileşme başladığında: Her koşu denemesi her birkaç saniyede bir kalp atışı gönderir. Örneğin, worker çöktüğü, yeniden başlatıldığı veya yeniden dağıtım sırasında değiştirildiği için kalp atışları durursa, sunucu duran çalıştırmayı saniyeler içinde tespit eder ve yerine geçecek yeni bir deneme başlatır.
  • Kurtarma başlamadığında: İşleyiciniz bir istisna çıkarırsa, çağrı başarısız olur ve sunucu bunu yeniden denemez. Kurtarma, kesintiye uğrayan worker'ları kapsar, ajan kodunuzdaki hataları değil.
  • Deneme sayısı: Sunucu, kurtarma denemelerinin sayısını sınırlamaz. Her değiştirme denemesi context.attempt değerini bir artırır. Birkaç denemeden sonra durdurmak için, context.attempt kurtarma işleyicisini kontrol edip hata oluştur.
  • Manuel kurtarma: Kurtarmayı manuel olarak tetikleyemezsiniz. Aynı çağırım kimliğiyle bir isteği yeniden göndermek, yeni bir deneme başlatmak yerine mevcut çağırımı döndürür.

Kurtarma, aynı çağrı için ajan kodunuzu birden fazla kez çalıştırabilir. Kesintiye uğramış bir girişim, değiştirme girişimi başlamadan önce harici sistemleri çağrımış olabilir, bu yüzden bu çağrıları idempotent yapın.

Kullanıcı yetkilendirmesi isteği

Varsayılan olarak, ajanınızın araçları uygulamanın service principal'ının izinleriyle çalışır. Talebi gönderen kullanıcının izinleriyle bir araç çalıştırmak için, kullanıcı yetkilendirmesini agent.toml içinde bildirin:

  • Yönetilen bir araç için, araç girişinde auth = "user" ayarlayın. agentbricks tools add MCP sunucuları, sandboxlar ve Genie Agentleri için komutlar varsayılan olarak auth = "user" yazar. Bunun yerine uygulamanın kimliğini kullanmak için --auth app öğesini iletin.

  • Kod içinde yazdığınız bir araç için, gereksinimi ve Agent Bricks'in çıkaramadığı API kapsamlarını belirtin:

    [auth.user]
required = true
additional_api_scopes = ["sql"]

Bir ajan kullanıcı yetkilendirmesi gerektirdiğinde, DurableAgentServer kullanıcının kimlik bilgilerini güvenilir Databricks Apps istek başlıklarından okur ve sadece aktif deneme için hafızada tutar. Runtime Store kimlik bilgilerini depolamaz. Handler'ınızda, kullanıcı için bir workspace istemcisi alın:context.request_auth

@app.invoke
async def invoke(input, context: InvocationContext) -> dict:
    user_client = context.request_auth.client_for("user")
    me = user_client.current_user.me()
    return {"answer": f"Hello, {me.user_name}"}

client_for("app") uygulamanın service principal'ını kullanan bir istemci geri döndürür. Resolver deneme bittiğinde kapanır, bu yüzden istemciyi depolamak yerine işleyicinin içinde çağırın. Aracıyı yerel ortamda çalıştırdığınızda agentbricks dev, client_for("user") yerel kimlik bilgilerinizi kullanır.

Dağıtım yaptığınızda, agentbricks deploy araçlarınızın ihtiyaç duyduğu Databricks Apps kullanıcı kapsamlarını talep eder. Mevcut bir uygulamaya eksik kapsamları eklemek için --allow-user-scope-update. Bkz . Databricks uygulamasında yetkilendirmeyi yapılandırma.

İstek-kullanıcı çağrıları aynı senkron, streaming, arka plan ve yeniden bağlantı API'lerini kullanır. Sunucu, kullanıcının kimlik bilgilerini saklamadığı için, kesintiye uğramış istek-kullanıcı çağrımını kurtaramaz. Değiştirme denemesi, işleyicileriniz çalışmadan önce MCP_USER_AUTH_RECOVERY_UNSUPPORTED hatasıyla başarısız oluyor.

Özel uç noktaları ekle

DurableAgentServer bir FastAPI uygulamasıdır. Rotaları, herhangi bir FastAPI uygulamasına eklediğiniz şekilde, çağrı API'siyle birlikte ekleyin:

@app.get("/status")
async def status() -> dict:
    return {"ready": True}

Çerçeve şablonları

agentbricks init İki dizin oluşturur:

  • agent/ Çerçeve kodunuzu içerir: model, istemler ve araçlar.
  • runtime/, çerçeveyi DurableAgentServer’e bağlayan adaptörü ve adaptörün çağrı ve kurtarma işleyicilerini kaydeden giriş noktasını içerir.

Adaptör, her çağrıyı çerçevenin ajan döngüsüne çağrıya çevirir ve framework'ün çıktısını olaylara ve bir sonuca çevirir. Her iki şablon da bir kurtarma işleyicisi kaydeder. LangGraph şablonu, oturum deposundaki son kontrol noktasından devam eder ve OpenAI Agents SDK şablonu isteği aynı oturumda yeniden yürütür. Mevcut bir ajanı eklemek için bir adaptör ve bir DurableAgentServer giriş noktası ekleyin ve agent.toml içindeki [agent] bölümünde server = "agentbricks" ayarlayın.

Azure Databricks'te aracı sunucularını karşılaştırın

Azure Databricks, üç ajan sunucusu sağlar: AgentKit ajan sunucusu ve önceki uygulama şablonlarında kullanılan iki eski ajan sunucusu.

Agent sunucusu Package İstemci API'si Dayanıklı yürütme Tarafından kullanılır
AgentKit (DurableAgentServer, tavsiye edilir) databricks_agentkit, databricks-agentbricks paketinde Invocation API /api/invocations: senkron, akış ve arka plan çalıştırmaları, akış yeniden bağlantısıyla agentbricks deploy sağlayan ve ayrıca bir kurtarma işleyicisi aracılığıyla çökme kurtarma sunan bir Çalışma Zamanı Deposu Agent Bricks CLI ile oluşturduğunuz projeler
LongRunningAgentServer (eski) databricks_ai_bridge.long_running, databricks-ai-bridge[agent-server] paketinde /responses konumunda, arka plan çalıştırmaları ve akış sürdürme ile OpenAI Responses API'si Yapılandırdığınız bir Lakebase veritabanındaki çalıştırma durumu Bir çökme sonrası, yeni bir deneme kesilen denemenin olay günlüğü kaydından çalışmaya devam eder. agent-openai-advanced ve agent-langgraph-advanceduygulama şablonları
MLflow AgentServer (eski) mlflow.genai.agent_server, mlflow paketinde /responses konumunda OpenAI Responses API'de: senkron ve akışlı çalıştırmalar None Temel uygulama şablonları, örneğin agent-openai-agents-sdk

LongRunningAgentServer, MLflow AgentServer öğesini genişletir ve her ikisi de MLflow ResponsesAgent arayüzünü uygulayan ajanlara hizmet eder. Bunlardan birini kullanan bir ajanı dağıtmak ve sürdürmek için bkz. Legacy agent sunucusunu kullanarak Databricks Apps üzerinde ajan çalıştırma. Bu sunuculardan herhangi birinde bir aracıyı sorgulamak için bkz. Azure Databricks'te dağıtılan aracıları sorgulama.

Sınırlamalar

  • Mevcut bir dağıtımın ajan sunucusunu değiştiremezsiniz. DurableAgentServer ile kendi sunucunuz arasında geçiş yapmak için, istediğiniz agentbricks init --server seçeneğiyle yeni bir proje oluşturun ve bunu yeni bir ad altında dağıtın.
  • agent.toml içindeki server alanını değiştirmek mevcut sunucu kodunu DurableAgentServer'ye dönüştürmez.
  • İstek-kullanıcı yetkilendirmesi için server = "agentbricks" gereklidir.

Ek kaynaklar

  • Son güncelleştirme: