Migration von Google ADK

from google.adk.agents import LlmAgent, SequentialAgent
from google.adk.tools import FunctionTool

def web_search(query: str) -> dict:
    """Searches the web for information on the given query."""
    # Implementation
    return {"results": ["result 1", "result 2"]}

search_agent = LlmAgent(
    name="searcher",
    model="gemini-2.5-flash",
    instruction="Search for: {query}",
    tools=[web_search],
    output_key="search_results",
)

analyst = LlmAgent(
    name="analyst",
    model="gemini-2.5-pro",
    instruction=(
        "Analyze these results: {search_results}. "
        "Write a comprehensive summary with key findings."
    ),
    output_key="summary",
)

pipeline = SequentialAgent(
    name="research_assistant",
    sub_agents=[search_agent, analyst],
)

# Pipeline ausführen

from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

runner = Runner(
    agent=pipeline,
    app_name="research_app",
    session_service=InMemorySessionService(),
)

session = runner.session_service.create_session(
    app_name="research_app", user_id="user1"
)

events = runner.run(
    user_id="user1",
    session_id=session.id,
    new_message=types.Content(
        parts=[types.Part(text="durable AI workflow orchestration")]
    ),
)

for event in events:
    if event.is_final_response():
        print(event.content.parts[0].text)

id: research-assistant
version: "0.1.0"

nodes:
  search:
    type: tool
    server: brave-search
    tool: web_search
    arguments:
      query: "{{ state.query }}"
    output_key: search_results
    retry:
      max_attempts: 3
      backoff: exponential
    next: analyze

  analyze:
    type: model
    model: claude-sonnet-4-6
    prompt: |
      Analyze these results: {{ state.search_results }}.
      Write a comprehensive summary with key findings.
    output_key: summary
    next: evaluate

  evaluate:
    type: eval
    scorers:
      - type: llm_judge
        model: claude-haiku-4-5-20251001
        rubric: "Is the summary accurate, well-structured, and comprehensive?"
    fail_under: 4.0

from openai import OpenAI
from pydantic import BaseModel
from jamjet import Workflow

client = OpenAI()

class State(BaseModel):
    query: str
    search_results: str = ""
    summary: str = ""

wf = Workflow("research-assistant")

@wf.state
class ResearchState(State):
    pass

@wf.step
async def search(state: ResearchState) -> ResearchState:
    # In production: type: tool + MCP server handles this
    results = f"[results for: {state.query}]"
    return state.model_copy(update={"search_results": results})

@wf.step
async def analyze(state: ResearchState) -> ResearchState:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": (
                "You are an expert research analyst. "
                "Write a comprehensive summary with key findings."
            )},
            {"role": "user", "content": (
                f"Analyze these results: {state.search_results}"
            )},
        ],
    )
    return state.model_copy(update={"summary": resp.choices[0].message.content or ""})

result = wf.run_sync(ResearchState(query="durable AI workflow orchestration"))
print(result.state.summary)

# ADK: alles geht

session.state["results"] = 42      # sollte das eine Liste sein?
session.state["resutls"] = [...]   # Tippfehler — kein Fehler, stiller Bug

# JamJet: sofort erkannt

class State(BaseModel):
    results: list[str] = []

state.model_copy(update={"results": 42})  # ValidationError

# Jede frühere Ausführung wiederholen

jamjet replay exec-abc123

# Von einem bestimmten Schritt abzweigen, um einen anderen Pfad zu testen

jamjet replay exec-abc123 --fork-at analyze

# ADK: Tool lebt im Code Ihres Agents

# def web_search(query: str) -> dict:

#     """Durchsucht das Web."""

# JamJet: Tool ist ein unabhängiger MCP-Server

nodes:
  search:
    type: tool
    server: brave-search      # jeder MCP-kompatible Server
    tool: web_search
    arguments:
      query: "{{ state.query }}"
    retry:
      max_attempts: 3
      backoff: exponential

jamjet eval run evals/research.jsonl \
  --workflow research-assistant \
  --fail-under 0.85

nodes:
  fast-search:
    type: model
    model: claude-haiku-4-5-20251001     # Anthropic
    # ...

  deep-analysis:
    type: model
    model: gpt-4o              # OpenAI
    # ...

  local-draft:
    type: model
    model: ollama/llama3       # Local
    # ...

pip install jamjet