feat: implement core agent primitives (ReAct and Planner)

src/dynavec/agents/__init__.py

@@ -0,0 +1,25 @@
+"""Agent primitives and execution loops for dynaflow."""
+
+from __future__ import annotations
+
+from .base import (
+    AgentResult,
+    AgentStep,
+    AgentTool,
+    Plan,
+    PlanStep,
+    tool,
+)
+from .planner import Planner
+from .react import ReActAgent
+
+__all__ = [
+    "AgentResult",
+    "AgentStep",
+    "AgentTool",
+    "Plan",
+    "PlanStep",
+    "Planner",
+    "ReActAgent",
+    "tool",
+]

src/dynavec/agents/base.py

@@ -0,0 +1,192 @@
+"""Core data models and tool primitives for dynaflow agents."""
+
+from __future__ import annotations
+
+import inspect
+import json
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from typing import Any
+
+from ..chat.base import Tool, ToolCall
+
+
+@dataclass
+class AgentStep:
+    """A single execution step in an agent's reasoning loop."""
+
+    step_number: int
+    thought: str | None = None
+    tool_calls: list[ToolCall] = field(default_factory=list)
+    observations: list[str] = field(default_factory=list)
+
+
+@dataclass
+class AgentResult:
+    """The final result of an agent run."""
+
+    output: str
+    steps: list[AgentStep] = field(default_factory=list)
+    finished: bool = True
+    termination_reason: str = "completed"  # "completed", "max_steps_reached", "error"
+    total_steps: int = 0
+    tool_calls_count: int = 0
+
+
+@dataclass
+class PlanStep:
+    """A single step in a decomposed task plan."""
+
+    step_number: int
+    description: str
+    tool_hint: str | None = None
+
+
+@dataclass
+class Plan:
+    """An ordered decomposition of a goal into actionable steps."""
+
+    goal: str
+    steps: list[PlanStep] = field(default_factory=list)
+
+
+def _python_type_to_json_type(py_type: Any) -> str:
+    """Map standard Python types and type annotation strings to JSON Schema data types."""
+    if py_type in (str, "str", "string"):
+        return "string"
+    if py_type in (int, "int", "integer"):
+        return "integer"
+    if py_type in (float, "float", "number"):
+        return "number"
+    if py_type in (bool, "bool", "boolean"):
+        return "boolean"
+    if py_type in (list, tuple, set, "list", "tuple", "set") or (
+        isinstance(py_type, str) and py_type.startswith(("list[", "Sequence[", "tuple["))
+    ):
+        return "array"
+    if py_type in (dict, Any, "dict", "dict[str, Any]", "Mapping") or (
+        isinstance(py_type, str) and py_type.startswith("dict[")
+    ):
+        return "object"
+    return "string"
+
+
+def _generate_json_schema(fn: Callable[..., Any]) -> dict[str, Any]:
+    """Generate a JSON schema parameters dictionary from a function signature."""
+    sig = inspect.signature(fn)
+    properties: dict[str, Any] = {}
+    required: list[str] = []
+
+    for param_name, param in sig.parameters.items():
+        if param_name in ("self", "cls"):
+            continue
+
+        param_type = param.annotation
+        json_type = (
+            _python_type_to_json_type(param_type)
+            if param_type is not inspect.Parameter.empty
+            else "string"
+        )
+
+        properties[param_name] = {
+            "type": json_type,
+            "description": f"Parameter '{param_name}'",
+        }
+
+        if param.default is inspect.Parameter.empty:
+            required.append(param_name)
+
+    schema: dict[str, Any] = {
+        "type": "object",
+        "properties": properties,
+    }
+    if required:
+        schema["required"] = required
+    return schema
+
+
+class AgentTool:
+    """An executable tool wrapped with a typed JSON schema for agent use."""
+
+    def __init__(
+        self,
+        fn: Callable[..., Any],
+        name: str | None = None,
+        description: str | None = None,
+        parameters: dict[str, Any] | 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)
+        )
+
+    def to_chat_tool(self) -> Tool:
+        """Convert to a dynavec.chat.Tool schema."""
+        return Tool(
+            name=self.name,
+            description=self.description,
+            parameters=self.parameters,
+        )
+
+    def execute(self, arguments: dict[str, Any] | str | None = None) -> str:
+        """Execute the wrapped function and return a string observation."""
+        parsed_args: dict[str, Any] = {}
+        if isinstance(arguments, str):
+            if arguments.strip():
+                try:
+                    loaded = json.loads(arguments)
+                    if isinstance(loaded, dict):
+                        parsed_args = loaded
+                    else:
+                        parsed_args = {"input": loaded}
+                except Exception:
+                    parsed_args = {"input": arguments}
+        elif isinstance(arguments, dict):
+            parsed_args = arguments
+
+        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)
+            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)
+
+            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}"
+
+    def __call__(self, *args: Any, **kwargs: Any) -> Any:
+        return self.fn(*args, **kwargs)
+
+
+def tool(
+    name: str | None = None,
+    description: str | None = None,
+    parameters: dict[str, Any] | None = None,
+) -> Callable[[Callable[..., Any]], AgentTool]:
+    """Decorator to convert a standard Python function into an AgentTool."""
+
+    def decorator(fn: Callable[..., Any]) -> AgentTool:
+        return AgentTool(
+            fn=fn,
+            name=name,
+            description=description,
+            parameters=parameters,
+        )
+
+    return decorator

src/dynavec/agents/planner.py

@@ -0,0 +1,124 @@
+"""Planner primitive for decomposing goals into structured action plans."""
+
+from __future__ import annotations
+
+import json
+import re
+from typing import Any
+
+from ..chat.base import ChatModel, Message
+from .base import Plan, PlanStep
+
+DEFAULT_PLANNER_SYSTEM_PROMPT = """You are an expert task planning agent.
+Your job is to break down complex goals into an ordered, clear sequence of actionable steps.
+Output your plan strictly as a JSON object matching this schema:
+{
+  "steps": [
+    {
+      "step_number": 1,
+      "description": "Description of what needs to be done in this step",
+      "tool_hint": "Optional name of the tool or action suited for this step"
+    }
+  ]
+}
+Do not include any conversational filler or markdown other than the valid JSON."""
+
+
+def _extract_json_block(text: str) -> str:
+    """Extract JSON content from markdown code fences or plain text."""
+    trimmed = text.strip()
+    match = re.search(r"```(?:json)?\s*([\s\S]*?)\s*```", trimmed, re.IGNORECASE)
+    if match:
+        return match.group(1).strip()
+    return trimmed
+
+
+class Planner:
+    """Decomposes goals into structured, ordered execution plans."""
+
+    def __init__(
+        self,
+        model: ChatModel,
+        system_prompt: str | None = None,
+    ) -> None:
+        self.model = model
+        self.system_prompt = (
+            system_prompt if system_prompt is not None else DEFAULT_PLANNER_SYSTEM_PROMPT
+        )
+
+    def _parse_plan(self, goal: str, response_text: str) -> Plan:
+        """Parse raw model output into a Plan object."""
+        cleaned = _extract_json_block(response_text)
+        try:
+            data = json.loads(cleaned)
+        except Exception:
+            # Fallback: if json parsing fails, split lines into steps
+            lines = [
+                line.strip()
+                for line in response_text.splitlines()
+                if line.strip() and not line.startswith("```")
+            ]
+            steps = [
+                PlanStep(step_number=idx, description=line)
+                for idx, line in enumerate(lines, start=1)
+            ]
+            return Plan(goal=goal, steps=steps)
+
+        steps_data = data.get("steps", []) if isinstance(data, dict) else data
+        plan_steps: list[PlanStep] = []
+
+        if isinstance(steps_data, list):
+            for idx, item in enumerate(steps_data, start=1):
+                if isinstance(item, dict):
+                    step_num = item.get("step_number", idx)
+                    desc = item.get("description", str(item))
+                    tool_hint = item.get("tool_hint")
+                    plan_steps.append(
+                        PlanStep(
+                            step_number=step_num,
+                            description=desc,
+                            tool_hint=tool_hint,
+                        )
+                    )
+                elif isinstance(item, str):
+                    plan_steps.append(PlanStep(step_number=idx, description=item))
+
+        return Plan(goal=goal, steps=plan_steps)
+
+    def plan(
+        self,
+        goal: str,
+        context: str | None = None,
+        **kwargs: Any,
+    ) -> Plan:
+        """Decompose a goal into an ordered Plan synchronously."""
+        messages: list[Message] = []
+        if self.system_prompt:
+            messages.append(Message(role="system", content=self.system_prompt))
+
+        user_content = f"Goal: {goal}"
+        if context:
+            user_content += f"\n\nContext:\n{context}"
+        messages.append(Message(role="user", content=user_content))
+
+        res = self.model.invoke(messages, **kwargs)
+        return self._parse_plan(goal, res.message.content or "")
+
+    async def aplan(
+        self,
+        goal: str,
+        context: str | None = None,
+        **kwargs: Any,
+    ) -> Plan:
+        """Decompose a goal into an ordered Plan asynchronously."""
+        messages: list[Message] = []
+        if self.system_prompt:
+            messages.append(Message(role="system", content=self.system_prompt))
+
+        user_content = f"Goal: {goal}"
+        if context:
+            user_content += f"\n\nContext:\n{context}"
+        messages.append(Message(role="user", content=user_content))
+
+        res = await self.model.ainvoke(messages, **kwargs)
+        return self._parse_plan(goal, res.message.content or "")