{reasoners.length === 0 ? (
No recent reasoner activity. Trigger a workflow or execution to
diff --git a/examples/python_agent_nodes/deep_research/README.md b/examples/python_agent_nodes/deep_research/README.md
new file mode 100644
index 00000000..01eb5206
--- /dev/null
+++ b/examples/python_agent_nodes/deep_research/README.md
@@ -0,0 +1,78 @@
+# Deep Research Agent
+
+A research agent that uses recursive planning to break down research questions into subtasks, forms a topological graph for parallel execution, deduplicates tasks, and synthesizes comprehensive reports.
+
+## Features
+
+- **Recursive Planning** β Breaks down questions into subtasks with configurable depth
+- **Task Deduplication** β Merges redundant tasks
+- **Smart Search Strategy** β Decides synthesis-only vs enhanced search for parent tasks
+- **Topological Execution** β Parallel execution with dependency management
+- **Web Search** β Tavily API integration with citation tracking
+
+## Quick Start
+
+### 1. Install & Setup
+
+```bash
+pip install -r examples/python_agent_nodes/deep_research/requirements.txt
+export TAVILY_API_KEY="your-tavily-api-key"
+```
+
+### 2. Run Agent
+
+```bash
+python examples/python_agent_nodes/deep_research/main.py
+```
+
+### 3. Execute Research
+
+```bash
+curl -X POST http://localhost:8080/reasoners/planning_execute_deep_research \
+ -H "Content-Type: application/json" \
+ -d '{
+ "research_question": "What is AgentField.ai?",
+ "max_depth": 3,
+ "max_tasks_per_level": 5
+ }'
+```
+
+## Architecture
+
+1. **Plan Creation** β Recursively breaks question into distinct tasks
+2. **Deduplication** β Merges redundant tasks
+3. **Dependency Identification** β Maps task dependencies
+4. **Execution** β Topological order:
+ - Leaf tasks β web search
+ - Parent tasks β synthesis or enhanced search (with context)
+5. **Synthesis** β Final report from all findings
+
+## Endpoints
+
+### Main
+- **`/reasoners/planning_execute_deep_research`** β Complete workflow (recommended)
+ - Params: `research_question`, `max_depth` (default: 3), `max_tasks_per_level` (default: 5)
+
+### Planning
+- `/reasoners/planning_create_research_plan` β Create plan only
+- `/reasoners/planning_deduplicate_tasks` β Deduplicate tasks
+- `/reasoners/planning_identify_dependencies` β Identify dependencies
+- `/reasoners/planning_decide_search_strategy` β Decide search strategy
+
+### Research
+- `/reasoners/research_execute_research_task` β Execute leaf task (search)
+- `/reasoners/research_execute_research_task_with_context` β Execute with dependency context
+- `/reasoners/research_synthesize_from_dependencies` β Synthesize from children
+
+## Task Types
+
+- **Leaf Tasks** (no deps) β Web search
+- **Parent Tasks** (has deps) β Synthesis-only or enhanced search with context
+
+## Environment Variables
+
+| Variable | Description | Default |
+| ------------------- | ------------------------- | -------------------------------------------- |
+| `TAVILY_API_KEY` | Tavily API key (required) | - |
+| `AGENTFIELD_SERVER` | Control plane URL | `http://localhost:8080` |
+| `AI_MODEL` | LLM model | `openrouter/deepseek/deepseek-v3.1-terminus` |
diff --git a/examples/python_agent_nodes/deep_research/__init__.py b/examples/python_agent_nodes/deep_research/__init__.py
new file mode 100644
index 00000000..47056d65
--- /dev/null
+++ b/examples/python_agent_nodes/deep_research/__init__.py
@@ -0,0 +1 @@
+"""Deep Research Agent with Recursive Planning."""
diff --git a/examples/python_agent_nodes/deep_research/main.py b/examples/python_agent_nodes/deep_research/main.py
new file mode 100644
index 00000000..c45d8d7e
--- /dev/null
+++ b/examples/python_agent_nodes/deep_research/main.py
@@ -0,0 +1,51 @@
+"""Deep Research Agent with Recursive Planning."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+import sys
+
+from agentfield import AIConfig, Agent
+
+if __package__ in (None, ""):
+ current_dir = Path(__file__).resolve().parent
+ if str(current_dir) not in sys.path:
+ sys.path.insert(0, str(current_dir))
+
+from routers import planning_router, research_router
+
+app = Agent(
+ node_id="deep-research",
+ agentfield_server=f"{os.getenv('AGENTFIELD_SERVER', 'http://localhost:8080')}",
+ ai_config=AIConfig(
+ model=os.getenv("AI_MODEL", "openrouter/deepseek/deepseek-v3.1-terminus"),
+ ),
+)
+
+app.include_router(planning_router)
+app.include_router(research_router)
+
+
+if __name__ == "__main__":
+ print("π¬ Deep Research Agent")
+ print("π§ Node ID: deep-research")
+ print(f"π Control Plane: {app.agentfield_server}")
+ print("\nπ― Architecture: Recursive Planning + Research Execution")
+ print(" 1. Recursive Task Decomposition β Breaks research into subtasks")
+ print(" 2. Topological Graph β Identifies dependencies and parallelization")
+ print(" 3. Research Execution β Tavily search with citation tracking")
+ print(" 4. Findings Synthesis β Structured results with sources")
+ print("\n⨠Features:")
+ print(" - Recursive task breakdown with configurable depth")
+ print(" - Automatic dependency detection")
+ print(" - Parallel execution planning")
+ print(" - Web search integration (Tavily)")
+ print(" - Citation tracking and structured findings")
+ print(" - Elegant and simple AgentField primitives")
+
+ port_env = os.getenv("PORT")
+ if port_env is None:
+ app.run(auto_port=True, host="::")
+ else:
+ app.run(port=int(port_env), host="::")
diff --git a/examples/python_agent_nodes/deep_research/requirements.txt b/examples/python_agent_nodes/deep_research/requirements.txt
new file mode 100644
index 00000000..afc26f24
--- /dev/null
+++ b/examples/python_agent_nodes/deep_research/requirements.txt
@@ -0,0 +1,3 @@
+pydantic>=2.7.4
+agentfield>=0.1.6
+tavily-python>=0.3.0
diff --git a/examples/python_agent_nodes/deep_research/routers/__init__.py b/examples/python_agent_nodes/deep_research/routers/__init__.py
new file mode 100644
index 00000000..650e25bd
--- /dev/null
+++ b/examples/python_agent_nodes/deep_research/routers/__init__.py
@@ -0,0 +1,6 @@
+"""Router modules for deep research agent."""
+
+from .planning import planning_router
+from .research import research_router
+
+__all__ = ["planning_router", "research_router"]
diff --git a/examples/python_agent_nodes/deep_research/routers/planning.py b/examples/python_agent_nodes/deep_research/routers/planning.py
new file mode 100644
index 00000000..500b4703
--- /dev/null
+++ b/examples/python_agent_nodes/deep_research/routers/planning.py
@@ -0,0 +1,641 @@
+"""Recursive planning router for deep research agent."""
+
+from __future__ import annotations
+
+import asyncio
+from typing import List
+
+from agentfield import AgentRouter
+
+from schemas import (
+ ResearchPlan,
+ ResearchReport,
+ SearchStrategy,
+ Subtask,
+ TaskDescriptions,
+ TaskDependenciesList,
+ TaskMergeList,
+ TaskResult,
+)
+
+
+planning_router = AgentRouter(prefix="planning")
+
+
+def _build_topological_groups(tasks: List[Subtask]) -> List[List[str]]:
+ """
+ Build topological groups using topological sort.
+
+ Groups tasks by dependency level - all tasks at the same level
+ can run in parallel. Returns list of groups, where each group
+ contains task IDs that can execute in parallel.
+ """
+ task_map = {task.task_id: task for task in tasks}
+
+ # Calculate in-degree for each task (how many dependencies it has)
+ in_degree = {task_id: len(task.dependencies) for task_id, task in task_map.items()}
+
+ # Topological sort - group by level
+ level_groups: List[List[str]] = []
+ remaining = set(task_map.keys())
+
+ while remaining:
+ # Find all tasks with no remaining dependencies (current level)
+ current_level = [task_id for task_id in remaining if in_degree[task_id] == 0]
+
+ if not current_level:
+ # Circular dependency or error - add remaining tasks
+ current_level = list(remaining)
+
+ level_groups.append(current_level)
+ remaining -= set(current_level)
+
+ # Decrease in-degree for tasks that depend on current level
+ for task_id in current_level:
+ for dep_task in tasks:
+ if task_id in dep_task.dependencies and dep_task.task_id in remaining:
+ in_degree[dep_task.task_id] -= 1
+
+ # Return groups with more than one task (parallelizable)
+ return [group for group in level_groups if len(group) > 1]
+
+
+def _get_execution_levels(tasks: List[Subtask]) -> List[List[str]]:
+ """
+ Get all execution levels using topological sort.
+
+ Returns ALL levels (not just parallelizable groups).
+ Level 0 = leaf tasks (no dependencies, need search)
+ Level 1+ = parent tasks (have dependencies, synthesize from children)
+ """
+ task_map = {task.task_id: task for task in tasks}
+
+ # Calculate in-degree for each task
+ in_degree = {task_id: len(task.dependencies) for task_id, task in task_map.items()}
+
+ # Topological sort - group by level
+ level_groups: List[List[str]] = []
+ remaining = set(task_map.keys())
+
+ while remaining:
+ # Find all tasks with no remaining dependencies (current level)
+ current_level = [task_id for task_id in remaining if in_degree[task_id] == 0]
+
+ if not current_level:
+ # Circular dependency or error - add remaining tasks
+ current_level = list(remaining)
+
+ level_groups.append(current_level)
+ remaining -= set(current_level)
+
+ # Decrease in-degree for tasks that depend on current level
+ for task_id in current_level:
+ for dep_task in tasks:
+ if task_id in dep_task.dependencies and dep_task.task_id in remaining:
+ in_degree[dep_task.task_id] -= 1
+
+ return level_groups
+
+
+@planning_router.reasoner()
+async def create_research_plan(
+ research_question: str,
+ max_depth: int = 3,
+ max_tasks_per_level: int = 5,
+) -> ResearchPlan:
+ """
+ Create a recursive research plan by breaking down a question into subtasks.
+
+ This reasoner recursively decomposes a research question into smaller,
+ manageable subtasks, forming a topological graph that can be executed
+ in parallel where dependencies allow.
+ """
+ # First, get initial breakdown from LLM - simplified to just descriptions
+ initial_response = await planning_router.ai(
+ system=(
+ "You are an expert research planner breaking down questions into a hierarchical task graph.\n\n"
+ "## PHILOSOPHY\n"
+ "You are creating two types of tasks:\n"
+ "1. **Leaf tasks** (no dependencies) = Specific questions that need web search\n"
+ " - These are atomic research questions answerable via web search\n"
+ " - Example: 'What is AgentField?' or 'What are AgentField's features?'\n\n"
+ "2. **Parent tasks** (have dependencies) = Questions answered by synthesizing children\n"
+ " - These combine answers from child tasks\n"
+ " - Example: 'Summarize AgentField' depends on 'What is AgentField?' and 'What are features?'\n\n"
+ "## CRITICAL: DISTINCT TASKS\n"
+ "Each task must be DISTINCT and NON-OVERLAPPING.\n"
+ "- Do NOT create tasks that ask the same question in different words\n"
+ "- Each task should cover a unique aspect or angle\n"
+ "- Avoid redundancy - if two tasks would find the same information, merge them\n\n"
+ "## DEPENDENCY MEANING\n"
+ "A task depends on another if it needs that task's ANSWER to proceed.\n"
+ "Dependencies mean: 'I need answers from those tasks to answer this.'\n\n"
+ "## YOUR TASK\n"
+ "Break the research question into 3-5 DISTINCT major research areas.\n"
+ "Each area should be a specific, searchable question (leaf task).\n"
+ "Ensure each task is unique and non-overlapping.\n"
+ "These will be refined recursively, and synthesis tasks will be created later.\n\n"
+ f"Maximum {max_tasks_per_level} tasks.\n\n"
+ "Return ONLY a JSON object with a 'descriptions' array."
+ ),
+ user=(
+ f"Research Question: {research_question}\n\n"
+ f"Break this into 3-5 DISTINCT major research areas as specific, searchable questions. "
+ f"Ensure each task is unique and covers different aspects. Return only the descriptions array."
+ ),
+ schema=TaskDescriptions,
+ )
+
+ # Build initial Subtask objects from descriptions
+ initial_plan = [
+ Subtask(
+ task_id=f"task_{idx + 1}",
+ description=desc,
+ dependencies=[],
+ depth=1,
+ can_parallelize=True,
+ )
+ for idx, desc in enumerate(initial_response.descriptions)
+ ]
+
+ # Recursively refine each task
+ all_tasks: List[Subtask] = []
+
+ async def refine_recursively(
+ task: Subtask, parent_context: str = ""
+ ) -> List[Subtask]:
+ """Recursively refine a task by calling the refine_task reasoner."""
+ if task.depth >= max_depth:
+ # Base case: task is specific enough
+ return [task]
+
+ # Directly call the refine_task reasoner - this creates a workflow node
+ refined_tasks = await refine_task(
+ task_description=task.description,
+ parent_context=parent_context or research_question,
+ current_depth=task.depth,
+ max_depth=max_depth,
+ )
+
+ # Update task IDs to maintain hierarchy and recursively refine
+ result_tasks: List[Subtask] = []
+ for idx, refined_task in enumerate(refined_tasks):
+ new_task_id = f"{task.task_id}_{idx + 1}"
+ new_task = Subtask(
+ task_id=new_task_id,
+ description=refined_task.description,
+ dependencies=refined_task.dependencies,
+ depth=task.depth + 1,
+ can_parallelize=len(refined_task.dependencies) == 0,
+ )
+
+ # Recursively refine if still not at max depth
+ if new_task.depth < max_depth:
+ further_refined = await refine_recursively(new_task, task.description)
+ result_tasks.extend(further_refined)
+ else:
+ result_tasks.append(new_task)
+
+ # If no refinement happened, return the original task
+ return result_tasks if result_tasks else [task]
+
+ # Refine initial tasks - can run in parallel since they're independent
+ refinement_tasks = [
+ refine_recursively(initial_task, research_question)
+ for initial_task in initial_plan
+ ]
+ refined_results = await asyncio.gather(*refinement_tasks)
+
+ # Flatten the results
+ for refined_list in refined_results:
+ all_tasks.extend(refined_list)
+
+ # Deduplicate similar/redundant tasks
+ tasks_for_dedup = [
+ {"task_id": task.task_id, "description": task.description} for task in all_tasks
+ ]
+ merge_response = await deduplicate_tasks(tasks_for_dedup, research_question)
+
+ # Apply merges: keep the keep_task_id, remove merge_task_ids, update dependencies
+ task_map = {task.task_id: task for task in all_tasks}
+ tasks_to_remove = set()
+
+ for merge in merge_response.merges:
+ if merge.keep_task_id in task_map:
+ # Update dependencies: if any task depends on a merged task, point to kept task
+ for task in all_tasks:
+ if merge.keep_task_id != task.task_id:
+ # Replace merged task IDs with kept task ID in dependencies
+ if merge.keep_task_id in task.dependencies:
+ # Already depends on kept task, remove merged ones
+ task.dependencies = [
+ dep
+ for dep in task.dependencies
+ if dep not in merge.merge_task_ids
+ ]
+ else:
+ # Replace merged task IDs with kept task ID
+ task.dependencies = [
+ merge.keep_task_id if dep in merge.merge_task_ids else dep
+ for dep in task.dependencies
+ ]
+
+ # Mark merged tasks for removal
+ tasks_to_remove.update(merge.merge_task_ids)
+
+ # Remove merged tasks
+ all_tasks = [task for task in all_tasks if task.task_id not in tasks_to_remove]
+
+ # Identify dependencies between tasks
+ tasks_for_analysis = [
+ {"task_id": task.task_id, "description": task.description} for task in all_tasks
+ ]
+ dependencies_response = await identify_dependencies(
+ tasks_for_analysis, research_question
+ )
+
+ # Update task dependencies
+ task_map = {task.task_id: task for task in all_tasks}
+ dependency_map = {
+ dep.task_id: dep.depends_on for dep in dependencies_response.dependencies
+ }
+
+ for task in all_tasks:
+ if task.task_id in dependency_map:
+ task.dependencies = dependency_map[task.task_id]
+ # Validate dependencies exist
+ task.dependencies = [dep for dep in task.dependencies if dep in task_map]
+ task.can_parallelize = len(task.dependencies) == 0
+
+ # Build topological groups using topological sort
+ parallelizable_groups = _build_topological_groups(all_tasks)
+
+ return ResearchPlan(
+ research_question=research_question,
+ tasks=all_tasks,
+ max_depth=max(task.depth for task in all_tasks) if all_tasks else 0,
+ total_tasks=len(all_tasks),
+ parallelizable_groups=parallelizable_groups,
+ )
+
+
+@planning_router.reasoner()
+async def refine_task(
+ task_description: str,
+ parent_context: str = "",
+ current_depth: int = 0,
+ max_depth: int = 3,
+) -> List[Subtask]:
+ """
+ Recursively refine a single task into smaller subtasks.
+
+ This is a helper reasoner that can be called recursively to break down
+ complex tasks into more manageable pieces. Each call creates a workflow node.
+ """
+
+ if current_depth >= max_depth:
+ # Return the task as-is if we've hit max depth
+ return [
+ Subtask(
+ task_id=f"leaf_{current_depth}",
+ description=task_description,
+ dependencies=[],
+ depth=current_depth,
+ can_parallelize=True,
+ )
+ ]
+
+ # Use AI to determine if task needs further breakdown - simplified schema
+ result_response = await planning_router.ai(
+ system=(
+ "You are a task decomposition expert. Break down research tasks into "
+ "smaller subtasks.\n\n"
+ "## CRITICAL: DISTINCT SUBTASKS\n"
+ "When breaking down, create DISTINCT, NON-OVERLAPPING subtasks:\n"
+ "- Each subtask should cover a unique aspect\n"
+ "- Do NOT create subtasks that ask the same question\n"
+ "- Ensure each subtask is independently researchable\n\n"
+ "Return ONLY a JSON object with a 'descriptions' array.\n"
+ "- If the task is specific enough, return it as a single-item array\n"
+ "- If it needs breakdown, return 2-4 DISTINCT smaller task descriptions\n"
+ "- Each description must be unique and non-overlapping\n\n"
+ f"Current depth: {current_depth} of {max_depth}. "
+ f"Only break down if still too complex."
+ ),
+ user=(
+ f"Task: {task_description}\n"
+ f"{f'Context: {parent_context}' if parent_context else ''}\n\n"
+ f"Break this into DISTINCT smaller tasks if needed, or return as-is if specific enough. "
+ f"Ensure each subtask is unique and non-overlapping."
+ ),
+ schema=TaskDescriptions,
+ )
+
+ # Build Subtask objects from descriptions
+ refined_tasks = [
+ Subtask(
+ task_id=f"subtask_{idx}",
+ description=desc,
+ dependencies=[],
+ depth=current_depth + 1,
+ can_parallelize=True,
+ )
+ for idx, desc in enumerate(result_response.descriptions)
+ ]
+
+ return refined_tasks
+
+
+@planning_router.reasoner()
+async def identify_dependencies(
+ tasks: List[dict], research_question: str
+) -> TaskDependenciesList:
+ """
+ Identify dependencies between tasks based on their descriptions.
+
+ Takes a list of tasks with id and description, and identifies which
+ tasks need results from other tasks before they can execute.
+ """
+ # Format tasks for the prompt
+ tasks_text = "\n".join(
+ f"- {task['task_id']}: {task['description']}" for task in tasks
+ )
+
+ response = await planning_router.ai(
+ system=(
+ "You are a dependency analysis expert identifying task relationships.\n\n"
+ "## DEPENDENCY PHILOSOPHY\n"
+ "A task depends on another if it needs that task's ANSWER to proceed.\n"
+ "Dependencies mean: 'I need answers from those tasks to answer this.'\n\n"
+ "## CRITICAL: AVOID REDUNDANT DEPENDENCIES\n"
+ "- Do NOT mark dependencies if tasks ask the same question\n"
+ "- Do NOT mark dependencies if a task can be answered independently\n"
+ "- Only mark dependencies when a task truly needs another task's answer\n"
+ "- If two tasks would find the same information, they should NOT depend on each other\n\n"
+ "## DEPENDENCY PATTERNS\n"
+ "- **Comparison tasks** depend on individual research tasks\n"
+ " Example: 'Compare X and Y' depends on 'What is X?' and 'What is Y?'\n\n"
+ "- **Synthesis tasks** depend on all component tasks\n"
+ " Example: 'Summarize AgentField' depends on 'What is AgentField?' and 'What are features?'\n\n"
+ "- **Analysis tasks** depend on data-gathering tasks\n"
+ " Example: 'Analyze trends' depends on 'What are the trends?'\n\n"
+ "## KEY PRINCIPLE\n"
+ "If a task can be answered by combining other tasks' findings, mark dependencies.\n"
+ "Leaf tasks (no dependencies) will be answered via web search.\n"
+ "Parent tasks (have dependencies) will synthesize from children's answers.\n\n"
+ "Return ONLY a JSON object with a 'dependencies' array. Each item:\n"
+ "- task_id: the task ID\n"
+ "- depends_on: array of task IDs it depends on (empty if independent/leaf task)\n\n"
+ "Be conservative - only mark dependencies when clearly needed. "
+ "Avoid redundant dependencies between similar tasks."
+ ),
+ user=(
+ f"Research Question: {research_question}\n\n"
+ f"Tasks:\n{tasks_text}\n\n"
+ f"Identify dependencies. For each task, determine if it needs answers from other tasks. "
+ f"DO NOT mark dependencies if tasks ask the same question or can be answered independently. "
+ f"Return dependency mappings for all tasks."
+ ),
+ schema=TaskDependenciesList,
+ )
+
+ return response
+
+
+@planning_router.reasoner()
+async def deduplicate_tasks(tasks: List[dict], research_question: str) -> TaskMergeList:
+ """
+ Identify and merge similar/redundant tasks.
+
+ Finds tasks that ask the same question or cover the same information,
+ and suggests which tasks to merge into others.
+ """
+ tasks_text = "\n".join(
+ f"- {task['task_id']}: {task['description']}" for task in tasks
+ )
+
+ response = await planning_router.ai(
+ system=(
+ "You are a task deduplication expert. Identify tasks that are redundant or ask the same question.\n\n"
+ "## YOUR TASK\n"
+ "Find tasks that:\n"
+ "- Ask the same question in different words\n"
+ "- Cover the same information/research area\n"
+ "- Are essentially duplicates\n\n"
+ "For each group of similar tasks:\n"
+ "- Keep the most specific/clear task (keep_task_id)\n"
+ "- Mark others to merge (merge_task_ids)\n\n"
+ "## IMPORTANT\n"
+ "- Only merge tasks that are truly redundant\n"
+ "- Keep tasks that cover different aspects, even if related\n"
+ "- Be conservative - when in doubt, don't merge\n\n"
+ "Return ONLY a JSON object with a 'merges' array. "
+ "Each merge has keep_task_id and merge_task_ids."
+ ),
+ user=(
+ f"Research Question: {research_question}\n\n"
+ f"Tasks:\n{tasks_text}\n\n"
+ f"Identify redundant tasks and suggest merges. Return merge instructions."
+ ),
+ schema=TaskMergeList,
+ )
+
+ return response
+
+
+@planning_router.reasoner()
+async def decide_search_strategy(
+ task_description: str,
+ research_question: str,
+ dependency_findings: List[TaskResult],
+) -> SearchStrategy:
+ """
+ Decide whether a parent task should:
+ - synthesize_only: Answer from children's findings alone
+ - enhanced_search: Need additional web search with dependency context
+
+ Examples:
+ - "Summarize X" when we have "What is X?" β synthesize_only
+ - "Compare X to Y" when we have "What is X?" β enhanced_search (need Y info)
+ """
+ children_text = "\n".join(
+ f"- {dep.task_id}: {dep.description}\n Answer: {dep.findings[:200]}..."
+ for dep in dependency_findings
+ )
+
+ response = await planning_router.ai(
+ system=(
+ "You are a task strategy expert deciding how to answer a parent task.\n\n"
+ "## TWO STRATEGIES\n"
+ "1. **synthesize_only**: Can answer from children's findings alone\n"
+ " - Example: 'Summarize X' when we have 'What is X?' and 'What are X's features?'\n"
+ " - Example: 'List all features' when we have individual feature tasks\n\n"
+ "2. **enhanced_search**: Needs additional web search with dependency context\n"
+ " - Example: 'Compare X to Y' when we have 'What is X?' but need Y info\n"
+ " - Example: 'Analyze market trends' when we have data but need external context\n"
+ " - Example: 'Find competitors' when we have product info but need market research\n\n"
+ "## DECISION CRITERIA\n"
+ "- If the task asks for information NOT in children β enhanced_search\n"
+ "- If the task can be answered by combining children β synthesize_only\n"
+ "- If the task needs external context/comparison β enhanced_search\n"
+ "- If the task is pure synthesis/summary β synthesize_only\n\n"
+ "Return ONLY a JSON object with 'strategy' ('synthesize_only' or 'enhanced_search') "
+ "and 'reasoning' (brief explanation)."
+ ),
+ user=(
+ f"Research Question: {research_question}\n\n"
+ f"Parent Task: {task_description}\n\n"
+ f"Child Task Answers:\n{children_text}\n\n"
+ f"Decide: Can we answer the parent task from children alone, or do we need additional search? "
+ f"Return strategy and reasoning."
+ ),
+ schema=SearchStrategy,
+ )
+
+ return response
+
+
+async def _execute_task_smart(
+ task: Subtask,
+ research_question: str,
+ completed_results: dict[str, TaskResult],
+) -> TaskResult:
+ """
+ Smart task executor that routes to search or synthesis based on dependencies and strategy.
+
+ - Leaf tasks (no deps) β search via execute_research_task
+ - Parent tasks (has deps) β decide strategy:
+ - synthesize_only β synthesize_from_dependencies
+ - enhanced_search β execute_research_task_with_context
+ """
+ from routers.research import (
+ execute_research_task,
+ execute_research_task_with_context,
+ synthesize_from_dependencies,
+ )
+
+ if not task.dependencies:
+ # Leaf task - search for answers
+ return await execute_research_task(
+ task_id=task.task_id,
+ task_description=task.description,
+ research_question=research_question,
+ )
+ else:
+ # Parent task - get dependency findings
+ dependency_findings = [
+ completed_results[dep_id]
+ for dep_id in task.dependencies
+ if dep_id in completed_results
+ ]
+
+ # Decide strategy: synthesize_only or enhanced_search
+ strategy = await decide_search_strategy(
+ task_description=task.description,
+ research_question=research_question,
+ dependency_findings=dependency_findings,
+ )
+
+ if strategy.strategy == "synthesize_only":
+ # Pure synthesis from children
+ return await synthesize_from_dependencies(
+ task_id=task.task_id,
+ task_description=task.description,
+ research_question=research_question,
+ dependency_findings=dependency_findings,
+ )
+ else:
+ # Enhanced search with dependency context
+ return await execute_research_task_with_context(
+ task_id=task.task_id,
+ task_description=task.description,
+ research_question=research_question,
+ dependency_findings=dependency_findings,
+ )
+
+
+@planning_router.reasoner()
+async def execute_deep_research(
+ research_question: str,
+ max_depth: int = 3,
+ max_tasks_per_level: int = 5,
+) -> ResearchReport:
+ """
+ Complete deep research orchestrator.
+
+ Takes a research question and:
+ 1. Creates recursive research plan with dependencies
+ 2. Executes tasks in topological order (leaf tasks search, parent tasks synthesize)
+ 3. Synthesizes final comprehensive report
+ """
+ # Step 1: Create research plan
+ plan = await create_research_plan(
+ research_question=research_question,
+ max_depth=max_depth,
+ max_tasks_per_level=max_tasks_per_level,
+ )
+
+ # Step 2: Get execution levels (topological order)
+ execution_levels = _get_execution_levels(plan.tasks)
+ task_map = {task.task_id: task for task in plan.tasks}
+ completed_results: dict[str, TaskResult] = {}
+
+ # Step 3: Execute tasks level by level
+ for level_idx, level_tasks in enumerate(execution_levels):
+ # Execute tasks in parallel within each level
+ execution_coroutines = [
+ _execute_task_smart(
+ task=task_map[task_id],
+ research_question=research_question,
+ completed_results=completed_results,
+ )
+ for task_id in level_tasks
+ if task_id in task_map
+ ]
+
+ if execution_coroutines:
+ level_results = await asyncio.gather(*execution_coroutines)
+ for result in level_results:
+ completed_results[result.task_id] = result
+
+ # Step 4: Synthesize final report
+ all_findings = list(completed_results.values())
+
+ # Build context for final synthesis
+ findings_text = "\n\n".join(
+ f"Task {r.task_id}: {r.description}\n"
+ f"Findings: {r.findings}\n"
+ f"Sources: {', '.join(r.sources) if r.sources else 'Synthesized from child tasks'}\n"
+ f"Confidence: {r.confidence}"
+ for r in all_findings
+ )
+
+ synthesis = await planning_router.ai(
+ system=(
+ "You are a research synthesis expert creating a comprehensive final report.\n\n"
+ "## CONTEXT\n"
+ "You are synthesizing findings from multiple research tasks into a final report.\n"
+ "Each task has been answered (either via web search for leaf tasks, or "
+ "synthesis for parent tasks).\n\n"
+ "## YOUR TASK\n"
+ "Create a comprehensive research report that:\n"
+ "1. Provides an executive summary answering the main research question\n"
+ "2. Organizes detailed findings by topic/theme\n"
+ "3. Draws conclusions and key insights\n"
+ "4. Assesses overall confidence based on all task findings\n\n"
+ "Structure the report clearly and comprehensively."
+ ),
+ user=(
+ f"Research Question: {research_question}\n\n"
+ f"All Task Findings:\n{findings_text}\n\n"
+ f"Create a comprehensive research report that synthesizes all findings "
+ f"to answer the research question. Include executive summary, detailed findings, "
+ f"conclusions, and overall confidence assessment."
+ ),
+ schema=ResearchReport,
+ )
+
+ # Attach detailed findings
+ synthesis.detailed_findings = all_findings
+
+ return synthesis
diff --git a/examples/python_agent_nodes/deep_research/routers/research.py b/examples/python_agent_nodes/deep_research/routers/research.py
new file mode 100644
index 00000000..dcc71a3a
--- /dev/null
+++ b/examples/python_agent_nodes/deep_research/routers/research.py
@@ -0,0 +1,355 @@
+"""Research execution router with Tavily integration."""
+
+from __future__ import annotations
+
+import os
+from typing import List
+
+from agentfield import AgentRouter
+
+from schemas import Citation, ResearchFindings, SearchQueries, TaskResult
+
+
+research_router = AgentRouter(prefix="research")
+
+
+@research_router.reasoner()
+async def generate_search_queries(
+ task_description: str,
+ research_question: str,
+ dependency_context: str = "",
+) -> SearchQueries:
+ """Generate focused search queries for a research task, optionally with dependency context."""
+ context_section = ""
+ if dependency_context:
+ context_section = (
+ "\n## DEPENDENCY CONTEXT\n"
+ "You have answers from child tasks. Use this context to enhance your search queries:\n"
+ f"{dependency_context}\n\n"
+ "Your queries should:\n"
+ "- Build on the provided context\n"
+ "- Fill gaps or find additional information needed\n"
+ "- Use context to make queries more specific and targeted\n"
+ )
+
+ response = await research_router.ai(
+ system=(
+ "You are a search query expert generating queries for a research task.\n\n"
+ "## CONTEXT\n"
+ "This task will be answered via web search.\n"
+ "Generate queries optimized for finding specific answers.\n\n"
+ f"{context_section}"
+ "## YOUR TASK\n"
+ "Generate 2-4 focused search queries optimized for web search.\n\n"
+ "Each query should be:\n"
+ "- Specific and targeted to answer the task question\n"
+ "- Use relevant keywords and terminology\n"
+ "- Cover different angles/aspects of the task\n"
+ "- Designed to find specific answers, not general information\n\n"
+ "Return ONLY a JSON object with a 'queries' array of search strings."
+ ),
+ user=(
+ f"Research Question: {research_question}\n"
+ f"Task (needs web search): {task_description}\n\n"
+ f"{f'Context from dependencies: {dependency_context}' if dependency_context else ''}\n\n"
+ f"Generate 2-4 search queries that will find specific answers to this question."
+ ),
+ schema=SearchQueries,
+ )
+ return response
+
+
+@research_router.reasoner()
+async def execute_search(queries: List[str]) -> dict:
+ """Execute Tavily search for given queries."""
+ try:
+ from tavily import TavilyClient
+ except ImportError:
+ raise ImportError("tavily-python not installed. Run: pip install tavily-python")
+
+ api_key = os.getenv("TAVILY_API_KEY")
+ if not api_key:
+ raise ValueError("TAVILY_API_KEY environment variable not set")
+
+ client = TavilyClient(api_key=api_key)
+
+ # Execute searches in parallel
+ all_results = []
+ for query in queries:
+ try:
+ result = client.search(
+ query=query,
+ search_depth="advanced",
+ max_results=5,
+ include_answer=False,
+ include_raw_content=True,
+ )
+ all_results.append(result)
+ except Exception as e:
+ # Continue with other queries if one fails
+ all_results.append({"error": str(e), "query": query})
+
+ # Combine results
+ combined = {
+ "results": [],
+ "queries": queries,
+ }
+
+ for result in all_results:
+ if "error" not in result and "results" in result:
+ combined["results"].extend(result["results"])
+
+ return combined
+
+
+@research_router.reasoner()
+async def synthesize_findings(
+ task_description: str,
+ research_question: str,
+ search_results: dict,
+) -> ResearchFindings:
+ """Synthesize search results into structured findings with citations."""
+ # Format search results for the prompt
+ results_text = ""
+ citations_data = []
+
+ for idx, result in enumerate(search_results.get("results", [])[:10]):
+ title = result.get("title", "Untitled")
+ url = result.get("url", "")
+ content = result.get("content", "")
+ raw_content = result.get("raw_content", "")
+
+ excerpt = raw_content[:500] if raw_content else content[:300]
+ results_text += f"\n[{idx + 1}] {title}\nURL: {url}\nContent: {excerpt}\n"
+
+ citations_data.append({"url": url, "title": title, "excerpt": excerpt})
+
+ response = await research_router.ai(
+ system=(
+ "You are a research synthesis expert. Synthesize search results into "
+ "structured findings with numbered points.\n\n"
+ "Format findings as:\n"
+ "1. First finding [1] (reference by number)\n"
+ "2. Second finding [2]\n"
+ "etc.\n\n"
+ "Extract citations: for each numbered reference, provide URL, title, and excerpt.\n\n"
+ "Assess confidence: 'high' if comprehensive, 'medium' if partial, 'low' if insufficient."
+ ),
+ user=(
+ f"Research Question: {research_question}\n"
+ f"Task: {task_description}\n\n"
+ f"Search Results:{results_text}\n\n"
+ f"Synthesize findings with numbered points and citations. "
+ f"Reference sources by number [1], [2], etc."
+ ),
+ schema=ResearchFindings,
+ )
+
+ # Map citations from search results
+ citation_map = {idx + 1: Citation(**c) for idx, c in enumerate(citations_data)}
+
+ # Update citations in response based on references in findings
+ final_citations = []
+ for i in range(1, len(citations_data) + 1):
+ if f"[{i}]" in response.findings and i in citation_map:
+ final_citations.append(citation_map[i])
+
+ response.citations = final_citations
+ return response
+
+
+@research_router.reasoner()
+async def execute_research_task(
+ task_id: str,
+ task_description: str,
+ research_question: str,
+) -> TaskResult:
+ """Orchestrate research execution: queries β search β synthesize."""
+ # Generate search queries
+ queries_response = await generate_search_queries(
+ task_description, research_question
+ )
+
+ # Execute search
+ search_results = await execute_search(queries_response.queries)
+
+ # Synthesize findings
+ findings_response = await synthesize_findings(
+ task_description, research_question, search_results
+ )
+
+ # Convert to TaskResult
+ sources = [f"{c.title} ({c.url})" for c in findings_response.citations]
+
+ return TaskResult(
+ task_id=task_id,
+ description=task_description,
+ findings=findings_response.findings,
+ sources=sources,
+ confidence=findings_response.confidence,
+ )
+
+
+@research_router.reasoner()
+async def synthesize_from_dependencies(
+ task_id: str,
+ task_description: str,
+ research_question: str,
+ dependency_findings: List[TaskResult],
+) -> TaskResult:
+ """
+ Synthesize answer for a parent task from its children's findings.
+
+ This is for tasks that have dependencies - they synthesize from
+ child task answers rather than searching the web.
+ """
+ # Format dependency findings for the prompt
+ children_text = "\n\n".join(
+ f"Child Task {idx + 1} ({dep.task_id}): {dep.description}\n"
+ f"Answer: {dep.findings}\n"
+ f"Sources: {', '.join(dep.sources) if dep.sources else 'N/A'}"
+ for idx, dep in enumerate(dependency_findings)
+ )
+
+ response = await research_router.ai(
+ system=(
+ "You are answering a PARENT TASK by synthesizing answers from child tasks.\n\n"
+ "## CONTEXT\n"
+ "This is a parent task - it depends on child tasks that have already been answered.\n"
+ "Your job is to combine the child answers to answer the parent question.\n\n"
+ "## IMPORTANT\n"
+ "- Do NOT search the web - use ONLY the provided child findings\n"
+ "- Synthesize and combine the child answers\n"
+ "- Create a coherent answer to the parent question\n"
+ "- Reference which child tasks contributed to each point\n\n"
+ "## OUTPUT FORMAT\n"
+ "Provide a structured answer with numbered points.\n"
+ "Reference child tasks like: 'Based on Child Task 1, ...'\n"
+ "Assess confidence based on completeness of child answers."
+ ),
+ user=(
+ f"Research Question: {research_question}\n\n"
+ f"Parent Task: {task_description}\n\n"
+ f"Child Task Answers:\n{children_text}\n\n"
+ f"Synthesize an answer to the parent task by combining the child task findings. "
+ f"Do not search - use only the provided child answers."
+ ),
+ schema=ResearchFindings,
+ )
+
+ # Collect sources from all dependencies
+ all_sources = []
+ for dep in dependency_findings:
+ all_sources.extend(dep.sources)
+
+ # Add references to child tasks
+ child_refs = [f"Child task {dep.task_id}" for dep in dependency_findings]
+ all_sources.extend(child_refs)
+
+ return TaskResult(
+ task_id=task_id,
+ description=task_description,
+ findings=response.findings,
+ sources=list(set(all_sources)), # Deduplicate
+ confidence=response.confidence,
+ )
+
+
+@research_router.reasoner()
+async def execute_research_task_with_context(
+ task_id: str,
+ task_description: str,
+ research_question: str,
+ dependency_findings: List[TaskResult],
+) -> TaskResult:
+ """
+ Execute research task with dependency context for enhanced search.
+
+ This is for parent tasks that need additional web search beyond
+ what their children provide. Uses dependency context to enhance queries.
+ """
+ # Format dependency context for query generation
+ dependency_context = "\n\n".join(
+ f"Child Task {idx + 1} ({dep.task_id}): {dep.description}\n"
+ f"Answer: {dep.findings}"
+ for idx, dep in enumerate(dependency_findings)
+ )
+
+ # Generate enhanced search queries with context
+ queries_response = await generate_search_queries(
+ task_description=task_description,
+ research_question=research_question,
+ dependency_context=dependency_context,
+ )
+
+ # Execute search
+ search_results = await execute_search(queries_response.queries)
+
+ # Synthesize findings (can reference both search results and dependency context)
+ results_text = ""
+ citations_data = []
+
+ for idx, result in enumerate(search_results.get("results", [])[:10]):
+ title = result.get("title", "Untitled")
+ url = result.get("url", "")
+ content = result.get("content", "")
+ raw_content = result.get("raw_content", "")
+
+ excerpt = raw_content[:500] if raw_content else content[:300]
+ results_text += f"\n[{idx + 1}] {title}\nURL: {url}\nContent: {excerpt}\n"
+
+ citations_data.append({"url": url, "title": title, "excerpt": excerpt})
+
+ # Synthesize with both search results and dependency context
+ response = await research_router.ai(
+ system=(
+ "You are a research synthesis expert. Synthesize findings from web search "
+ "AND dependency context into structured findings.\n\n"
+ "## CONTEXT\n"
+ "You have:\n"
+ "1. Web search results (new information)\n"
+ "2. Dependency context (answers from child tasks)\n\n"
+ "## YOUR TASK\n"
+ "Combine both sources to answer the task question comprehensively.\n"
+ "Format findings as:\n"
+ "1. First finding [1] (reference by number)\n"
+ "2. Second finding [2]\n"
+ "etc.\n\n"
+ "Extract citations: for each numbered reference, provide URL, title, and excerpt.\n\n"
+ "Assess confidence: 'high' if comprehensive, 'medium' if partial, 'low' if insufficient."
+ ),
+ user=(
+ f"Research Question: {research_question}\n"
+ f"Task: {task_description}\n\n"
+ f"Dependency Context (from child tasks):\n{dependency_context}\n\n"
+ f"Web Search Results:{results_text}\n\n"
+ f"Synthesize findings combining both sources. "
+ f"Reference web sources by number [1], [2], etc."
+ ),
+ schema=ResearchFindings,
+ )
+
+ # Map citations from search results
+ citation_map = {idx + 1: Citation(**c) for idx, c in enumerate(citations_data)}
+
+ # Update citations in response based on references in findings
+ final_citations = []
+ for i in range(1, len(citations_data) + 1):
+ if f"[{i}]" in response.findings and i in citation_map:
+ final_citations.append(citation_map[i])
+
+ response.citations = final_citations
+
+ # Collect sources from dependencies and search
+ all_sources = []
+ for dep in dependency_findings:
+ all_sources.extend(dep.sources)
+ all_sources.extend([f"{c.title} ({c.url})" for c in final_citations])
+
+ return TaskResult(
+ task_id=task_id,
+ description=task_description,
+ findings=response.findings,
+ sources=list(set(all_sources)), # Deduplicate
+ confidence=response.confidence,
+ )
diff --git a/examples/python_agent_nodes/deep_research/schemas.py b/examples/python_agent_nodes/deep_research/schemas.py
new file mode 100644
index 00000000..f7f7a8d4
--- /dev/null
+++ b/examples/python_agent_nodes/deep_research/schemas.py
@@ -0,0 +1,156 @@
+"""Schemas for deep research agent with recursive planning."""
+
+from __future__ import annotations
+
+from typing import List
+from pydantic import BaseModel, Field
+
+
+class Subtask(BaseModel):
+ """A single subtask in the research plan."""
+
+ task_id: str = Field(description="Unique identifier for this task")
+ description: str = Field(
+ description="Clear description of what needs to be researched"
+ )
+ dependencies: List[str] = Field(
+ default_factory=list,
+ description="List of task_ids that must complete before this task can start",
+ )
+ depth: int = Field(
+ default=0, description="Recursion depth - how many levels deep this task is"
+ )
+ can_parallelize: bool = Field(
+ default=True,
+ description="Whether this task can run in parallel with others (if no dependencies)",
+ )
+
+
+class SubtaskList(BaseModel):
+ """Wrapper for a list of subtasks - used for AI schema validation."""
+
+ tasks: List[Subtask] = Field(description="List of subtasks")
+
+
+class TaskDescriptions(BaseModel):
+ """Simple schema for task descriptions - just a list of strings."""
+
+ descriptions: List[str] = Field(
+ description="List of task descriptions, one per line. Each should be a clear, specific research task."
+ )
+
+
+class TaskDependencies(BaseModel):
+ """Dependency mapping for a single task."""
+
+ task_id: str = Field(description="The task ID")
+ depends_on: List[str] = Field(
+ default_factory=list,
+ description="List of task IDs that must complete before this task can start",
+ )
+
+
+class TaskDependenciesList(BaseModel):
+ """Wrapper for a list of task dependencies - used for AI schema validation."""
+
+ dependencies: List[TaskDependencies] = Field(
+ description="List of task dependency mappings"
+ )
+
+
+class ResearchPlan(BaseModel):
+ """Complete research plan with topological task graph."""
+
+ research_question: str = Field(description="The original research question")
+ tasks: List[Subtask] = Field(description="All subtasks in the plan")
+ max_depth: int = Field(
+ description="Maximum recursion depth reached during planning"
+ )
+ total_tasks: int = Field(description="Total number of tasks in the plan")
+ parallelizable_groups: List[List[str]] = Field(
+ default_factory=list,
+ description="Groups of task_ids that can be executed in parallel",
+ )
+
+
+class TaskResult(BaseModel):
+ """Result from executing a single research task."""
+
+ task_id: str
+ description: str
+ findings: str = Field(description="Research findings for this task")
+ sources: List[str] = Field(
+ default_factory=list, description="Sources or references used"
+ )
+ confidence: str = Field(description="Confidence level: 'high', 'medium', or 'low'")
+
+
+class ResearchReport(BaseModel):
+ """Final synthesized research report."""
+
+ research_question: str
+ executive_summary: str = Field(description="High-level summary of findings")
+ detailed_findings: List[TaskResult] = Field(
+ description="Detailed findings for each task"
+ )
+ conclusions: str = Field(description="Final conclusions and recommendations")
+ overall_confidence: str = Field(
+ description="Overall confidence: 'high', 'medium', or 'low'"
+ )
+
+
+class SearchQueries(BaseModel):
+ """Simple schema for search queries - list of strings."""
+
+ queries: List[str] = Field(
+ description="List of search query strings optimized for web search"
+ )
+
+
+class Citation(BaseModel):
+ """Citation for a source with URL, title, and excerpt."""
+
+ url: str = Field(description="Source URL")
+ title: str = Field(description="Source title")
+ excerpt: str = Field(description="Relevant excerpt from the source")
+
+
+class ResearchFindings(BaseModel):
+ """Structured research findings with numbered points and citations."""
+
+ findings: str = Field(
+ description="Research findings as numbered points (1. ... 2. ... etc.) with inline citations"
+ )
+ citations: List[Citation] = Field(
+ default_factory=list, description="List of citations referenced in findings"
+ )
+ confidence: str = Field(description="Confidence level: 'high', 'medium', or 'low'")
+
+
+class TaskMerge(BaseModel):
+ """Task merge instruction for deduplication."""
+
+ keep_task_id: str = Field(description="Task ID to keep (most specific)")
+ merge_task_ids: List[str] = Field(
+ description="Task IDs to merge into the kept task"
+ )
+
+
+class TaskMergeList(BaseModel):
+ """List of task merges for deduplication."""
+
+ merges: List[TaskMerge] = Field(
+ default_factory=list, description="List of task merge instructions"
+ )
+
+
+class SearchStrategy(BaseModel):
+ """Search strategy decision for a parent task."""
+
+ strategy: str = Field(
+ description="Strategy: 'synthesize_only' or 'enhanced_search'"
+ )
+ reasoning: str = Field(
+ default="",
+ description="Brief reasoning for the strategy choice",
+ )
-> **π Welcome Early Adopter!**
->
-> You've discovered AgentField before our official launch. We're currently in private beta, gathering feedback from early users to shape the future of the autonomous software. Feel free to explore and test, and we'd love to hear your thoughts! Share feedback via [GitHub Issues](https://github.com/Agent-Field/agentfield/issues) or email us at contact@agentfield.ai. Please note that features and APIs are still being refined before our public release.
-
-### Kubernetes for AI Agents - **Deploy, Scale, Observe, and Prove**
-
-Open-source (Apache-2.0) **control plane** that runs AI agents like microservices.
-Every agent gets **REST/gRPC APIs**, **async execution & webhooks**, **built-in observability**, and **cryptographic identity & audit**.
+# Kubernetes for AI Agents
+### **Deploy, Scale, Observe, and Prove.**
[](LICENSE)
[](https://go.dev/)
[](https://www.python.org/)
[](https://docs.docker.com/)
-**[π Docs](https://agentfield.ai/docs)** β’ **[β‘ Quickstart](#quickstart-in-60-seconds)** β’ **[π§ Why AgentField](#why-agentfield)**
+**[π Documentation](https://agentfield.ai/docs)** β’ **[β‘ Quick Start](#-quick-start-in-60-seconds)** β’ **[π§ Why AgentField](#-why-agentfield)**
---
-## π **Ship Production-Ready AI Agents in Minutes**
-
-β
**Write agents in Python/Go** (or any language via REST/gRPC)
-
-β
**Deploy independently** like microservices-zero coordination between teams
+> **π Welcome Early Adopter!**
+>
+> You've discovered AgentField before our official launch. We're currently in private beta, gathering feedback from early users to shape the future of the autonomous software. Feel free to explore and test, and we'd love to hear your thoughts! Share feedback via [GitHub Issues](https://github.com/Agent-Field/agentfield/issues) or email us at contact@agentfield.ai. Please note that features and APIs are still being refined before our public release.
-β
**Get production infrastructure automatically**:
-- **IAM & cryptographic audit trails** - W3C DIDs + Verifiable Credentials
-- **REST APIs, streaming, async queues** - auto-generated endpoints
-- **Built-in observability & metrics** - Prometheus + workflow DAGs
+---
-β
**Run anywhere**: local dev, Docker, Kubernetes, cloud
+## π What is AgentField?
-```bash
-curl -fsSL https://agentfield.ai/install.sh | bash && af init my-agent --defaults
-```
+**AgentField is "Kubernetes for AI Agents."**
-**[π Full Docs](https://agentfield.ai/docs)** β’ **[β‘ Quick Start](https://agentfield.ai/docs/quick-start)** β’ **[π― Examples](https://github.com/agentfield/agentfield-examples)**
+It is an open-source **Control Plane** that treats AI agents as first-class citizens. Instead of building fragile, monolithic scripts, AgentField lets you deploy agents as **independent microservices** that can discover each other, coordinate complex workflows, and scale infinitelyβall with built-in observability and cryptographic trust.
----
-## π Try AgentField in 2 Minutes
+Write standard Python (or Go). Get a production-grade distributed system automatically.
-### Option 1: Local Install
+```python
+from agentfield import Agent
-```bash
-# macOS/Linux - install CLI
-curl -fsSL https://agentfield.ai/install.sh | bash
+# 1. Define an Agent (It's just a microservice)
+app = Agent(node_id="researcher", model="gpt-4o")
-# Start control plane + create your first agent
-af init my-agent --defaults && cd my-agent
-af run
-```
+# 2. Create a Skill (Deterministic code)
+@app.skill()
+def fetch_url(url: str) -> str:
+ return requests.get(url).text
-### Option 2: Docker Compose
-
-```bash
-git clone https://github.com/agentfield/agentfield
-cd agentfield && docker compose up
+# 3. Create a Reasoner (AI-powered logic)
+# This automatically becomes a REST API endpoint: POST /execute/researcher.summarize
+@app.reasoner()
+async def summarize(url: str) -> dict:
+ content = fetch_url(url)
+ # Native AI call with structured output
+ return await app.ai(f"Summarize this content: {content}")
+
+# 4. Run it
+if __name__ == "__main__":
+ app.run()
```
-Your control plane is running at `http://localhost:8080`
+**What you get for free:**
+* β
**Instant API:** `POST /api/v1/execute/researcher.summarize`
+* β
**Durable Execution:** Resumes automatically if the server crashes.
+* β
**Observability:** You get a full execution DAG, metrics, and logs automatically.
+* β
**Audit:** Every step produces a cryptographically signed Verifiable Credential.
-
-
+
-