diff --git a/examples-agents/README.md b/examples-agents/README.md index a2654c80..03e20d4c 100644 --- a/examples-agents/README.md +++ b/examples-agents/README.md @@ -47,7 +47,7 @@ To talk with an agent from a terminal, as it works, run the agents' command line (it needs the `client` extra: `pip install "restate-sdk[client]"`): ```sh -python -m restate.agents chat assistant/alice # what you type goes to the agent; /help for the commands +python -m restate.agents chat openai_assistant/alice # what you type goes to the agent; /help for the commands ``` It shows the conversation so far, then each message, tool call and wait as it diff --git a/examples-agents/basic/adk_agent.py b/examples-agents/basic/adk_agent.py index 1c3ef442..a666583c 100644 --- a/examples-agents/basic/adk_agent.py +++ b/examples-agents/basic/adk_agent.py @@ -6,8 +6,8 @@ GOOGLE_API_KEY=... RESTATE_AGENT_MODEL=google/ uv run --extra adk --extra serve examples-agents/basic/adk_agent.py npx @restatedev/restate-server # Restate, in another terminal npx @restatedev/restate -y deployments register --force http://localhost:9080 - curl localhost:8080/assistant/alice/chat -d 'What is the weather in Lisbon?' - curl localhost:8080/assistant/alice/messages --json '{}' # the conversation + curl localhost:8080/adk_assistant/alice/chat -d 'What is the weather in Lisbon?' + curl localhost:8080/adk_assistant/alice/messages --json '{}' # the conversation The model is RESTATE_AGENT_MODEL, a Gemini model (``google/``). For another provider, pass an ADK model of your own: ``model=LiteLlm(...)``. @@ -25,7 +25,7 @@ async def get_weather(city: str) -> str: return f"Sunny and 21°C in {city}" -assistant = adk.agent("assistant", system="Answer briefly.", tools=[get_weather]) +assistant = adk.agent("adk_assistant", system="Answer briefly.", tools=[get_weather]) if __name__ == "__main__": diff --git a/examples-agents/basic/create_agent.py b/examples-agents/basic/create_agent.py index fe1fe413..80189abe 100644 --- a/examples-agents/basic/create_agent.py +++ b/examples-agents/basic/create_agent.py @@ -1,7 +1,7 @@ """ LangChain's ``create_agent`` as a Restate agent: an order desk, each conversation in its key's database. Its tool is Restate's: a call to the -``orders`` service, made once per tool call even if the process crashes +``shipping`` service, made once per tool call even if the process crashes mid-turn:: OPENAI_API_KEY=... uv run --extra langgraph --extra langchain --extra serve --with langchain-openai examples-agents/basic/create_agent.py @@ -21,10 +21,10 @@ import restate from restate.agents import call, langgraph -orders = restate.Service("orders") +shipping = restate.Service("shipping") -@orders.handler() +@shipping.handler() async def status(_ctx: restate.Context, order: str) -> str: return f"Order {order} was delivered yesterday." @@ -38,4 +38,4 @@ async def status(_ctx: restate.Context, order: str) -> str: if __name__ == "__main__": - asyncio.run(restate.serve(restate.app([orders, desk]))) + asyncio.run(restate.serve(restate.app([shipping, desk]))) diff --git a/examples-agents/basic/langgraph_agent.py b/examples-agents/basic/langgraph_agent.py index 337809c5..78573bd5 100644 --- a/examples-agents/basic/langgraph_agent.py +++ b/examples-agents/basic/langgraph_agent.py @@ -6,8 +6,8 @@ OPENAI_API_KEY=... uv run --extra langgraph --extra serve --with langchain-openai examples-agents/basic/langgraph_agent.py npx @restatedev/restate-server # Restate, in another terminal npx @restatedev/restate -y deployments register --force http://localhost:9080 - curl localhost:8080/assistant/alice/chat -d 'What is the weather in Lisbon?' - curl localhost:8080/assistant/alice/messages --json '{}' # the conversation + curl localhost:8080/langgraph_assistant/alice/chat -d 'What is the weather in Lisbon?' + curl localhost:8080/langgraph_assistant/alice/messages --json '{}' # the conversation The model is ``openai/gpt-6-sol`` unless RESTATE_AGENT_MODEL names another: ``anthropic/`` takes ``langchain-anthropic`` (and @@ -26,7 +26,7 @@ async def get_weather(city: str) -> str: return f"Sunny and 21°C in {city}" -assistant = langgraph.agent("assistant", system="Answer briefly.", tools=[get_weather]) +assistant = langgraph.agent("langgraph_assistant", system="Answer briefly.", tools=[get_weather]) if __name__ == "__main__": diff --git a/examples-agents/basic/openai_agent.py b/examples-agents/basic/openai_agent.py index 33b5331f..94a2e9ed 100644 --- a/examples-agents/basic/openai_agent.py +++ b/examples-agents/basic/openai_agent.py @@ -6,12 +6,12 @@ OPENAI_API_KEY=... uv run --extra openai --extra serve examples-agents/basic/openai_agent.py npx @restatedev/restate-server # Restate, in another terminal npx @restatedev/restate -y deployments register --force http://localhost:9080 - curl localhost:8080/assistant/alice/chat -d 'What is the weather in Lisbon?' - curl localhost:8080/assistant/alice/messages --json '{}' # the conversation + curl localhost:8080/openai_assistant/alice/chat -d 'What is the weather in Lisbon?' + curl localhost:8080/openai_assistant/alice/messages --json '{}' # the conversation The model is ``openai/gpt-6-sol`` unless RESTATE_AGENT_MODEL names another OpenAI model. To bring an ``Agent`` of your own, with handoffs and guardrails, -pass it as ``openai.agent("assistant", agent=Agent(...))``. +pass it as ``openai.agent("openai_assistant", agent=Agent(...))``. """ import asyncio @@ -26,7 +26,7 @@ async def get_weather(city: str) -> str: return f"Sunny and 21°C in {city}" -assistant = openai.agent("assistant", system="Answer briefly.", tools=[get_weather]) +assistant = openai.agent("openai_assistant", system="Answer briefly.", tools=[get_weather]) if __name__ == "__main__": diff --git a/examples-agents/basic/pydantic_ai_agent.py b/examples-agents/basic/pydantic_ai_agent.py index ef4c22a0..9413cd68 100644 --- a/examples-agents/basic/pydantic_ai_agent.py +++ b/examples-agents/basic/pydantic_ai_agent.py @@ -6,8 +6,8 @@ OPENAI_API_KEY=... uv run --extra pydantic_ai --extra serve --with 'pydantic-ai-slim[openai]' examples-agents/basic/pydantic_ai_agent.py npx @restatedev/restate-server # Restate, in another terminal npx @restatedev/restate -y deployments register --force http://localhost:9080 - curl localhost:8080/assistant/alice/chat -d 'What is the weather in Lisbon?' - curl localhost:8080/assistant/alice/messages --json '{}' # the conversation + curl localhost:8080/pydantic_ai_assistant/alice/chat -d 'What is the weather in Lisbon?' + curl localhost:8080/pydantic_ai_assistant/alice/messages --json '{}' # the conversation The model is ``openai/gpt-6-sol`` unless RESTATE_AGENT_MODEL names another, of any of Pydantic AI's providers: ``anthropic/`` takes @@ -26,7 +26,7 @@ async def get_weather(city: str) -> str: return f"Sunny and 21°C in {city}" -assistant = pydantic_ai.agent("assistant", system="Answer briefly.", tools=[get_weather]) +assistant = pydantic_ai.agent("pydantic_ai_assistant", system="Answer briefly.", tools=[get_weather]) if __name__ == "__main__": diff --git a/examples-agents/intermediate/loop.py b/examples-agents/intermediate/loop.py index dddbb0e8..4239d61a 100644 --- a/examples-agents/intermediate/loop.py +++ b/examples-agents/intermediate/loop.py @@ -7,15 +7,15 @@ OPENAI_API_KEY=... uv run --extra serve --with openai examples-agents/intermediate/loop.py npx @restatedev/restate-server # Restate, in another terminal npx @restatedev/restate -y deployments register --force http://localhost:9080 - curl localhost:8080/refunds/alice/chat -d 'Order 103 arrived broken, can I get my $30 back?' + curl localhost:8080/refund_loop/alice/chat -d 'Order 103 arrived broken, can I get my $30 back?' The agent waits for a manager to approve the refund. In another terminal, approve it:: - curl localhost:8080/refunds/alice/signal --json '{"name": "approval", "value": "approved"}' + curl localhost:8080/refund_loop/alice/signal --json '{"name": "approval", "value": "approved"}' Or talk with the agent, and approve, from the agents' command line: -``python -m restate.agents chat refunds/alice``, then ``/signal approval approved``. +``python -m restate.agents chat refund_loop/alice``, then ``/signal approval approved``. The loop is plain Python, and Restate makes it durable: @@ -204,9 +204,9 @@ async def run_call(call: Call) -> None: return await answer_once(ctx, text, MESSAGES, run) -refunds = AgentObject("refunds") -refunds.chat(answer, items=MESSAGES) +refund_loop = AgentObject("refund_loop") +refund_loop.chat(answer, items=MESSAGES) if __name__ == "__main__": - asyncio.run(restate.serve(restate.app([refunds]))) + asyncio.run(restate.serve(restate.app([refund_loop]))) diff --git a/examples-agents/intermediate/router.py b/examples-agents/intermediate/router.py index 98f7f661..42838e53 100644 --- a/examples-agents/intermediate/router.py +++ b/examples-agents/intermediate/router.py @@ -10,8 +10,8 @@ --with langchain-openai --with 'pydantic-ai-slim[openai]' examples-agents/intermediate/router.py npx @restatedev/restate-server # Restate, in another terminal npx @restatedev/restate -y deployments register --force http://localhost:9080 - curl localhost:8080/desk/alice/route --json '"I was charged twice for order 103"' - curl localhost:8080/desk/alice/route --json '"The app crashes when I open my orders"' + curl localhost:8080/front_desk/alice/route --json '"I was charged twice for order 103"' + curl localhost:8080/front_desk/alice/route --json '"The app crashes when I open my orders"' curl localhost:8080/billing/alice/messages --json '{}' # what billing was asked, and answered """ @@ -31,10 +31,10 @@ system="You handle technical problems with the app. Ask for what you need to know, one question at a time.", ) -desk = sessions.SessionObject("desk") +front_desk = sessions.SessionObject("front_desk") -@desk.handler +@front_desk.handler async def route(ctx: sessions.SessionContext, message: str) -> str: # Asked once: a retried attempt gets the same answer. team = await pydantic_ai.prompt( @@ -49,4 +49,4 @@ async def route(ctx: sessions.SessionContext, message: str) -> str: if __name__ == "__main__": - asyncio.run(restate.serve(restate.app([desk, billing, tech]))) + asyncio.run(restate.serve(restate.app([front_desk, billing, tech]))) diff --git a/python/restate/agents/_conversation.py b/python/restate/agents/_conversation.py index 49184ca8..e75d5258 100644 --- a/python/restate/agents/_conversation.py +++ b/python/restate/agents/_conversation.py @@ -294,6 +294,22 @@ def summary_items(items: Items, summary: str) -> list[Item]: return [items.prompt(SUMMARY + summary), items.reply("Understood.")] +def load_item(items: Items, stored: str) -> Item: + """Return the item stored as ``stored``, or fail for good: a retry would read the same row again. + + A row that ``items`` cannot read was stored by another framework's agent of the same name, or by another + version of this framework, say. + """ + try: + return items.load(stored) + except Exception as error: + raise TerminalError( + f"This conversation holds a message that this agent cannot read (stored by an agent of another " + f"framework under the same name, or by another version of its framework?): {type(error).__name__}: " + f"{error}" + ) from error + + @dataclass(frozen=True) class Start: """Where a conversation starts: its first position, the last ``seq`` given, and the summary of the earliest turns.""" @@ -370,7 +386,7 @@ async def latest_snapshot(db: DatabaseReader, items: Items) -> Snapshot | None: row = await db.get(f"SELECT covers, turn, items FROM {SNAPSHOTS} ORDER BY covers DESC, id DESC LIMIT 1") if row is None: return None - held = [items.load(str(stored)) for stored in json.loads(str(row["items"]))] + held = [load_item(items, str(stored)) for stored in json.loads(str(row["items"]))] return _snapshot_of(items, int(typing.cast(int, row["covers"])), _int(row["turn"]), held) @@ -948,7 +964,7 @@ async def load( if whole: stored = await db.values(f"SELECT n, item, reply, agent FROM {STEPS} WHERE n > ? ORDER BY n", [whole[0].n]) steps = [ - Entry(int(typing.cast(int, n)), items.load(str(item)), reply=_int(reply), agent=_str(agent)) + Entry(int(typing.cast(int, n)), load_item(items, str(item)), reply=_int(reply), agent=_str(agent)) for n, item, reply, agent in stored ] entries = sorted([*whole, *steps], key=_position) @@ -976,7 +992,7 @@ async def _from_snapshot(cls, db: Database, items: Items, start: Start, last: in if whole: stored = await db.values(f"SELECT n, item, reply, agent FROM {STEPS} WHERE n > ? ORDER BY n", [whole[0].n]) steps = [ - Entry(int(typing.cast(int, n)), items.load(str(item)), reply=_int(reply), agent=_str(agent)) + Entry(int(typing.cast(int, n)), load_item(items, str(item)), reply=_int(reply), agent=_str(agent)) for n, item, reply, agent in stored ] entries = sorted([*whole, *steps], key=_position) @@ -1245,7 +1261,7 @@ async def _transcript(db: DatabaseReader, items: Items, first: int, bounded: tup at = int(typing.cast(int, n)) entry = Entry( at, - items.load(str(item)), + load_item(items, str(item)), str(prompt), _str(answer), agent=_str(agent), diff --git a/python/restate/agents/_messages.py b/python/restate/agents/_messages.py index a9ba5d73..303fdd88 100644 --- a/python/restate/agents/_messages.py +++ b/python/restate/agents/_messages.py @@ -30,7 +30,7 @@ from dataclasses import dataclass, field from restate.agents._api import ChatMessage, Messages, MessagesRequest, Part, Role, ToolPart, text_part -from restate.agents._conversation import SCHEMA, STEPS, TURNS, Items, last_of, start_of +from restate.agents._conversation import SCHEMA, STEPS, TURNS, Items, last_of, load_item, start_of from restate.sessions import DatabaseReader @@ -50,7 +50,7 @@ def render(items: Items, rows: typing.Sequence[typing.Sequence[typing.Any]]) -> """Each tool call's message, by the call's id.""" for n, seq, stored in rows: position, changed = int(n), int(seq) - said, calls, results = items.read(items.load(str(stored))) + said, calls, results = items.read(load_item(items, str(stored))) if said is not None: role, text = said messages.append(_Message(position, "user" if role == "user" else "assistant")) @@ -187,7 +187,7 @@ def _summarized(items: Items, rows: typing.Sequence[typing.Sequence[typing.Any]] """ messages: list[_Message] = [] for n, seq, stored, answer, answered_at, answered_seq in rows: - said = items.read(items.load(str(stored))).said + said = items.read(load_item(items, str(stored))).said text = said[1] if said else "" messages.append(_Message(int(n), "user", [text_part(text)] if text.strip() else [], int(seq))) if answer is not None and answered_at is not None: diff --git a/python/restate/agents/pydantic_ai/_agent.py b/python/restate/agents/pydantic_ai/_agent.py index 0e1fc615..57f5edbf 100644 --- a/python/restate/agents/pydantic_ai/_agent.py +++ b/python/restate/agents/pydantic_ai/_agent.py @@ -131,6 +131,7 @@ ) from pydantic_core import to_json, to_jsonable_python +import pydantic_ai from pydantic_ai import ( Agent, DeferredToolRequests, @@ -165,6 +166,12 @@ from restate.exceptions import TerminalError from restate.sessions import SessionContext +# Pydantic AI shows a banner on a process's first agent run, unless an installed +# package started the run. Ours start in a Restate handler, behind the event +# loop, where it cannot tell: turn it off, as it advises a library that runs +# agents on its users' behalf. +pydantic_ai.BANNER_ENABLED = False + M = typing.TypeVar("M", bound=BaseModel) AnyModel: typing.TypeAlias = str | Model diff --git a/python/restate/agents/pydantic_ai/_stores.py b/python/restate/agents/pydantic_ai/_stores.py index 95008fc6..b72be899 100644 --- a/python/restate/agents/pydantic_ai/_stores.py +++ b/python/restate/agents/pydantic_ai/_stores.py @@ -42,7 +42,7 @@ from pydantic_ai_harness.planning import PlanItem, TaskStatus from pydantic_ai_harness.step_persistence import RunRecord -from restate.agents._conversation import STEPS, TURNS, Item, Items +from restate.agents._conversation import STEPS, TURNS, Item, Items, load_item from restate.agents._tools import current_call from restate.sessions import SessionContext from restate.sessions.internal import Schema, definition_of @@ -322,7 +322,8 @@ async def run_history(self, *, run_id: str) -> list[ModelMessage]: steps = await db.values(f"SELECT item FROM {STEPS} WHERE n > ? ORDER BY n", [n]) else: steps = await db.values(f"SELECT item FROM {STEPS} WHERE n > ? AND n < ? ORDER BY n", [n, end]) - loaded: list[Item] = [self.items.load(str(row["item"]))] + [self.items.load(str(item)) for [item] in steps] + stored = [str(row["item"])] + [str(item) for [item] in steps] + loaded: list[Item] = [load_item(self.items, item) for item in stored] return typing.cast(list[ModelMessage], loaded) diff --git a/tests/agents_conversation.py b/tests/agents_conversation.py index 80853c24..76ee2eb1 100644 --- a/tests/agents_conversation.py +++ b/tests/agents_conversation.py @@ -228,6 +228,38 @@ async def failure(text: str, invocation: Any = None) -> tuple[str, int]: assert await failure("gone", gone) == (ENDED, 500) +class _OtherFramework(Items): + """Another framework's items, which have a ``kind``: it cannot read the made-up framework's.""" + + def load(self, stored: str) -> Item: + item = json.loads(stored) + if "kind" not in item: + raise ValueError("an item has a kind") + return item + + +async def test_a_conversation_this_agent_cannot_read_fails_for_good(): + """An agent of the same name on another framework fails with a terminal error: a retry would read the same row.""" + obj = sessions.SessionObject("asker") + + async def run(conversation: Conversation) -> str: + await conversation.turn.append_reply({"role": "assistant", "text": "hello"}) + return "hello" + + @obj.handler + async def ask(ctx: sessions.SessionContext, text: str) -> str: + return await answer_once(ctx, text, ITEMS, run) + + @obj.handler + async def ask_other(ctx: sessions.SessionContext, text: str) -> str: + return await answer_once(ctx, text, _OtherFramework(), run) + + backend = Backend() + assert await invoke(obj, "ask", Host(backend, service="asker", answers=PUBLISHED), request(), "hi") == "hello" + with pytest.raises(restate.TerminalError, match="cannot read .* ValueError: an item has a kind"): + await invoke(obj, "ask_other", Host(backend, service="asker", answers=PUBLISHED), request(), "hi") + + async def test_a_dropped_reply_takes_what_followed_it_and_each_item_is_read_once(): reads: list[Item] = [] diff --git a/tests/examples_agents.py b/tests/examples_agents.py index e37a90fb..dda25c92 100644 --- a/tests/examples_agents.py +++ b/tests/examples_agents.py @@ -157,12 +157,12 @@ async def test_basic_langgraph_agent(): """The agent serves, with its companion; a message needs a model's provider, and is left to a real run.""" pytest.importorskip("langgraph") async with serving("basic/langgraph_agent.py") as ingress: - assert (await ingress.post("assistant/alice/messages", {})).json()["messages"] == [] - assert (await ingress.post("assistant/alice/status")).json() == {"answering": None, "waiting": []} + assert (await ingress.post("langgraph_assistant/alice/messages", {})).json()["messages"] == [] + assert (await ingress.post("langgraph_assistant/alice/status")).json() == {"answering": None, "waiting": []} async def test_basic_create_agent(monkeypatch: pytest.MonkeyPatch): - """create_agent's model (scripted here) calls the orders service through the Restate tool, and answers.""" + """create_agent's model (scripted here) calls the shipping service through the Restate tool, and answers.""" pytest.importorskip("langchain.agents") import langchain.agents.factory # pylint: disable=import-outside-toplevel from langchain_core.language_models import BaseChatModel # pylint: disable=import-outside-toplevel @@ -182,7 +182,7 @@ def _generate(self, messages: list[BaseMessage], *args: Any, **kwargs: Any) -> C if isinstance(last, ToolMessage): reply = AIMessage(str(last.text)) else: - reply = AIMessage("", tool_calls=[{"id": "c1", "name": "orders_status", "args": {"input": "103"}}]) + reply = AIMessage("", tool_calls=[{"id": "c1", "name": "shipping_status", "args": {"input": "103"}}]) return ChatResult(generations=[ChatGeneration(message=reply)]) # The example names an OpenAI model; the scripted one stands in for it, with no API key. @@ -196,8 +196,8 @@ async def test_basic_pydantic_ai_agent(): """The agent serves, with its companion; a message needs a model's provider, and is left to a real run.""" pytest.importorskip("pydantic_ai") async with serving("basic/pydantic_ai_agent.py") as ingress: - assert (await ingress.post("assistant/alice/messages", {})).json()["messages"] == [] - assert (await ingress.post("assistant/alice/status")).json() == {"answering": None, "waiting": []} + assert (await ingress.post("pydantic_ai_assistant/alice/messages", {})).json()["messages"] == [] + assert (await ingress.post("pydantic_ai_assistant/alice/status")).json() == {"answering": None, "waiting": []} async def test_basic_adk_agent(monkeypatch: pytest.MonkeyPatch): @@ -205,16 +205,16 @@ async def test_basic_adk_agent(monkeypatch: pytest.MonkeyPatch): pytest.importorskip("google.adk") monkeypatch.setenv("RESTATE_AGENT_MODEL", "google/gemini-x") # an ADK agent needs a model async with serving("basic/adk_agent.py") as ingress: - assert (await ingress.post("assistant/alice/messages", {})).json()["messages"] == [] - assert (await ingress.post("assistant/alice/status")).json() == {"answering": None, "waiting": []} + assert (await ingress.post("adk_assistant/alice/messages", {})).json()["messages"] == [] + assert (await ingress.post("adk_assistant/alice/status")).json() == {"answering": None, "waiting": []} async def test_basic_openai_agent(): """The agent serves, with its companion; a message needs a model's provider, and is left to a real run.""" pytest.importorskip("agents") async with serving("basic/openai_agent.py") as ingress: - assert (await ingress.post("assistant/alice/messages", {})).json()["messages"] == [] - assert (await ingress.post("assistant/alice/status")).json() == {"answering": None, "waiting": []} + assert (await ingress.post("openai_assistant/alice/messages", {})).json()["messages"] == [] + assert (await ingress.post("openai_assistant/alice/status")).json() == {"answering": None, "waiting": []} async def test_intermediate_support(): @@ -368,20 +368,20 @@ def counted(order: str, amount: float) -> str: module.TOOLS["refund"] = module.tool(counted, name="refund", description="Refund an order.") async with serving("intermediate/loop.py", setup) as ingress: - chat = asyncio.create_task(ingress.post("refunds/alice/chat", "Order 103 arrived broken")) + chat = asyncio.create_task(ingress.post("refund_loop/alice/chat", "Order 103 arrived broken")) waiting: list[dict[str, Any]] = [] for _ in range(100): - if waiting := (await ingress.post("refunds/alice/pending")).json(): + if waiting := (await ingress.post("refund_loop/alice/pending")).json(): break await asyncio.sleep(0.1) assert [w["name"] for w in waiting] == ["approval"] and not refunded # it waits, and has not refunded - await ingress.post("refunds/alice/signal", {"name": "approval", "value": "approved"}) + await ingress.post("refund_loop/alice/signal", {"name": "approval", "value": "approved"}) assert (await chat).json() == "Done: Refunded $30 for order 103." assert refunded == ["103"] and crashed == [True] and unanswered == [True] # The model was asked once for each step (and again after the crash that cut its connection), not after the # crash that came once its final reply was stored. assert asked == ["user", asked[1], "c2", "c3", "c3"] and asked[1] in ("c1", "c1b") - shown = (await ingress.post("refunds/alice/messages", {})).json()["messages"] + shown = (await ingress.post("refund_loop/alice/messages", {})).json()["messages"] tools = [(p["name"], p["output"]) for p in shown[1]["parts"]] assert tools == [ ("order_status", "Order 103 was delivered yesterday."),