diff --git a/README.md b/README.md index ac70b0d..5b39e57 100644 --- a/README.md +++ b/README.md @@ -131,35 +131,41 @@ Replace `REGION` and `ACCOUNT_ID`. `dynamodb:Scan` is only needed for the GraphR ```python from dynavec import Dynavec, DynavecConfig, Document -from dynavec.embeddings import OpenAIEmbedder # or Gemini / Bedrock / SentenceTransformer +from dynavec.embeddings import OpenAIEmbedder # or Gemini / Bedrock / SentenceTransformer cfg = DynavecConfig( - vector_bucket="my-vectors", # S3 vector bucket - index="docs", # vector index - table="dynavec_docs", # DynamoDB table + vector_bucket="my-vectors", # S3 vector bucket + index="docs", # vector index + table="dynavec_docs", # DynamoDB table dimension=1536, distance_metric="cosine", region="us-east-1", - auto_provision=True, # create bucket + index + table if missing + auto_provision=True, # create bucket + index + table if missing ) db = Dynavec(cfg, embedder=OpenAIEmbedder(model="text-embedding-3-small")) db.upsert( [ - Document(id="a", text="The mitochondria is the powerhouse of the cell.", - metadata={"topic": "biology", "year": 2021}), - Document(id="b", text="Rockets reach orbit at roughly 28,000 km/h.", - metadata={"topic": "space", "year": 2023}), + Document( + id="a", + text="The mitochondria is the powerhouse of the cell.", + metadata={"topic": "biology", "year": 2021}, + ), + Document( + id="b", + text="Rockets reach orbit at roughly 28,000 km/h.", + metadata={"topic": "space", "year": 2023}, + ), ], - auto_metadata=True, # also attach hash/timestamp/word counts + auto_metadata=True, # also attach hash/timestamp/word counts ) hits = db.search( "how do cells make energy?", top_k=3, - filter={"topic": "biology"}, # S3 Vectors metadata pre-filter - rerank="mmr", # diversity-aware reranking + filter={"topic": "biology"}, # S3 Vectors metadata pre-filter + rerank="mmr", # diversity-aware reranking ) for h in hits: print(h.score, h.id, h.text) @@ -190,10 +196,12 @@ Expire ephemeral documents, session memories, or cache entries automatically via ```python # Pass ttl_seconds on Document or upsert() -db.upsert([ - Document(id="session-1", text="Temporary session context", ttl_seconds=3600), - Document(id="cached-doc", text="Ephemeral cache data", ttl_seconds=86400), -]) +db.upsert( + [ + Document(id="session-1", text="Temporary session context", ttl_seconds=3600), + Document(id="cached-doc", text="Ephemeral cache data", ttl_seconds=86400), + ] +) # Or set a default TTL across an entire batch ns = db.namespace("sessions") @@ -259,11 +267,13 @@ from dynavec.integrations.tools import make_retriever_fn retrieve = make_retriever_fn(db, namespace="kb", top_k=2) + @tool def search_knowledge_base(query: str) -> str: """Search the dynavec knowledge base for relevant passages.""" return retrieve(query) + agent = Agent(tools=[search_knowledge_base]) agent("Where is dynavec vector data stored?") ``` @@ -426,7 +436,7 @@ from dynavec.dashboard import serve rec = TelemetryRecorder() db = Dynavec(cfg, embedder=emb, cache=SemanticCache(), telemetry=rec) # ... your app runs searches; the recorder fills automatically ... -serve(rec, port=8779) # JSON API at http://127.0.0.1:8779 +serve(rec, port=8779) # JSON API at http://127.0.0.1:8779 ``` **2. Run the dashboard** (points at that API; falls back to sample data if unset): @@ -501,75 +511,6 @@ db.graph_delete_edge("acme", "competes_with", "globex", namespace="kb", bidirect db.graph_delete_node("globex", namespace="kb") ``` -Load graph records with grouped DynamoDB updates: - -```python -result = db.graph_import( - namespace="kb", - nodes=[{"id": "acme", "ntype": "company", "props": {"country": "US"}}], - edges=[{"src": "acme", "relation": "supplies", "dst": "globex", - "weight": 2.5, "props": {"created_at": "2026-10-07T12:00:00Z"}}], - batch_size=100, -) -print(result.nodes_processed, result.edges_appended) # 2, 1 - -db.graph_import_file(namespace="kb", json_file="graph.json") -db.graph_import_file(namespace="kb", nodes_csv="nodes.csv", edges_csv="edges.csv") -# Either CSV file can also be supplied alone. -``` - -Both client methods default to namespace `"default"`. The storage APIs are -`GraphStore.import_graph(ns, *, nodes=None, edges=None, batch_size=100)` and -`GraphStore.import_graph_file(ns, *, json_file=None, nodes_csv=None, edges_csv=None, -batch_size=100)`. They return `GraphImportResult`: `nodes_processed` counts unique -IDs, including inferred edge endpoints and already-existing nodes; -`edges_appended` counts input edge records. Empty record collections return zeros. - -- **Python:** `nodes` and `edges` are sequences of dictionaries. Nodes require a - non-empty string `id`; optional `ntype` is a string or `None`, and `props` is a - dictionary or `None`. Edges require non-empty strings `src`, `relation`, and - `dst`; optional `weight` defaults to `1.0`, and omitted/`None` `props` defaults - to `{}`. Identifiers are stored unchanged, including spaces. -- **JSON:** a UTF-8 object containing optional `nodes` and `edges` arrays using - those same fields. Missing arrays mean empty collections; `null` arrays are - invalid. For example: `{"nodes": [{"id": "acme"}], "edges": [{"src": "acme", - "relation": "supplies", "dst": "globex"}]}`. -- **CSV:** separate UTF-8 files. Node headers: required `id`, optional `ntype,props`. - Edge headers: required `src,relation,dst`, optional `weight,props`. A `props` - cell must contain a JSON object; use standard CSV quoting, e.g. - `acme,company,"{""country"":""US""}"`. Blank optional cells mean omitted values. - Supply JSON or CSV files in one call; combining formats is invalid. - -Existing edges and document links are preserved. Omitted/`None` node attributes -preserve existing values; explicit node `props` replaces the entire dictionary, -so `{}` clears it. For repeated node IDs, later supplied attributes win, while -omitted values preserve earlier ones. Missing edge endpoints are created without -overwriting existing endpoint metadata. Duplicate edges remain duplicates, and -repeating an import appends them again: imports are **not idempotent**. - -Finite numeric weights may be negative or greater than one; booleans, NaN, -infinity, and numbers outside DynamoDB's supported precision/range are rejected. -Nested floats in properties are copied as `Decimal` values for DynamoDB, caller -inputs are not mutated, and timestamp strings remain caller-provided strings. -Unsupported DynamoDB values are rejected before writing. - -Every supplied record/file is loaded and validated before any database writes, -including when both CSV files are supplied. Errors identify a record index or -filename/CSV row. Complete prevalidation uses memory proportional to the import. -Each unique node is prepared once, then edges are grouped by source and appended -in input order within that source. `batch_size` must be a positive integer -(excluding booleans) and limits edges per append request. Three edges from one -source to three destinations use five `UpdateItem` requests with the default -batch size, versus nine via three `add_edge()` calls. This is grouping, not -compression or a single AWS request; `BatchWriteItem` is not used. - -The existing **400 KB DynamoDB item limit** still applies to each node's complete -adjacency list, even with small chunks. AWS failures can leave partial writes; -there is no whole-import rollback or exactly-once guarantee. Safe node preparation -uses the existing retry policy, but the importer does not retry edge appends or -restart completed chunks. SDK retries can still make ambiguous append failures -unsafe to replay; inspect the stored graph before deciding how to recover. - ## Ecosystem dynavec started as a vector database, and it's growing into a full agent stack that runs inside your own cloud account. The plan has four parts: diff --git a/src/dynavec/agents/base.py b/src/dynavec/agents/base.py index fe4e84a..98f1141 100644 --- a/src/dynavec/agents/base.py +++ b/src/dynavec/agents/base.py @@ -9,6 +9,7 @@ from typing import Any from ..chat.base import Tool, ToolCall +from ..exceptions import NodeInterrupt @dataclass @@ -31,6 +32,7 @@ class AgentResult: termination_reason: str = "completed" # "completed", "max_steps_reached", "error" total_steps: int = 0 tool_calls_count: int = 0 + interrupt_payload: dict[str, Any] | None = None @dataclass @@ -191,6 +193,8 @@ def execute(self, arguments: dict[str, Any] | str | None = None) -> str: if isinstance(result, str): return result return json.dumps(result, ensure_ascii=False) + except NodeInterrupt: + raise except Exception as exc: # noqa: BLE001 return f"Error executing tool {self.name!r}: {exc}" @@ -209,6 +213,8 @@ async def aexecute(self, arguments: dict[str, Any] | str | None = None) -> str: if isinstance(result, str): return result return json.dumps(result, ensure_ascii=False) + except NodeInterrupt: + raise except Exception as exc: # noqa: BLE001 return f"Error executing tool {self.name!r}: {exc}" @@ -234,3 +240,10 @@ def decorator(fn: Callable[..., Any]) -> AgentTool: ) return decorator + + +def interrupt(thread_id: str, node_id: str, payload: dict[str, Any] | None = None) -> None: + """Pause current run and wait for human input/approval.""" + from ..exceptions import NodeInterrupt + + raise NodeInterrupt(thread_id=thread_id, node_id=node_id, payload=payload) \ No newline at end of file diff --git a/src/dynavec/agents/react.py b/src/dynavec/agents/react.py index d8be15c..cfa4759 100644 --- a/src/dynavec/agents/react.py +++ b/src/dynavec/agents/react.py @@ -4,15 +4,11 @@ import json from collections.abc import Callable, Sequence -from typing import TYPE_CHECKING, Any +from typing import Any -from ..chat.base import ChatModel, Message, Tool, ToolCall +from ..chat.base import ChatModel, Message, Tool +from ..exceptions import NodeInterrupt from .base import AgentResult, AgentStep, AgentTool -from .registry import ToolRegistry -from .state import AgentState - -if TYPE_CHECKING: - from ..checkpoint import BaseCheckpointer DEFAULT_REACT_SYSTEM_PROMPT = """You are a helpful and precise reasoning agent. You solve tasks step-by-step using a ReAct (Reason + Act) approach. @@ -20,209 +16,71 @@ When you have the final answer to the user's goal, provide the clear and complete answer directly without calling any tools.""" -def _message_to_dict(msg: Message) -> dict[str, Any]: - return { - "role": msg.role, - "content": msg.content, - "tool_call_id": msg.tool_call_id, - "tool_calls": [ - {"id": tc.id, "name": tc.name, "arguments": tc.arguments} for tc in msg.tool_calls - ], - } - - -def _dict_to_message(d: dict[str, Any]) -> Message: - tcs = [ - ToolCall(id=t["id"], name=t["name"], arguments=t["arguments"]) - for t in d.get("tool_calls", []) - ] - return Message( - role=d["role"], - content=d.get("content"), - tool_calls=tcs, - tool_call_id=d.get("tool_call_id"), - ) - - -def _step_to_dict(step: AgentStep) -> dict[str, Any]: - return { - "step_number": step.step_number, - "thought": step.thought, - "tool_calls": [ - {"id": tc.id, "name": tc.name, "arguments": tc.arguments} for tc in step.tool_calls - ], - "observations": list(step.observations), - } - - -def _dict_to_step(d: dict[str, Any]) -> AgentStep: - tcs = [ - ToolCall(id=t["id"], name=t["name"], arguments=t["arguments"]) - for t in d.get("tool_calls", []) - ] - return AgentStep( - step_number=int(d["step_number"]), - thought=d.get("thought"), - tool_calls=tcs, - observations=list(d.get("observations") or []), - ) - - class ReActAgent: """A ReAct-style agent that alternates between thinking, calling tools, and observing results.""" def __init__( self, model: ChatModel, - tools: Sequence[AgentTool | Callable[..., Any]] | ToolRegistry | None = None, + tools: Sequence[AgentTool | Callable[..., Any]] | None = None, system_prompt: str | None = None, max_steps: int = 10, - checkpointer: BaseCheckpointer | None = None, ) -> None: self.model = model self.system_prompt = ( system_prompt if system_prompt is not None else DEFAULT_REACT_SYSTEM_PROMPT ) self.max_steps = max(1, max_steps) - self.checkpointer = checkpointer self._tools_map: dict[str, AgentTool] = {} self._chat_tools: list[Tool] = [] - if isinstance(tools, ToolRegistry): - for t in tools.list_tools(): - self._tools_map[t.name] = t - self._chat_tools.append(t.to_chat_tool()) - elif tools: - for raw_tool in tools: - agent_tool = raw_tool if isinstance(raw_tool, AgentTool) else AgentTool(raw_tool) + if tools: + for t in tools: + if isinstance(t, AgentTool): + agent_tool = t + elif callable(t): + agent_tool = AgentTool(t) + else: + continue self._tools_map[agent_tool.name] = agent_tool self._chat_tools.append(agent_tool.to_chat_tool()) - def _init_messages( - self, - goal: str, - state: AgentState | None = None, - ) -> list[Message]: + def run(self, goal: str, thread_id: str = "default", **kwargs: Any) -> AgentResult: + """Execute the ReAct loop synchronously until goal completion, max_steps, or interrupt.""" messages: list[Message] = [] if self.system_prompt: messages.append(Message(role="system", content=self.system_prompt)) + messages.append(Message(role="user", content=goal)) - user_content = goal - if state is not None and state.data: - shared_context = json.dumps(state.data, default=str, sort_keys=True) - user_content = f"{goal}\n\n" f"Shared workflow context (JSON):\n{shared_context}" - - messages.append(Message(role="user", content=user_content)) - return messages + return self._execute_loop(messages=messages, thread_id=thread_id, start_step=1, **kwargs) - def _dispatch_tool_call(self, tc: ToolCall) -> str: - tool_instance = self._tools_map.get(tc.name) - if tool_instance is not None: - return tool_instance.execute(tc.arguments) - return f"Error: Tool {tc.name!r} is not registered in available tools." - - async def _adispatch_tool_call(self, tc: ToolCall) -> str: - tool_instance = self._tools_map.get(tc.name) - if tool_instance is not None: - return await tool_instance.aexecute(tc.arguments) - return f"Error: Tool {tc.name!r} is not registered in available tools." - - def _process_tool_calls(self, tool_calls: list[ToolCall], messages: list[Message]) -> list[str]: - step_observations: list[str] = [] - for tc in tool_calls: - obs = self._dispatch_tool_call(tc) - step_observations.append(obs) - messages.append( - Message( - role="tool", - content=obs, - tool_call_id=tc.id, - ) - ) - return step_observations - - async def _aprocess_tool_calls( - self, tool_calls: list[ToolCall], messages: list[Message] - ) -> list[str]: - step_observations: list[str] = [] - for tc in tool_calls: - obs = await self._adispatch_tool_call(tc) - step_observations.append(obs) - messages.append( - Message( - role="tool", - content=obs, - tool_call_id=tc.id, - ) - ) - return step_observations - - def _finalize_result(self, steps: list[AgentStep], total_tool_calls: int) -> AgentResult: - last_output = ( - steps[-1].thought or (steps[-1].observations[-1] if steps[-1].observations else "") - if steps - else "" + def resume( + self, + messages: list[Message], + human_decision: Any, + thread_id: str = "default", + start_step: int = 1, + **kwargs: Any, + ) -> AgentResult: + """Resume an interrupted run synchronously by injecting the human decision into context.""" + decision_str = ( + json.dumps(human_decision) if isinstance(human_decision, dict) else str(human_decision) ) - return AgentResult( - output=last_output, - steps=steps, - finished=False, - termination_reason="max_steps_reached", - total_steps=self.max_steps, - tool_calls_count=total_tool_calls, + messages.append(Message(role="user", content=f"Human Decision / Input: {decision_str}")) + return self._execute_loop( + messages=messages, thread_id=thread_id, start_step=start_step, **kwargs ) - def run( + def _execute_loop( self, - goal: str, - *, - thread_id: str | None = None, - resume_from: str | None = None, - state: AgentState | None = None, + messages: list[Message], + thread_id: str, + start_step: int, **kwargs: Any, ) -> AgentResult: - """Execute the ReAct loop synchronously until goal completion or max_steps.""" - messages: list[Message] = [] steps: list[AgentStep] = [] total_tool_calls = 0 - start_step = 1 - - if self.checkpointer is not None and thread_id is not None: - existing_cp = self.checkpointer.get(thread_id, resume_from) - - if existing_cp is not None: - checkpoint_state = existing_cp.state - if checkpoint_state is not None: - if "messages" in checkpoint_state: - messages = [ - _dict_to_message(m) - for m in checkpoint_state["messages"] - ] - if "steps" in checkpoint_state: - steps = [ - _dict_to_step(s) - for s in checkpoint_state["steps"] - ] - total_tool_calls = int( - checkpoint_state.get("total_tool_calls", 0) - ) - start_step = existing_cp.step + 1 - - if not messages: - messages = self._init_messages(goal, state) - if self.checkpointer is not None and thread_id is not None: - self.checkpointer.put( - thread_id, - state={ - "goal": goal, - "messages": [_message_to_dict(m) for m in messages], - "steps": [], - "total_tool_calls": 0, - }, - node_id="init", - step=0, - ) for step_idx in range(start_step, self.max_steps + 1): chat_res = self.model.invoke( @@ -241,20 +99,6 @@ def run( observations=[], ) steps.append(step) - if self.checkpointer is not None and thread_id is not None: - self.checkpointer.put( - thread_id, - state={ - "goal": goal, - "messages": [_message_to_dict(m) for m in messages], - "steps": [_step_to_dict(s) for s in steps], - "total_tool_calls": total_tool_calls, - "output": msg.content or "", - "finished": True, - }, - node_id=f"step_{step_idx}", - step=step_idx, - ) return AgentResult( output=msg.content or "", steps=steps, @@ -264,8 +108,41 @@ def run( tool_calls_count=total_tool_calls, ) - total_tool_calls += len(msg.tool_calls) - step_observations = self._process_tool_calls(list(msg.tool_calls), messages) + step_observations: list[str] = [] + for tc in msg.tool_calls: + total_tool_calls += 1 + tool_instance = self._tools_map.get(tc.name) + + try: + if tool_instance is not None: + obs = tool_instance.execute(tc.arguments) + else: + obs = f"Error: Tool {tc.name!r} is not registered in available tools." + except NodeInterrupt as exc: + return AgentResult( + output=f"Execution interrupted at node {exc.node_id!r}", + steps=steps, + finished=False, + termination_reason="interrupted", + total_steps=step_idx, + tool_calls_count=total_tool_calls, + interrupt_payload={ + "thread_id": exc.thread_id, + "node_id": exc.node_id, + "payload": exc.payload, + "messages": messages, + "start_step": step_idx, + }, + ) + + step_observations.append(obs) + messages.append( + Message( + role="tool", + content=obs, + tool_call_id=tc.id, + ) + ) step = AgentStep( step_number=step_idx, @@ -275,86 +152,55 @@ def run( ) steps.append(step) - if self.checkpointer is not None and thread_id is not None: - self.checkpointer.put( - thread_id, - state={ - "goal": goal, - "messages": [_message_to_dict(m) for m in messages], - "steps": [_step_to_dict(s) for s in steps], - "total_tool_calls": total_tool_calls, - "finished": False, - }, - node_id=f"step_{step_idx}", - step=step_idx, - ) + last_output = steps[-1].thought or ( + steps[-1].observations[-1] if steps[-1].observations else "" + ) + return AgentResult( + output=last_output, + steps=steps, + finished=False, + termination_reason="max_steps_reached", + total_steps=self.max_steps, + tool_calls_count=total_tool_calls, + ) - res = self._finalize_result(steps, total_tool_calls) - if self.checkpointer is not None and thread_id is not None: - self.checkpointer.put( - thread_id, - state={ - "goal": goal, - "messages": [_message_to_dict(m) for m in messages], - "steps": [_step_to_dict(s) for s in steps], - "total_tool_calls": total_tool_calls, - "output": res.output, - "finished": False, - }, - node_id="max_steps_reached", - step=self.max_steps, - ) - return res + async def arun(self, goal: str, thread_id: str = "default", **kwargs: Any) -> AgentResult: + """Execute the ReAct loop asynchronously until goal completion, max_steps, or interrupt.""" + messages: list[Message] = [] + if self.system_prompt: + messages.append(Message(role="system", content=self.system_prompt)) + messages.append(Message(role="user", content=goal)) + + return await self._aexecute_loop( + messages=messages, thread_id=thread_id, start_step=1, **kwargs + ) - async def arun( + async def aresume( self, - goal: str, - *, - thread_id: str | None = None, - resume_from: str | None = None, - state: AgentState | None = None, + messages: list[Message], + human_decision: Any, + thread_id: str = "default", + start_step: int = 1, + **kwargs: Any, + ) -> AgentResult: + """Resume an interrupted run asynchronously by injecting human decision into context.""" + decision_str = ( + json.dumps(human_decision) if isinstance(human_decision, dict) else str(human_decision) + ) + messages.append(Message(role="user", content=f"Human Decision / Input: {decision_str}")) + return await self._aexecute_loop( + messages=messages, thread_id=thread_id, start_step=start_step, **kwargs + ) + + async def _aexecute_loop( + self, + messages: list[Message], + thread_id: str, + start_step: int, **kwargs: Any, ) -> AgentResult: - """Execute the ReAct loop asynchronously.""" - messages: list[Message] = [] steps: list[AgentStep] = [] total_tool_calls = 0 - start_step = 1 - - if self.checkpointer is not None and thread_id is not None: - existing_cp = self.checkpointer.get(thread_id, resume_from) - if existing_cp is not None: - checkpoint_state = existing_cp.state - if checkpoint_state is not None: - if "messages" in checkpoint_state: - messages = [ - _dict_to_message(m) - for m in checkpoint_state["messages"] - ] - if "steps" in checkpoint_state: - steps = [ - _dict_to_step(s) - for s in checkpoint_state["steps"] - ] - total_tool_calls = int( - checkpoint_state.get("total_tool_calls", 0) - ) - start_step = existing_cp.step + 1 - - if not messages: - messages = self._init_messages(goal, state) - if self.checkpointer is not None and thread_id is not None: - self.checkpointer.put( - thread_id, - state={ - "goal": goal, - "messages": [_message_to_dict(m) for m in messages], - "steps": [], - "total_tool_calls": 0, - }, - node_id="init", - step=0, - ) for step_idx in range(start_step, self.max_steps + 1): chat_res = await self.model.ainvoke( @@ -373,20 +219,6 @@ async def arun( observations=[], ) steps.append(step) - if self.checkpointer is not None and thread_id is not None: - self.checkpointer.put( - thread_id, - state={ - "goal": goal, - "messages": [_message_to_dict(m) for m in messages], - "steps": [_step_to_dict(s) for s in steps], - "total_tool_calls": total_tool_calls, - "output": msg.content or "", - "finished": True, - }, - node_id=f"step_{step_idx}", - step=step_idx, - ) return AgentResult( output=msg.content or "", steps=steps, @@ -396,8 +228,41 @@ async def arun( tool_calls_count=total_tool_calls, ) - total_tool_calls += len(msg.tool_calls) - step_observations = await self._aprocess_tool_calls(list(msg.tool_calls), messages) + step_observations: list[str] = [] + for tc in msg.tool_calls: + total_tool_calls += 1 + tool_instance = self._tools_map.get(tc.name) + + try: + if tool_instance is not None: + obs = await tool_instance.aexecute(tc.arguments) + else: + obs = f"Error: Tool {tc.name!r} is not registered in available tools." + except NodeInterrupt as exc: + return AgentResult( + output=f"Execution interrupted at node {exc.node_id!r}", + steps=steps, + finished=False, + termination_reason="interrupted", + total_steps=step_idx, + tool_calls_count=total_tool_calls, + interrupt_payload={ + "thread_id": exc.thread_id, + "node_id": exc.node_id, + "payload": exc.payload, + "messages": messages, + "start_step": step_idx, + }, + ) + + step_observations.append(obs) + messages.append( + Message( + role="tool", + content=obs, + tool_call_id=tc.id, + ) + ) step = AgentStep( step_number=step_idx, @@ -407,33 +272,14 @@ async def arun( ) steps.append(step) - if self.checkpointer is not None and thread_id is not None: - self.checkpointer.put( - thread_id, - state={ - "goal": goal, - "messages": [_message_to_dict(m) for m in messages], - "steps": [_step_to_dict(s) for s in steps], - "total_tool_calls": total_tool_calls, - "finished": False, - }, - node_id=f"step_{step_idx}", - step=step_idx, - ) - - res = self._finalize_result(steps, total_tool_calls) - if self.checkpointer is not None and thread_id is not None: - self.checkpointer.put( - thread_id, - state={ - "goal": goal, - "messages": [_message_to_dict(m) for m in messages], - "steps": [_step_to_dict(s) for s in steps], - "total_tool_calls": total_tool_calls, - "output": res.output, - "finished": False, - }, - node_id="max_steps_reached", - step=self.max_steps, - ) - return res + last_output = steps[-1].thought or ( + steps[-1].observations[-1] if steps[-1].observations else "" + ) + return AgentResult( + output=last_output, + steps=steps, + finished=False, + termination_reason="max_steps_reached", + total_steps=self.max_steps, + tool_calls_count=total_tool_calls, + ) diff --git a/src/dynavec/exceptions.py b/src/dynavec/exceptions.py index 5ae80ff..76b43b9 100644 --- a/src/dynavec/exceptions.py +++ b/src/dynavec/exceptions.py @@ -2,6 +2,8 @@ from __future__ import annotations +from typing import Any + class DynavecError(Exception): """Base class for all dynavec errors.""" @@ -27,14 +29,6 @@ class NotFoundError(DynavecError): """Raised when a namespace, index, or document does not exist.""" -class AccessDeniedError(DynavecError): - """Raised when access to an AWS resource is denied.""" - - -class ValidationError(DynavecError): - """Raised when AWS rejects an invalid request.""" - - class ItemTooLargeError(DynavecError): """Raised when a document would exceed DynamoDB's 400 KB item size limit.""" @@ -76,3 +70,40 @@ def __init__(self, feature: str, package: str, extra: str) -> None: self.feature = feature self.package = package self.extra = extra + + +class NodeInterrupt(DynavecError): + """Raised when an agent execution node pauses waiting for human approval or input.""" + + def __init__( + self, thread_id: str, node_id: str, payload: dict[str, Any] | None = None + ) -> None: + super().__init__( + f"Execution interrupted at node {node_id!r} for thread {thread_id!r}. " + "Awaiting human input." + ) + self.thread_id = thread_id + self.node_id = node_id + self.payload = payload or {} + + +class ApprovalTimeoutError(DynavecError): + """Raised when a pending human approval expires before a decision is submitted.""" + + def __init__(self, thread_id: str, node_id: str, timeout_seconds: int) -> None: + super().__init__( + f"Approval request at node {node_id!r} for thread {thread_id!r} " + f"timed out after {timeout_seconds} seconds." + ) + self.thread_id = thread_id + self.node_id = node_id + self.timeout_seconds = timeout_seconds + + + +class AccessDeniedError(DynavecError): + """Raised when AWS IAM permissions or authentication fails.""" + + +class ValidationError(DynavecError): + """Raised when input parameter validation fails.""" \ No newline at end of file diff --git a/tests/test_hitl_interrupts.py b/tests/test_hitl_interrupts.py new file mode 100644 index 0000000..b7a3405 --- /dev/null +++ b/tests/test_hitl_interrupts.py @@ -0,0 +1,171 @@ +"""Tests for Human-in-the-loop / interrupts functionality in dynavec agents.""" +from __future__ import annotations + +import asyncio +import json +from collections.abc import AsyncIterator, Iterator +from typing import Any + +from dynavec.agents.base import interrupt, tool +from dynavec.agents.react import ReActAgent +from dynavec.chat.base import ChatChunk, ChatModel, ChatResult, Message, ToolCall + + +class DummyChatModel(ChatModel): + """Simple mock chat model for HITL tests.""" + + def __init__(self, responses: list[ChatResult]) -> None: + super().__init__() + self.responses = responses + self.call_count = 0 + + def invoke( + self, messages: list[Message], tools: list[Any] | None = None, **kwargs: Any + ) -> ChatResult: + res = self.responses[self.call_count] + self.call_count += 1 + return res + + async def ainvoke( + self, messages: list[Message], tools: list[Any] | None = None, **kwargs: Any + ) -> ChatResult: + return self.invoke(messages, tools, **kwargs) + + def stream( + self, messages: list[Message], tools: list[Any] | None = None, **kwargs: Any + ) -> Iterator[ChatChunk]: + yield ChatChunk(content="") + + async def astream( + self, messages: list[Message], tools: list[Any] | None = None, **kwargs: Any + ) -> AsyncIterator[ChatChunk]: + yield ChatChunk(content="") + + +@tool() +def sensitive_action_tool(action: str) -> str: + """Action requiring human approval.""" + interrupt("approval_node", f"Action '{action}' requires approval. Confirm?") + return f"Action '{action}' executed successfully." + + +def test_interrupt_raised_and_caught() -> None: + mock_model = DummyChatModel( + responses=[ + ChatResult( + message=Message( + role="assistant", + content="", + tool_calls=[ + ToolCall( + id="tc_1", + name="sensitive_action_tool", + arguments=json.dumps({"action": "delete_database"}), + ) + ], + ) + ) + ] + ) + + agent = ReActAgent(model=mock_model, tools=[sensitive_action_tool]) + result = agent.run("Please delete the database.") + + assert result.finished is False + assert result.termination_reason == "interrupted" + assert result.interrupt_payload is not None + + +def test_resume_execution() -> None: + mock_model = DummyChatModel( + responses=[ + ChatResult( + message=Message( + role="assistant", + content="", + tool_calls=[ + ToolCall( + id="tc_1", + name="sensitive_action_tool", + arguments=json.dumps({"action": "delete_database"}), + ) + ], + ) + ), + ChatResult( + message=Message( + role="assistant", + content="Database action completed.", + ) + ), + ] + ) + + agent = ReActAgent(model=mock_model, tools=[sensitive_action_tool]) + result = agent.run("Please delete the database.") + + assert result.finished is False + assert result.interrupt_payload is not None + + payload = result.interrupt_payload + saved_messages = payload["messages"] + start_step = payload["start_step"] + + resume_agent = ReActAgent(model=mock_model, tools=[sensitive_action_tool]) + final_result = resume_agent.resume( + saved_messages, + start_step=start_step, + human_decision="approved", + ) + + assert final_result.finished is True + assert final_result.termination_reason == "completed" + assert "Database action completed" in final_result.output + + +def test_async_interrupt_and_resume() -> None: + async def _run_async() -> None: + mock_model = DummyChatModel( + responses=[ + ChatResult( + message=Message( + role="assistant", + content="", + tool_calls=[ + ToolCall( + id="tc_1", + name="sensitive_action_tool", + arguments=json.dumps({"action": "delete_database"}), + ) + ], + ) + ), + ChatResult( + message=Message( + role="assistant", + content="Deployment completed.", + ) + ), + ] + ) + + agent = ReActAgent(model=mock_model, tools=[sensitive_action_tool]) + result = await agent.arun("Deploy latest release.") + + assert result.finished is False + assert result.interrupt_payload is not None + + payload = result.interrupt_payload + assert payload is not None + + resume_agent = ReActAgent(model=mock_model, tools=[sensitive_action_tool]) + final_result = await resume_agent.aresume( + payload["messages"], + start_step=payload["start_step"], + human_decision="approved", + ) + + assert final_result.finished is True + assert final_result.termination_reason == "completed" + + asyncio.run(_run_async())