hi @MSJavaScript
I think you don’t have to choose between “not enough context” and “everything”. Separate what the executor can reach from what goes into its prompt. A few things that work well together:
- Hand off a typed brief, not free text. When the executor is a tool, its argument schema becomes the prompt the planner has to fill in. Required fields make the planner write down what it already decided:
from pydantic import BaseModel, Field
class TaskBrief(BaseModel):
goal: str = Field(description="What must be true when this step is done.")
inputs: list[str] = Field(description="File paths the executor must use. Paths, not contents.")
decisions: list[str] = Field(default_factory=list, description="Parameters already decided, do not change.")
done_when: str = Field(description="Acceptance criteria and what to report back.")
-
Pass paths, not content. Your executor already has run_shell / edit_text_file, so the planner can write plan.md / notes.md and send only the paths. Skills work the same way: give them to the executor directly, so the planner doesn’t have to summarize them.
-
Keep the planner history in the executor’s state instead of its prompt. A tool can read the parent state through ToolRuntime. Put the history into a separate state key and add a lookup tool, so the executor pulls only what it’s missing:
from typing import Literal
from langchain.agents import AgentState, create_agent
from langchain.messages import AnyMessage, HumanMessage, ToolMessage
from langchain.tools import ToolRuntime, tool
from langgraph.types import Command
class ExecutorState(AgentState):
planner_history: list[AnyMessage] # not sent to the model
class ExecutorReport(BaseModel):
status: Literal["done", "failed", "needs_context"]
summary: str
artifacts: list[str] = []
missing_context: list[str] = []
@tool
def lookup_planner_context(query: str, runtime: ToolRuntime[None, ExecutorState]) -> str:
"""Search the planner's earlier conversation for a missing fact. Use before guessing."""
hits = [m.text for m in runtime.state.get("planner_history", []) if query.lower() in m.text.lower()]
return "\n---\n".join(h[:800] for h in hits[-5:]) or "No match."
executor = create_agent(
model="anthropic:claude-sonnet-4-6",
tools=[run_shell, edit_text_file, *vasp_tools, lookup_planner_context],
state_schema=ExecutorState,
response_format=ExecutorReport,
system_prompt="Execute one step. Read referenced files yourself. If a fact is missing, "
"call lookup_planner_context, otherwise return status='needs_context'.",
)
@tool
def run_executor(brief: TaskBrief, runtime: ToolRuntime) -> Command:
"""Delegate ONE plan step to the executor."""
result = executor.invoke({
"messages": [HumanMessage(brief.model_dump_json())], # small prompt
"planner_history": runtime.state["messages"], # available on demand
})
report = result["structured_response"]
return Command(update={"messages": [
ToolMessage(report.model_dump_json(), tool_call_id=runtime.tool_call_id)
]})
The executor starts with just the brief, and the planner gets back only the short report, so both contexts stay small. With needs_context, the executor asks instead of guessing, which matters before a long VASP run.
- If you don’t want to build this yourself, Deep Agents subagents already do most of it. They share the parent’s filesystem, can have their own skills and response_format, and return only the final message. There’s also mode: “fork” (beta, deepagents>=0.7.13) for executors that continue work the planner already started:
from deepagents import create_deep_agent
agent = create_deep_agent(
model="anthropic:claude-sonnet-4-6",
tools=[run_shell, edit_text_file],
subagents=[{
"name": "vasp-executor",
"description": "Runs one VASP step. Pass file paths and fixed parameters.",
"tools": [*vasp_tools],
"skills": ["/skills/vasp/"],
"response_format": ExecutorReport,
# "mode": "fork",
}],
)
- You probably don’t need the separate Waiting Node. With a checkpointer, calling interrupt() inside the executor (even inside the tool) pauses the whole run, and you continue with Command(resume=…) on the same thread_id.
Docs: