diff --git a/src/agentling/agent.py b/src/agentling/agent.py index ed910da..dd1eb72 100644 --- a/src/agentling/agent.py +++ b/src/agentling/agent.py @@ -1,16 +1,17 @@ """The agent loop. -Agent assembles the framework's primitives — a Model, a set of Tools, and a -Memory of typed steps — into an async ReAct loop. A single async generator -(``_run_stream``) drives everything and yields Events; ``run()`` is a thin -dispatcher that either hands back that stream (``stream=True``) or drains it and -returns the final answer (``stream=False``). +Agent ties the framework's primitives (a Model, some Tools, and a Memory of +typed steps) into an async ReAct loop. One async generator, _run_stream, does +the real work and yields Events. run() is a thin dispatcher: it hands back that +stream when stream is True, or drains it and returns the final answer otherwise. """ from __future__ import annotations import asyncio import time +import json +from dataclasses import replace from collections.abc import AsyncIterator, Callable, Sequence from typing import Any @@ -25,6 +26,11 @@ "final_answer tool (or simply reply with plain text)." ) +_LOOP_NUDGE = ( + " Note: you already made this exact tool call and got the same result. " + "Try a different approach." +) + class Agent: """An async tool-calling agent. @@ -48,11 +54,11 @@ def __init__( self.max_steps = max_steps self.step_callbacks = list(step_callbacks) self.parallel_tools = parallel_tools - self.skills = list(skills) # wired in Day 6 + self.skills = list(skills) # accepted but not consumed yet self.memory = Memory() + self._interrupt = asyncio.Event() - # final_answer is always available so the model has an explicit way to stop. all_tools = [*tools, FinalAnswerTool()] self.tools: dict[str, Tool] = {tool.name: tool for tool in all_tools} self._tool_schemas = [tool.to_schema() for tool in all_tools] @@ -67,10 +73,15 @@ def run( ) -> Any: """Run the agent on a task. - With stream=False (default) returns a coroutine resolving to the final - answer string: ``answer = await agent.run(task)``. - With stream=True returns an async iterator of Events: - ``async for event in agent.run(task, stream=True): ...``. + With stream=False (the default) this returns a coroutine that resolves + to the final answer string: + + answer = await agent.run(task) + + With stream=True it returns an async iterator of Events instead: + + async for event in agent.run(task, stream=True): + ... """ events = self._run_stream(task, reset=reset, max_steps=max_steps) if stream: @@ -95,17 +106,34 @@ async def _run_stream( self.memory.add(TaskStep(task=task)) limit = max_steps or self.max_steps + + # Remember the previous step's calls so we can spot an exact repeat. + previous_signature: tuple[tuple[str, str], ...] | None = None + for _ in range(limit): + if self._interrupt.is_set(): + self._interrupt.clear() + yield FinalEvent(answer="Run interrupted.") + return + started = time.monotonic() messages = self.memory.to_messages(self.instructions) response = await self.model.generate(messages, tools=self._tool_schemas) - # Forgiving termination: a turn with no tool calls IS the final answer. + # Forgiving termination: no tool calls means the content is the answer. if not response.tool_calls: self.memory.add(FinalStep(answer=response.content)) yield FinalEvent(answer=response.content, usage=response.usage) return + # Fingerprint this step's calls: (name, canonical JSON args) each. + signature = tuple( + (tc.name, json.dumps(tc.arguments, sort_keys=True)) + for tc in response.tool_calls + ) + looping = signature == previous_signature + previous_signature = signature + # Announce every call, then run them (concurrently or in order). for tool_call in response.tool_calls: yield ToolCallEvent(tool_call=tool_call) @@ -117,6 +145,10 @@ async def _run_stream( else: results = [await self._execute_tool(tc) for tc in response.tool_calls] + # Same calls as the last step: nudge the model to change approach. + if looping: + results = [replace(r, content=r.content + _LOOP_NUDGE) for r in results] + for result in results: yield ToolResultEvent(result=result) @@ -169,8 +201,8 @@ async def _execute_tool(self, tool_call: ToolCall) -> ToolResult: try: output = await tool.call(tool_call.arguments) except Exception as exc: - # Deliberate broad catch: a failing tool becomes an observation the - # model can recover from, never a crash. (Day 5 refines the layers.) + # A failing tool becomes an observation the model can recover from + # rather than a crash, so catching broadly here is intentional. return ToolResult( tool_call_id=tool_call.id, name=tool_call.name, @@ -183,3 +215,11 @@ async def _execute_tool(self, tool_call: ToolCall) -> ToolResult: content=str(output), is_error=False, ) + + def interrupt(self) -> None: + """Request a graceful stop before the next step. + + The current run pauses rather than crashing; resume it later with + run(..., reset=False), which continues from the steps already in memory. + """ + self._interrupt.set() diff --git a/src/agentling/memory.py b/src/agentling/memory.py index 415e72b..b4f26ad 100644 --- a/src/agentling/memory.py +++ b/src/agentling/memory.py @@ -1,11 +1,11 @@ """Typed step records and the agent's memory. The agent loop records each turn as a typed Step (TaskStep, ActionStep, -FinalStep) rather than as raw messages. Steps carry metadata — token usage, -timing, per-tool results and errors — and know how to render themselves back -into the ChatMessage list the model sees. Memory holds the steps, renders the -full conversation via to_messages(), and round-trips to JSON for persistence -and replay. +FinalStep) rather than as raw messages. Steps carry metadata such as token +usage, timing, and per-tool results and errors, and they know how to render +themselves back into the ChatMessage list the model sees. Memory holds the +steps, renders the full conversation via to_messages(), and round-trips to JSON +for persistence and replay. """ from __future__ import annotations diff --git a/src/agentling/models.py b/src/agentling/models.py index ed00f27..2ef4f3b 100644 --- a/src/agentling/models.py +++ b/src/agentling/models.py @@ -1,10 +1,9 @@ """Provider-neutral model layer and the OpenAI-compatible adapter. This module defines the framework's message types (ChatMessage, ToolCall, -Usage) and streaming types (Delta, ToolCallDelta), the Model protocol that the -agent loop depends on, and OpenAIModel — an async adapter for any -OpenAI-compatible chat-completions endpoint, with rate-limit retries, -streaming, and token-usage capture. +Usage) and streaming types (Delta, ToolCallDelta), the Model protocol the agent +loop depends on, and OpenAIModel: an async adapter for any OpenAI-compatible +chat-completions endpoint, with retries, streaming, and token-usage capture. """ from __future__ import annotations @@ -190,10 +189,14 @@ async def stream( yield delta async def _create_with_retry(self, **kwargs: Any) -> Any: - """Call the OpenAI chat-completions API with rate-limit retries. - - Retries cover establishing the request, including opening a stream. - A 429 raised mid-stream (after chunks have been yielded) is not retried. + """Call the OpenAI chat-completions API, retrying transient failures. + + Rate limits, connection/timeout errors, and 5xx responses are retried + with exponential backoff, then re-raised once max_retries is exhausted. + That is the agent's one fatal path. Permanent errors such as a bad + request or bad auth are not retried and propagate immediately. Retries + cover establishing the request, including opening a stream; a failure + that happens mid-stream is not retried. """ total_attempts = max(1, self.max_retries + 1) @@ -202,7 +205,11 @@ async def _create_with_retry(self, **kwargs: Any) -> Any: try: return await self.client.chat.completions.create(**kwargs) - except openai.RateLimitError: + except ( + openai.RateLimitError, + openai.APIConnectionError, # also covers APITimeoutError + openai.InternalServerError, + ): if attempt == total_attempts - 1: raise diff --git a/tests/test_agent.py b/tests/test_agent.py index 34e18bb..8d99496 100644 --- a/tests/test_agent.py +++ b/tests/test_agent.py @@ -189,3 +189,135 @@ async def test_max_steps_forces_a_final_answer() -> None: assert await agent.run("loop forever") == "forced final" assert len(model.calls) == 3 # 2 loop steps + 1 forced tool-free answer + + +def _tool_turn(call_id: str, name: str, **arguments: Any) -> ChatMessage: + """A model turn that requests a single tool call.""" + return _assistant( + tool_calls=[ToolCall(id=call_id, name=name, arguments=arguments)] + ) + + +# --------------------------------------------------------------------------- # +# Loop detector +# --------------------------------------------------------------------------- # +_REPEAT_MARKER = "already made this exact tool call" + + +async def test_loop_detector_nudges_on_repeat() -> None: + model = FakeModel( + [ + _tool_turn("c1", "add", a=2, b=3), + _tool_turn("c2", "add", a=2, b=3), # identical (name, args) + _assistant(content="done"), + ] + ) + agent = Agent(model=model, tools=[add]) + await agent.run("repeat the same call") + + first, second = agent.memory.steps[1], agent.memory.steps[2] + assert isinstance(first, ActionStep) + assert isinstance(second, ActionStep) + assert _REPEAT_MARKER not in first.tool_results[0].content + assert _REPEAT_MARKER in second.tool_results[0].content + + +async def test_loop_detector_ignores_different_args() -> None: + model = FakeModel( + [ + _tool_turn("c1", "add", a=2, b=3), + _tool_turn("c2", "add", a=4, b=5), # different args -> not a loop + _assistant(content="done"), + ] + ) + agent = Agent(model=model, tools=[add]) + await agent.run("two different sums") + + for step in agent.memory.steps: + if isinstance(step, ActionStep): + assert _REPEAT_MARKER not in step.tool_results[0].content + + +# --------------------------------------------------------------------------- # +# Self-heal: unknown tool and always-raising tool +# --------------------------------------------------------------------------- # +async def test_unknown_tool_recovers() -> None: + model = FakeModel([_tool_turn("c1", "ghost"), _assistant(content="recovered")]) + agent = Agent(model=model) # only final_answer is registered + answer = await agent.run("call a missing tool") + + assert answer == "recovered" + action = agent.memory.steps[1] + assert isinstance(action, ActionStep) + assert action.tool_results[0].is_error is True + assert "Unknown tool" in action.tool_results[0].content + + +async def test_always_raising_tool_recovers() -> None: + @tool + def boom() -> str: + """Always raises.""" + raise RuntimeError("nope") + + model = FakeModel( + [ + _tool_turn("c1", "boom"), + _tool_turn("c2", "final_answer", answer="handled"), + ] + ) + agent = Agent(model=model, tools=[boom]) + answer = await agent.run("trigger boom") + + assert answer == "handled" + action = agent.memory.steps[1] + assert isinstance(action, ActionStep) + assert action.tool_results[0].is_error is True + assert "RuntimeError" in action.tool_results[0].content + + +# --------------------------------------------------------------------------- # +# Interruption + resume +# --------------------------------------------------------------------------- # +async def test_interrupt_stops_run_gracefully() -> None: + model = FakeModel( + [_tool_turn("c1", "add", a=1, b=1), _assistant(content="unreached")] + ) + agent = Agent(model=model, tools=[add]) + + fired: list[bool] = [] + + def interrupt_after_first(step: object) -> None: + if not fired: + fired.append(True) + agent.interrupt() + + agent.step_callbacks.append(interrupt_after_first) + + assert await agent.run("loop") == "Run interrupted." + assert isinstance(agent.memory.steps[-1], ActionStep) # paused before a FinalStep + assert len(model.calls) == 1 + + +async def test_interrupt_then_resume_completes() -> None: + model = FakeModel( + [ + _tool_turn("c1", "add", a=1, b=1), + _assistant(content="finished on resume"), + ] + ) + agent = Agent(model=model, tools=[add]) + + fired: list[bool] = [] + + def interrupt_once(step: object) -> None: + if not fired: + fired.append(True) + agent.interrupt() + + agent.step_callbacks.append(interrupt_once) + + assert await agent.run("start") == "Run interrupted." + # reset=False keeps the interrupted run's memory; the loop resumes from it. + assert await agent.run("continue", reset=False) == "finished on resume" + assert len(model.calls) == 2 + assert sum(isinstance(s, TaskStep) for s in agent.memory.steps) == 2 diff --git a/tests/test_models.py b/tests/test_models.py index 0d59087..657e2b1 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -288,3 +288,41 @@ async def always_fails(**kwargs: Any) -> Any: with pytest.raises(openai.RateLimitError): await model.generate([ChatMessage(role="user", content="hi")]) + + +async def test_retry_on_transient_api_error() -> None: + calls = 0 + result = SimpleNamespace( + choices=[ + SimpleNamespace(message=SimpleNamespace(content="ok", tool_calls=None)) + ], + usage=SimpleNamespace(prompt_tokens=1, completion_tokens=1), + ) + + async def flaky(**kwargs: Any) -> Any: + nonlocal calls + calls += 1 + if calls == 1: + raise openai.APIConnectionError.__new__(openai.APIConnectionError) + return result + + model = _make_model(flaky) + msg = await model.generate([ChatMessage(role="user", content="hi")]) + + assert msg.content == "ok" + assert calls == 2 # transient error retried, then succeeded + + +async def test_permanent_error_is_not_retried() -> None: + calls = 0 + + async def always_bad(**kwargs: Any) -> Any: + nonlocal calls + calls += 1 + raise openai.BadRequestError.__new__(openai.BadRequestError) + + model = _make_model(always_bad) + + with pytest.raises(openai.BadRequestError): + await model.generate([ChatMessage(role="user", content="hi")]) + assert calls == 1 # permanent error fails fast — no retry