Add enforce_row_independence parameter for agent maps (#97)

Author: jackwildman

src/everyrow/generated/api/operations/single_agent_operations_single_agent_post.py

@@ -190,6 +190,7 @@ async def asyncio_detailed(
     )
 
     response = await client.get_async_httpx_client().request(**kwargs)
+
     return _build_response(client=client, response=response)
 
 

src/everyrow/generated/models/agent_map_operation.py

@@ -42,6 +42,10 @@ class AgentMapOperation:
             not set.
         include_research (bool | None | Unset): Include research notes in the response. Required when effort_level is
             not set.
+        enforce_row_independence (bool | Unset): If True, each agent runs completely independently without being
+            affected by other agents. Disables adaptive budget adjustment and straggler management, ensuring agents are not
+            hurried or given iteration limits based on other agents' progress. Use this when consistent per-row behavior is
+            more important than overall throughput. Default: False.
     """
 
     input_: AgentMapOperationInputType2 | list[AgentMapOperationInputType1Item] | UUID
@@ -53,6 +57,7 @@ class AgentMapOperation:
     join_with_input: bool | Unset = True
     iteration_budget: int | None | Unset = UNSET
     include_research: bool | None | Unset = UNSET
+    enforce_row_independence: bool | Unset = False
     additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict)
 
     def to_dict(self) -> dict[str, Any]:
@@ -118,6 +123,8 @@ def to_dict(self) -> dict[str, Any]:
         else:
             include_research = self.include_research
 
+        enforce_row_independence = self.enforce_row_independence
+
         field_dict: dict[str, Any] = {}
         field_dict.update(self.additional_properties)
         field_dict.update(
@@ -140,6 +147,8 @@ def to_dict(self) -> dict[str, Any]:
             field_dict["iteration_budget"] = iteration_budget
         if include_research is not UNSET:
             field_dict["include_research"] = include_research
+        if enforce_row_independence is not UNSET:
+            field_dict["enforce_row_independence"] = enforce_row_independence
 
         return field_dict
 
@@ -271,6 +280,8 @@ def _parse_include_research(data: object) -> bool | None | Unset:
 
         include_research = _parse_include_research(d.pop("include_research", UNSET))
 
+        enforce_row_independence = d.pop("enforce_row_independence", UNSET)
+
         agent_map_operation = cls(
             input_=input_,
             task=task,
@@ -281,6 +292,7 @@ def _parse_include_research(data: object) -> bool | None | Unset:
             join_with_input=join_with_input,
             iteration_budget=iteration_budget,
             include_research=include_research,
+            enforce_row_independence=enforce_row_independence,
         )
 
         agent_map_operation.additional_properties = d

src/everyrow/ops.py

@@ -111,9 +111,7 @@ async def create_scalar_artifact(input: BaseModel, session: Session) -> UUID:
         data=CreateArtifactRequestDataType1.from_dict(input.model_dump()),
         session_id=session.session_id,
     )
-    response = await create_artifact_artifacts_post.asyncio(
-        client=session.client, body=body
-    )
+    response = await create_artifact_artifacts_post.asyncio(client=session.client, body=body)
     response = handle_response(response)
     return response.artifact_id
 
@@ -125,9 +123,7 @@ async def create_table_artifact(input: DataFrame, session: Session) -> UUID:
         data=[CreateArtifactRequestDataType0Item.from_dict(r) for r in records],
         session_id=session.session_id,
     )
-    response = await create_artifact_artifacts_post.asyncio(
-        client=session.client, body=body
-    )
+    response = await create_artifact_artifacts_post.asyncio(client=session.client, body=body)
     response = handle_response(response)
     return response.artifact_id
 
@@ -231,35 +227,25 @@ async def single_agent_async[T: BaseModel](
     return_table: bool = False,
 ) -> EveryrowTask[T]:
     """Submit a single_agent task asynchronously."""
-    input_data = _prepare_single_input(
-        input, SingleAgentOperationInputType1Item, SingleAgentOperationInputType2
-    )
+    input_data = _prepare_single_input(input, SingleAgentOperationInputType1Item, SingleAgentOperationInputType2)
 
     # Build the operation body with either preset or custom params
     body = SingleAgentOperation(
         input_=input_data,  # type: ignore
         task=task,
         session_id=session.session_id,
-        response_schema=SingleAgentOperationResponseSchemaType0.from_dict(
-            response_model.model_json_schema()
-        ),
-        effort_level=PublicEffortLevel(effort_level.value)
-        if effort_level is not None
-        else UNSET,
+        response_schema=SingleAgentOperationResponseSchemaType0.from_dict(response_model.model_json_schema()),
+        effort_level=PublicEffortLevel(effort_level.value) if effort_level is not None else UNSET,
         llm=LLMEnumPublic(llm.value) if llm is not None else UNSET,
         iteration_budget=iteration_budget if iteration_budget is not None else UNSET,
         include_research=include_research if include_research is not None else UNSET,
         return_list=return_table,
     )
 
-    response = await single_agent_operations_single_agent_post.asyncio(
-        client=session.client, body=body
-    )
+    response = await single_agent_operations_single_agent_post.asyncio(client=session.client, body=body)
     response = handle_response(response)
 
-    cohort_task: EveryrowTask[T] = EveryrowTask(
-        response_model=response_model, is_map=False, is_expand=return_table
-    )
+    cohort_task: EveryrowTask[T] = EveryrowTask(response_model=response_model, is_map=False, is_expand=return_table)
     cohort_task.set_submitted(response.task_id, response.session_id, session.client)
     return cohort_task
 
@@ -275,6 +261,7 @@ async def agent_map(
     llm: LLM | None = None,
     iteration_budget: int | None = None,
     include_research: bool | None = None,
+    enforce_row_independence: bool = False,
     response_model: type[BaseModel] = DefaultAgentResponse,
 ) -> TableResult:
     """Execute an AI agent task on each row of the input table.
@@ -305,6 +292,7 @@ async def agent_map(
                 llm=llm,
                 iteration_budget=iteration_budget,
                 include_research=include_research,
+                enforce_row_independence=enforce_row_independence,
                 response_model=response_model,
             )
             result = await cohort_task.await_result()
@@ -319,6 +307,7 @@ async def agent_map(
         llm=llm,
         iteration_budget=iteration_budget,
         include_research=include_research,
+        enforce_row_independence=enforce_row_independence,
         response_model=response_model,
     )
     result = await cohort_task.await_result()
@@ -335,6 +324,7 @@ async def agent_map_async(
     llm: LLM | None = None,
     iteration_budget: int | None = None,
     include_research: bool | None = None,
+    enforce_row_independence: bool = False,
     response_model: type[BaseModel] = DefaultAgentResponse,
 ) -> EveryrowTask[BaseModel]:
     """Submit an agent_map task asynchronously."""
@@ -345,26 +335,19 @@ async def agent_map_async(
         input_=input_data,  # type: ignore
         task=task,
         session_id=session.session_id,
-        response_schema=AgentMapOperationResponseSchemaType0.from_dict(
-            response_model.model_json_schema()
-        ),
-        effort_level=PublicEffortLevel(effort_level.value)
-        if effort_level is not None
-        else UNSET,
+        response_schema=AgentMapOperationResponseSchemaType0.from_dict(response_model.model_json_schema()),
+        effort_level=PublicEffortLevel(effort_level.value) if effort_level is not None else UNSET,
         llm=LLMEnumPublic(llm.value) if llm is not None else UNSET,
         iteration_budget=iteration_budget if iteration_budget is not None else UNSET,
         include_research=include_research if include_research is not None else UNSET,
         join_with_input=True,
+        enforce_row_independence=enforce_row_independence,
     )
 
-    response = await agent_map_operations_agent_map_post.asyncio(
-        client=session.client, body=body
-    )
+    response = await agent_map_operations_agent_map_post.asyncio(client=session.client, body=body)
     response = handle_response(response)
 
-    cohort_task = EveryrowTask(
-        response_model=response_model, is_map=True, is_expand=False
-    )
+    cohort_task = EveryrowTask(response_model=response_model, is_map=True, is_expand=False)
     cohort_task.set_submitted(response.task_id, response.session_id, session.client)
     return cohort_task
 
@@ -403,9 +386,7 @@ async def screen[T: BaseModel](
             if isinstance(result, TableResult):
                 return result
             raise EveryrowError("Screen task did not return a table result")
-    cohort_task = await screen_async(
-        task=task, session=session, input=input, response_model=response_model
-    )
+    cohort_task = await screen_async(task=task, session=session, input=input, response_model=response_model)
     result = await cohort_task.await_result()
     if isinstance(result, TableResult):
         return result
@@ -426,14 +407,10 @@ async def screen_async[T: BaseModel](
         input_=input_data,  # type: ignore
         task=task,
         session_id=session.session_id,
-        response_schema=ScreenOperationResponseSchemaType0.from_dict(
-            actual_response_model.model_json_schema()
-        ),
+        response_schema=ScreenOperationResponseSchemaType0.from_dict(actual_response_model.model_json_schema()),
     )
 
-    response = await screen_operations_screen_post.asyncio(
-        client=session.client, body=body
-    )
+    response = await screen_operations_screen_post.asyncio(client=session.client, body=body)
     response = handle_response(response)
 
     cohort_task: EveryrowTask[T] = EveryrowTask(
@@ -520,9 +497,7 @@ async def rank_async[T: BaseModel](
         # Validate that field_name exists in the model
         properties = response_schema.get("properties", {})
         if field_name not in properties:
-            raise ValueError(
-                f"Field {field_name} not in response model {response_model.__name__}"
-            )
+            raise ValueError(f"Field {field_name} not in response model {response_model.__name__}")
     else:
         # Build a minimal JSON schema with just the sort field
         json_type_map = {
@@ -533,9 +508,7 @@ async def rank_async[T: BaseModel](
         }
         response_schema = {
             "type": "object",
-            "properties": {
-                field_name: {"type": json_type_map.get(field_type, field_type)}
-            },
+            "properties": {field_name: {"type": json_type_map.get(field_type, field_type)}},
             "required": [field_name],
         }
 
@@ -641,9 +614,7 @@ async def merge_async(
         session_id=session.session_id,
     )
 
-    response = await merge_operations_merge_post.asyncio(
-        client=session.client, body=body
-    )
+    response = await merge_operations_merge_post.asyncio(client=session.client, body=body)
     response = handle_response(response)
 
     cohort_task = EveryrowTask(response_model=BaseModel, is_map=True, is_expand=False)
@@ -707,9 +678,7 @@ async def dedupe_async(
         session_id=session.session_id,
     )
 
-    response = await dedupe_operations_dedupe_post.asyncio(
-        client=session.client, body=body
-    )
+    response = await dedupe_operations_dedupe_post.asyncio(client=session.client, body=body)
     response = handle_response(response)
 
     cohort_task = EveryrowTask(response_model=BaseModel, is_map=True, is_expand=False)

src/everyrow/session.py

@@ -63,9 +63,7 @@ async def create_session(
     try:
         response = await create_session_endpoint_sessions_post.asyncio(
             client=client,
-            body=CreateSession(
-                name=name or f"everyrow-sdk-session-{datetime.now().isoformat()}"
-            ),
+            body=CreateSession(name=name or f"everyrow-sdk-session-{datetime.now().isoformat()}"),
         )
         response = handle_response(response)
         session = Session(client=client, session_id=response.session_id)