Rozwijanie agenta LangGraph w AWS Bedrock AgentCore, krok po kroku

Wiele agentów LangGraph dobrze funkcjonuje na laptopie, ale zatrzymuje się na etapie wdrożenia w produkcji. AWS Bedrock AgentCore eliminuje tę lukę dzięki hostowaniu bez serwerów, które umożliwia skalowanie bez konieczności przeprojektowania rozwiązania. Poniższe sekcje opisują, jak przekształcić istniejącego agenta pod AgentCore, utworzyć projekt, uruchomić go lokalnie, przesłać do AWS oraz używać go po wdrożeniu.

Czym jest AgentCore

AgentCore to platforma hostowana przez AWS, na której agenty są wdrażane, skalowane i uruchamiane bez konieczności zarządzania serwerami. Działają na niej takie frameworki jak LangGraph, Strands, CrewAI i LlamaIndex, a także modele od OpenAI, Google i Anthropic, dzięki czemu kod agenta pozostaje niezmieniony. Wbudowane są funkcje izolacji sesji, pamięci trwałej, możliwości obserwacji oraz zarządzania tożsamością.

Wszystko jest sterowane przez AgentCore CLI. Jest to program Node.js, który tworzy szkielet projektu, uruchamia agenta lokalnie oraz wysyła go do AWS przy użyciu CDK działającego w tle.

Wymagania wstępne

• Python 3.12 lub nowsza wersja
• uv lub pip do zarządzania pakietami Pythona
• Node.js 20 lub nowsza wersja, ponieważ CLI jest dystrybuowany przez npm
• AWSCDK zainstalowany globalnie za pomocą npm install -g aws-cdk
• Konto AWS z ustawionymi lokalnie uprawnieniami za pomocą aws configure
• Istniejący kod LangGraph generujący skompilowany StateGraph

Krok 1: instalacja CLI

Jedno narzędzie wiersza poleceń steruje całym procesem: tworzeniem projektu, uruchamianiem aplikacji lokalnie oraz jej wdrażaniem w AWS.

npm install -g @aws/agentcore

Potwierdź, że instalacja się udała:

agentcore --help

Krok 2: dodanie zależności

Dwa pakiety muszą zostać dodane do zależności agenta. bedrock-agentcore to SDK w języku Python, które dostarcza klasę wrapper BedrockAgentCoreApp, natomiast aws-opentelemetry-distro to wersja narzędzia OpenTelemetry dla Pythona stworzona przez AWS.

# pyproject.toml
[project]
name = "my-agent"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "aws-opentelemetry-distro==0.17.0",
 "bedrock-agentcore>=1.6.3",
    "boto3>=1.42.0",
    "langgraph>=1.1.0",
    "langchain-core>=1.2.0",
    # ... your other existing dependencies
]

Zainstaluj zgodnie z procesem pracy typowym dla danego projektu:

uv sync
# or: pip install -e .

Krok 3: napisanie punktu wejścia w main.py

AgentCore oczekuje jednego pliku w języku Python jako punktu wejścia, zawierającego instancję BedrockAgentCoreApp oraz funkcję ozdobioną tagiem @app.entrypoint. Logika LangGraph prawie się nie zmienia – jest po prostu otoczona małym punktem wejścia w czasie wykonywania.

# main.py
from langchain_core.messages import HumanMessage
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from graphs.my_agent_graph import build_my_agent  # your existing graph builder

# 1. Instantiate the AgentCore app and logger
app = BedrockAgentCoreApp()
log = app.logger

# 2. Build your graph at module load time (startup)
#    AgentCore initialises the module once, then handles concurrent invocations.
#    Any failure here will prevent a broken agent from going live.
def create_agent():
    log.info("Initialising agent...")
    graph = build_my_agent()   # returns your compiled LangGraph StateGraph
    log.info("Agent ready")
    return graph
try:
    graph = create_agent()
except Exception as e:
    log.error(f"Critical failure during agent initialisation: {e}")
    raise  # fail fast - don't let a broken agent start serving requests

# 3. Async helper that drives the LangGraph streaming loop
async def run_agent(user_input: str, session_id: str = "default-session") -> str:
    responses = []
    config = {"configurable": {"thread_id": session_id}}
    async for chunk in graph.astream(
        {"messages": [HumanMessage(content=user_input)]},
        config=config,
        stream_mode="values",
    ):
        messages = chunk.get("messages", [])
        if messages:
            last = messages[-1]
            if getattr(last, "type", None) == "ai":
                responses.append(last)
    if not responses:
        return "No response"
    content = getattr(responses[-1], "content", "")
    return content if isinstance(content, str) else str(content)

# 4. The entrypoint - this is what AgentCore calls on every invocation
@app.entrypoint
async def invoke(payload, context):
    try:
        log.info("Invoke received")
        user_input = payload.get("prompt", "")
        session_id = payload.get("session_id", "default-session")
        if not user_input.strip():
            return {"error": "Prompt cannot be empty"}
        response = await run_agent(user_input, session_id)
        return {"response": response}
    except Exception as e:
        log.error(f"Error: {e}")
        return {"error": str(e)}

# 5. Run the app (only executed when running locally using agentcore dev)
if __name__ == "__main__":
    app.run()

Co jest ważne w tym pliku

• BedrockAgentCoreApp() dostarcza środowisku uruchomieniowemu jego „skorupę”: serwer HTTP, ścieżki do sprawdzania stanu oraz ustrukturyzowane logi, wszystko już przygotowane.
• Graph jest budowany na poziomie modułu, raz po imporcie, a nie przy każdym wywołaniu. Dlatego każdy problem przy uruchamianiu pojawia się natychmiast, a uszkodzona aplikacja nigdy nie otrzymuje ruchu.
• @app.entrypoint oznacza funkcję obsługującą przychodzące wywołania. Otrzymuje ona payload – zdekodowany treść w formacie JSON – wraz z context dla żądania AgentCore. To, co musi zwrócić, to zwykły słownik, który można zserializować do JSON. Zwykła strona tekstowa lub klasa niestandardowa nie będą wystarczające.
• Przychodzący session_id staje się thread_id w LangGraph, dzięki czemu MemorySaver przechowuje oddzielną ilość pamięci dla każdej sesji.

Krok 4: utworzenie projektu

Następnie przychodzi kwestia układu projektu. Uruchom asystenta inicjalizacji w czystej folderze głównym i pozwól mu stworzyć szkielet:

agentcore create

Czarodziej prosi o takie szczegóły jak nazwa projektu, język, wersja Pythona oraz plik punktu wejścia. Punkt wejścia i wersja Pythona muszą odpowiadać aplikacji. Następnie w folderze agentcore pojawia się plik agentcore.json, który informuje AgentCore, gdzie znajduje się kod oraz jak go uruchomić.

{
  "name": "MyAgentOnAgentcore",
    "runtimes": [
    {
      "name": "my-agent",
      "build": "CodeZip",
      "entrypoint": "main.py",
      "codeLocation": "app/my-agent/",
      "runtimeVersion": "PYTHON_3_12",
      "networkMode": "PUBLIC",
      "protocol": "HTTP",
	  "envVars": []
	}
  ]
}

Krok 5: umieść kod w miejscu, gdzie szuka AgentCore

Przeciągnij lub skopiuj kod agenta do katalogu określonego przez codeLocation w pliku agentcore.json. Łatwo popełnić błąd, a niewłaściwe umieszczenie kodu sprawia, że proces wdrażania staje się niepotrzebnie trudny. Częstym błędem jest umieszczenie kodu bezpośrednio w folderze agentcore, co może kosztować dużo czasu. Struktura powinna wyglądać w ten sposób:

MyAgentOnAgentcore/
├──agentcore
	└── agentcore.json
├──app/
	└──my-agent/          			 ← codeLocation
		├── main.py                  ← entrypoint (matches agentcore.json)
		├── pyproject.toml           ← or requirements.txt
		├── src/
		│   └── my_agent_graph.py    ← your LangGraph graph builder
		│   └── tools.py             ← your tools, utilities, etc.
		└── .env					 ← your configuration variables

main.py musi znajdować się na najwyższym poziomie codeLocation i dokładnie odpowiadać punktowi wejścia w pliku agentcore.json.

Zmienne środowiskowe

AgentCore może przekazać zmienne środowiskowe do kontenera uruchamiania. Lokalnie najprostszym rozwiązaniem jest plik .env. W środowisku produkcyjnym wartości mogą znajdować się w agentcore.json w sekcji envVars, chociaż plik .env również działa. Tajne dane, takie jak hasła i klucze API, lepiej przechowywać w AWS Secrets Manager i ładować je w czasie uruchamiania; w takim przypadku rola wykonywania AgentCore musi mieć uprawnienia do ich odczytu. Należy pamiętać, że envVars to tablica obiektów JSON, z których każdy zawiera pola name i value.

Krok 6: weryfikacja

Zanim cokolwiek uruchomisz, pozwól interfejsowi CLI sprawdzić konfigurację:

agentcore validate

Proste błędy, takie jak brakujący plik punktu wejścia lub błędna ścieżka w konfiguracji, są wykrywane właśnie tutaj.

Krok 7: uruchom i przetestuj lokalnie

Uruchom lokalny środowisko wykonawcze za pomocą:

agentcore dev

Zapуска lokalny serwer HTTP, który w dużej mierze odpowiada środowisku produkcyjnemu. Gdy wszystko jest poprawnie skonfigurowane, wynik pokazuje, że aplikacja jest gotowa i słucha na lokalnym porcie.

Wywoływanie lokalnego serwera

Wywołanie może odbyć się przez CLI:

agentcore invoke "Summarise last month sales"

lub za pomocą curl:

curl -X POST http://localhost:8080/invocations \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Summarise last month sales", "session_id": "test-001"}'

Dobrą odpowiedzią jest JSON zawierający odpowiedź agenta:

{"response": "Last month, total sales were $1.2M across 3 regions..."}

Krok 8: wdrożenie na AWS

Gdy testy lokalne przyniosą pozytywne wyniki, wdrożenie odbywa się w środowisku AgentCore Runtime:

agentcore deploy

Za kulisami CLI pakuje kod, tworzy bucket S3 do bezpośredniej implementacji kodu, ustala rolę wykonywania IAM oraz implementuje AgentCore Runtime za pomocą CDK. Pierwsza implementacja trwa kilka minut, natomiast kolejne zazwyczaj są szybsze.

Gdy proces się zakończy, sprawdź stan środowiska wykonywania. Wynik pokazuje ARN środowiska wykonywania oraz adres URL HTTP służący do uruchomienia zaimplementowanego agenta.

agentcore status

Krok 9: uruchomienie zainstalowanego agenta

CLI to najprostszy sposób na przetestowanie wersji zainstalowanej:

agentcore invoke --prompt "Who are the top 5 customers by revenue?" --session-id "session-id-with-length-greater-than-or-equal-33"

Aby wywołać funkcję przez HTTP, konieczne jest podpisanie żądania za pomocą AWS SigV4:

curl -X POST "https://bedrock-agentcore.us-east-1.amazonaws.com/runtimes/arn-of-MyAgentOnAgentcore-xxxx/invocations" \
  --user "$AWS_ACCESS_KEY_ID:$AWS_SECRET_ACCESS_KEY" \
  --aws-sigv4 "aws:amz:us-east-1:bedrock-agentcore" \
  -H "Content-Type: application/json" \
  -H "x-amzn-bedrock-agentcore-runtime-session-id: session-id-with-length-greater-than-or-equal-33" \
  -d '{"prompt": "Who are the top 5 customers by revenue?"}'

Użyj adresu URL HTTP z agentcore status.

Dla wywołań z innego kodu Pythona SDK jest zazwyczaj najlepszą opcją. Przechowaj ARN w czasie wykonywania w zmiennej środowiskowej, przekaż odpowiedni prompt oraz wysyłaj poprzedni identyfikator sesji przy każdym kolejnym wywołaniu.

# invoke_agent.py
import json
import boto3
import os
from dotenv import load_dotenv

load_dotenv()

client = boto3.client("bedrock-agentcore", region_name="us-east-1")
runtime_arn = os.getenv("AGENTCORE_RUNTIME_ARN")

def invoke(prompt: str, session_id: str) -> str:
    payload = json.dumps({"prompt": prompt, "session_id": session_id})
    response = client.invoke_agent_runtime(
        agentRuntimeArn=runtime_arn,
        payload=payload,
        contentType="application/json"
    )
    body = json.loads(response["response"].read().decode())
    return body["response"]

if __name__ == "__main__":
    session_id = ""
    print("Ready. Type 'exit' to quit.")

    while True:
        prompt = input("Your question: ").strip()
        if prompt == "exit":
            break

        if not prompt:
            continue

        params = {
            "agentRuntimeArn": runtime_arn,
            "payload": json.dumps({"prompt": prompt}),
            "contentType": "application/json"
        }
        if session_id:
            params["runtimeSessionId"] = session_id

        try:
            response = client.invoke_agent_runtime(**params)
            body = json.loads(response["response"].read().decode())
            session_id = response.get("runtimeSessionId", session_id)
            print(f"Agent: {body['response']}\n")
        except Exception as e:
            print(f"Error: {e}")

Ustaw AGENTCORE_RUNTIME_ARN w pliku .env:

AGENTCORE_RUNTIME_ARN=arn:aws:bedrock-agentcore:us-east-1:123456789012:agent-runtime/MyAgentOnAgentcore-xxxx

Zachowywanie stanu sesji pomiędzy wywołaniami

Spójrz na runtimeSessionId w przykładzie boto3. AgentCore zwraca go przy każdej odpowiedzi. Przesłanie go ponownie w następnym żądaniu umożliwia kierowanie dalszych działań do tego samego kontekstu sesji, a w połączeniu z MemorySaver z LangGraph agent zachowuje pełną pamięć rozmowy pomimo oddzielnych wywołań HTTP.

Częste pułapki

• Błędy startowe są celowe. Gdy create_agent() wywołuje błąd podczas importu, AgentCore nie uruchamia środowiska wykonawczego. Takie zachowanie chroni przed uruchomieniem błędnie skonfigurowanego agenta, np. takiego bez wymaganych uprawnień lub z błędną adresem bazy danych, który mógłby działać bez żadnych problemów.
• Należy dodać uprawnienia. Rola wykonywania tworzona przez AgentCore wymaga wyraźnego dostępu do każdej usługi AWS używanej przez agenta, w tym do Secrets Manager i S3. Należy dodać te uprawnienia po pierwszym wdrożeniu.
• Wersje muszą być zgodne. Wersja Pythona określona w pliku pyproject.toml musi być zgodna z ustawieniem środowiska wykonywania w pliku agentcore.json.
• Należy zwracać zwykłe słowniki. Jeśli punkt wejścia zwraca ciąg znaków lub własny obiekt, środowisko wykonywania odrzuca to.

Co faktycznie się zmienia

Aby przejść od lokalnego agenta LangGraph do AgentCore, potrzebne są jedynie kilka zmian: należy zainstalować dwa dodatkowe pakiety, otoczyć graf w BedrockAgentCoreApp, stworzyć szkielet projektu, przetestować go lokalnie, a następnie wdrożyć i uruchomić. Graf, narzędzia, wywołania modelu oraz konfiguracja pamięci pozostają w większości nietknięte. AgentCore obsługuje warstwę infrastruktury, natomiast LangGraph nadal odpowiada za logikę agenta.