From 39bc6556dd6d755ae67090a49f305d1072a4d677 Mon Sep 17 00:00:00 2001 From: Uzmaa7 Date: Tue, 27 Oct 2026 21:12:34 +0530 Subject: [PATCH 1/3] feat(orchestration): rescope hitl interrupt & resume workflow onto latest development --- README.md | 113 ++------ src/dynavec/agents/base.py | 150 ++++++----- src/dynavec/agents/react.py | 469 ++++++++++++---------------------- src/dynavec/exceptions.py | 47 +++- tests/test_hitl_interrupts.py | 171 +++++++++++++ 5 files changed, 478 insertions(+), 472 deletions(-) create mode 100644 tests/test_hitl_interrupts.py 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..ee90c48 100644 --- a/src/dynavec/agents/base.py +++ b/src/dynavec/agents/base.py @@ -6,7 +6,7 @@ import json from collections.abc import Callable from dataclasses import dataclass, field -from typing import Any +from typing import Any, overload from ..chat.base import Tool, ToolCall @@ -28,9 +28,12 @@ class AgentResult: output: str steps: list[AgentStep] = field(default_factory=list) finished: bool = True - termination_reason: str = "completed" # "completed", "max_steps_reached", "error" + termination_reason: str = ( + "completed" # "completed", "max_steps_reached", "error", "interrupted" + ) total_steps: int = 0 tool_calls_count: int = 0 + interrupt_payload: dict[str, Any] | None = None @dataclass @@ -114,15 +117,14 @@ def __init__( name: str | None = None, description: str | None = None, parameters: dict[str, Any] | None = None, - tags: list[str] | None = None, ) -> None: self.fn = fn - self.name = name or fn.__name__ - self.description = description or (fn.__doc__ or f"Execute {self.name}").strip() - self.parameters = ( - parameters if parameters is not None else _generate_json_schema(fn) + # Extract name directly or fall back to function attribute + self.name = ( + name or getattr(fn, "name", None) or getattr(fn, "__name__", None) or "unnamed_tool" ) - self.tags = tags or [] + self.description = description or (fn.__doc__ or f"Execute {self.name}").strip() + self.parameters = parameters if parameters is not None else _generate_json_schema(fn) def to_chat_tool(self) -> Tool: """Convert to a dynavec.chat.Tool schema.""" @@ -132,8 +134,10 @@ def to_chat_tool(self) -> Tool: parameters=self.parameters, ) - def _parse_arguments(self, arguments: dict[str, Any] | str | None = None) -> dict[str, Any]: - """Parse raw arguments into a kwargs dictionary.""" + def execute(self, arguments: dict[str, Any] | str | None = None) -> str: + """Execute the wrapped function and return a string observation.""" + from ..exceptions import NodeInterrupt + parsed_args: dict[str, Any] = {} if isinstance(arguments, str): if arguments.strip(): @@ -147,68 +151,30 @@ def _parse_arguments(self, arguments: dict[str, Any] | str | None = None) -> dic parsed_args = {"input": arguments} elif isinstance(arguments, dict): parsed_args = arguments - return parsed_args - - def _prepare_call(self, parsed_args: dict[str, Any]) -> dict[str, Any]: - """Prepare validated kwargs for the function.""" - sig = inspect.signature(self.fn) - params = sig.parameters - if not params: - return {} - if len(params) == 1: - param = next(iter(params.values())) - if param.kind == inspect.Parameter.VAR_KEYWORD: - return parsed_args - if param.name not in parsed_args: - first_val = next(iter(parsed_args.values())) if parsed_args else None - return {param.name: first_val} - - valid_args = {} - for k, v in parsed_args.items(): - if k in params or any(p.kind == inspect.Parameter.VAR_KEYWORD for p in params.values()): - valid_args[k] = v - return valid_args - def execute(self, arguments: dict[str, Any] | str | None = None) -> str: - """Execute the wrapped function synchronously and return a string observation.""" try: - parsed_args = self._parse_arguments(arguments) - kwargs = self._prepare_call(parsed_args) - - if inspect.iscoroutinefunction(self.fn): - import asyncio - try: - loop = asyncio.get_running_loop() - except RuntimeError: - loop = None - - if loop and loop.is_running(): - return f"Error: Cannot execute async tool {self.name!r} synchronously in a running event loop." - result = asyncio.run(self.fn(**kwargs)) + # Check if function accepts kwargs or positional + sig = inspect.signature(self.fn) + params = sig.parameters + + if not params: + result = self.fn() + elif len(params) == 1 and list(params.keys())[0] not in parsed_args: + # If single param expected and keys don't match, pass the first val or raw dict + first_val = next(iter(parsed_args.values())) if parsed_args else arguments + result = self.fn(first_val) else: - result = self.fn(**kwargs) + # Filter only valid keyword arguments + valid_args = {k: v for k, v in parsed_args.items() if k in params} + result = self.fn(**valid_args) if isinstance(result, str): return result return json.dumps(result, ensure_ascii=False) - except Exception as exc: # noqa: BLE001 - return f"Error executing tool {self.name!r}: {exc}" - async def aexecute(self, arguments: dict[str, Any] | str | None = None) -> str: - """Execute the wrapped function asynchronously and return a string observation.""" - try: - parsed_args = self._parse_arguments(arguments) - kwargs = self._prepare_call(parsed_args) - - if inspect.iscoroutinefunction(self.fn): - result = await self.fn(**kwargs) - else: - import asyncio - result = await asyncio.to_thread(self.fn, **kwargs) + except NodeInterrupt: + raise - if isinstance(result, str): - return result - return json.dumps(result, ensure_ascii=False) except Exception as exc: # noqa: BLE001 return f"Error executing tool {self.name!r}: {exc}" @@ -216,21 +182,71 @@ def __call__(self, *args: Any, **kwargs: Any) -> Any: return self.fn(*args, **kwargs) + + + + + + + + + +@overload +def tool(fn: Callable[..., Any], /) -> AgentTool: ... + + +@overload +def tool( + *, + name: str | None = None, + description: str | None = None, + parameters: dict[str, Any] | None = None, +) -> Callable[[Callable[..., Any]], AgentTool]: ... + + def tool( + fn_or_name: Callable[..., Any] | str | None = None, + *, name: str | None = None, description: str | None = None, parameters: dict[str, Any] | None = None, - tags: list[str] | None = None, -) -> Callable[[Callable[..., Any]], AgentTool]: +) -> AgentTool | Callable[[Callable[..., Any]], AgentTool]: """Decorator to convert a standard Python function into an AgentTool.""" + custom_name = name if name is not None else (fn_or_name if isinstance(fn_or_name, str) else None) + + if callable(fn_or_name): + fn = fn_or_name + tool_name = custom_name or getattr(fn, "__name__", "unnamed_tool") + return AgentTool( + fn=fn, + name=tool_name, + description=description, + parameters=parameters, + ) def decorator(fn: Callable[..., Any]) -> AgentTool: + if isinstance(fn, AgentTool): + if custom_name: + fn.name = custom_name + if description: + fn.description = description + if parameters: + fn.parameters = parameters + return fn + + tool_name = custom_name or getattr(fn, "__name__", "unnamed_tool") return AgentTool( fn=fn, - name=name, + name=tool_name, description=description, parameters=parameters, - tags=tags, ) 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) diff --git a/src/dynavec/agents/react.py b/src/dynavec/agents/react.py index d8be15c..ce06fe3 100644 --- a/src/dynavec/agents/react.py +++ b/src/dynavec/agents/react.py @@ -2,17 +2,14 @@ from __future__ import annotations +import asyncio 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 +17,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 +100,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 +109,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 +153,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 +220,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 +229,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 asyncio.to_thread(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, @@ -407,33 +273,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..c48f281 --- /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()) From 228aeb38220bdc5bbebb9748fd0b62cb8913c551 Mon Sep 17 00:00:00 2001 From: Uzmaa7 Date: Tue, 27 Oct 2026 21:45:48 +0530 Subject: [PATCH 2/3] fix(orchestration): resolve HITL interrupt propagation, tool decorator types, and react loop integration --- src/dynavec/agents/base.py | 147 +++++++++++++++++----------------- src/dynavec/agents/react.py | 3 +- tests/test_hitl_interrupts.py | 2 +- 3 files changed, 75 insertions(+), 77 deletions(-) diff --git a/src/dynavec/agents/base.py b/src/dynavec/agents/base.py index ee90c48..98f1141 100644 --- a/src/dynavec/agents/base.py +++ b/src/dynavec/agents/base.py @@ -6,9 +6,10 @@ import json from collections.abc import Callable from dataclasses import dataclass, field -from typing import Any, overload +from typing import Any from ..chat.base import Tool, ToolCall +from ..exceptions import NodeInterrupt @dataclass @@ -28,9 +29,7 @@ class AgentResult: output: str steps: list[AgentStep] = field(default_factory=list) finished: bool = True - termination_reason: str = ( - "completed" # "completed", "max_steps_reached", "error", "interrupted" - ) + 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 @@ -117,14 +116,15 @@ def __init__( name: str | None = None, description: str | None = None, parameters: dict[str, Any] | None = None, + tags: list[str] | None = None, ) -> None: self.fn = fn - # Extract name directly or fall back to function attribute - self.name = ( - name or getattr(fn, "name", None) or getattr(fn, "__name__", None) or "unnamed_tool" - ) + self.name = name or fn.__name__ self.description = description or (fn.__doc__ or f"Execute {self.name}").strip() - self.parameters = parameters if parameters is not None else _generate_json_schema(fn) + self.parameters = ( + parameters if parameters is not None else _generate_json_schema(fn) + ) + self.tags = tags or [] def to_chat_tool(self) -> Tool: """Convert to a dynavec.chat.Tool schema.""" @@ -134,10 +134,8 @@ def to_chat_tool(self) -> Tool: parameters=self.parameters, ) - def execute(self, arguments: dict[str, Any] | str | None = None) -> str: - """Execute the wrapped function and return a string observation.""" - from ..exceptions import NodeInterrupt - + def _parse_arguments(self, arguments: dict[str, Any] | str | None = None) -> dict[str, Any]: + """Parse raw arguments into a kwargs dictionary.""" parsed_args: dict[str, Any] = {} if isinstance(arguments, str): if arguments.strip(): @@ -151,30 +149,72 @@ def execute(self, arguments: dict[str, Any] | str | None = None) -> str: parsed_args = {"input": arguments} elif isinstance(arguments, dict): parsed_args = arguments + return parsed_args + + def _prepare_call(self, parsed_args: dict[str, Any]) -> dict[str, Any]: + """Prepare validated kwargs for the function.""" + sig = inspect.signature(self.fn) + params = sig.parameters + if not params: + return {} + if len(params) == 1: + param = next(iter(params.values())) + if param.kind == inspect.Parameter.VAR_KEYWORD: + return parsed_args + if param.name not in parsed_args: + first_val = next(iter(parsed_args.values())) if parsed_args else None + return {param.name: first_val} + + valid_args = {} + for k, v in parsed_args.items(): + if k in params or any(p.kind == inspect.Parameter.VAR_KEYWORD for p in params.values()): + valid_args[k] = v + return valid_args + def execute(self, arguments: dict[str, Any] | str | None = None) -> str: + """Execute the wrapped function synchronously and return a string observation.""" try: - # Check if function accepts kwargs or positional - sig = inspect.signature(self.fn) - params = sig.parameters - - if not params: - result = self.fn() - elif len(params) == 1 and list(params.keys())[0] not in parsed_args: - # If single param expected and keys don't match, pass the first val or raw dict - first_val = next(iter(parsed_args.values())) if parsed_args else arguments - result = self.fn(first_val) + parsed_args = self._parse_arguments(arguments) + kwargs = self._prepare_call(parsed_args) + + if inspect.iscoroutinefunction(self.fn): + import asyncio + try: + loop = asyncio.get_running_loop() + except RuntimeError: + loop = None + + if loop and loop.is_running(): + return f"Error: Cannot execute async tool {self.name!r} synchronously in a running event loop." + result = asyncio.run(self.fn(**kwargs)) else: - # Filter only valid keyword arguments - valid_args = {k: v for k, v in parsed_args.items() if k in params} - result = self.fn(**valid_args) + result = self.fn(**kwargs) 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}" + + async def aexecute(self, arguments: dict[str, Any] | str | None = None) -> str: + """Execute the wrapped function asynchronously and return a string observation.""" + try: + parsed_args = self._parse_arguments(arguments) + kwargs = self._prepare_call(parsed_args) + + if inspect.iscoroutinefunction(self.fn): + result = await self.fn(**kwargs) + else: + import asyncio + result = await asyncio.to_thread(self.fn, **kwargs) + 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}" @@ -182,64 +222,21 @@ def __call__(self, *args: Any, **kwargs: Any) -> Any: return self.fn(*args, **kwargs) - - - - - - - - - -@overload -def tool(fn: Callable[..., Any], /) -> AgentTool: ... - - -@overload -def tool( - *, - name: str | None = None, - description: str | None = None, - parameters: dict[str, Any] | None = None, -) -> Callable[[Callable[..., Any]], AgentTool]: ... - - def tool( - fn_or_name: Callable[..., Any] | str | None = None, - *, name: str | None = None, description: str | None = None, parameters: dict[str, Any] | None = None, -) -> AgentTool | Callable[[Callable[..., Any]], AgentTool]: + tags: list[str] | None = None, +) -> Callable[[Callable[..., Any]], AgentTool]: """Decorator to convert a standard Python function into an AgentTool.""" - custom_name = name if name is not None else (fn_or_name if isinstance(fn_or_name, str) else None) - - if callable(fn_or_name): - fn = fn_or_name - tool_name = custom_name or getattr(fn, "__name__", "unnamed_tool") - return AgentTool( - fn=fn, - name=tool_name, - description=description, - parameters=parameters, - ) def decorator(fn: Callable[..., Any]) -> AgentTool: - if isinstance(fn, AgentTool): - if custom_name: - fn.name = custom_name - if description: - fn.description = description - if parameters: - fn.parameters = parameters - return fn - - tool_name = custom_name or getattr(fn, "__name__", "unnamed_tool") return AgentTool( fn=fn, - name=tool_name, + name=name, description=description, parameters=parameters, + tags=tags, ) return decorator @@ -249,4 +246,4 @@ def interrupt(thread_id: str, node_id: str, payload: dict[str, Any] | None = Non """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) + 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 ce06fe3..785edec 100644 --- a/src/dynavec/agents/react.py +++ b/src/dynavec/agents/react.py @@ -10,6 +10,7 @@ from ..chat.base import ChatModel, Message, Tool from ..exceptions import NodeInterrupt from .base import AgentResult, AgentStep, AgentTool +from ..exceptions import NodeInterrupt DEFAULT_REACT_SYSTEM_PROMPT = """You are a helpful and precise reasoning agent. You solve tasks step-by-step using a ReAct (Reason + Act) approach. @@ -236,7 +237,7 @@ async def _aexecute_loop( try: if tool_instance is not None: - obs = await asyncio.to_thread(tool_instance.execute, tc.arguments) + 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: diff --git a/tests/test_hitl_interrupts.py b/tests/test_hitl_interrupts.py index c48f281..b7a3405 100644 --- a/tests/test_hitl_interrupts.py +++ b/tests/test_hitl_interrupts.py @@ -42,7 +42,7 @@ async def astream( yield ChatChunk(content="") -@tool +@tool() def sensitive_action_tool(action: str) -> str: """Action requiring human approval.""" interrupt("approval_node", f"Action '{action}' requires approval. Confirm?") From db8d20aef4c41a4bfabc301507587c9ca3c74067 Mon Sep 17 00:00:00 2001 From: Uzmaa7 Date: Tue, 27 Oct 2026 22:04:20 +0530 Subject: [PATCH 3/3] style(agents): resolve ruff linting errors and duplicate imports in react.py --- src/dynavec/agents/react.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/dynavec/agents/react.py b/src/dynavec/agents/react.py index 785edec..cfa4759 100644 --- a/src/dynavec/agents/react.py +++ b/src/dynavec/agents/react.py @@ -2,7 +2,6 @@ from __future__ import annotations -import asyncio import json from collections.abc import Callable, Sequence from typing import Any @@ -10,7 +9,6 @@ from ..chat.base import ChatModel, Message, Tool from ..exceptions import NodeInterrupt from .base import AgentResult, AgentStep, AgentTool -from ..exceptions import NodeInterrupt DEFAULT_REACT_SYSTEM_PROMPT = """You are a helpful and precise reasoning agent. You solve tasks step-by-step using a ReAct (Reason + Act) approach.