Skip to content
Open
Show file tree
Hide file tree
Changes from 5 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions docs/handoffs.md
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,8 @@ handoff_obj = handoff(

When a handoff occurs, it's as though the new agent takes over the conversation, and gets to see the entire previous conversation history. If you want to change this, you can set an [`input_filter`][agents.handoffs.Handoff.input_filter]. An input filter is a function that receives the existing input via a [`HandoffInputData`][agents.handoffs.HandoffInputData], and must return a new `HandoffInputData`.

By default the runner now wraps the prior transcript inside a developer-role summary message (see [`RunConfig.nest_handoff_history`][agents.run.RunConfig.nest_handoff_history]). The summary appears inside a `<CONVERSATION HISTORY>` block that keeps appending new turns when multiple handoffs happen during the same run. That default only applies when neither the handoff nor the run supplies an explicit `input_filter`, so existing code that already customizes the payload (including the examples in this repository) keeps its current behavior without changes.

There are some common patterns (for example removing all tool calls from the history), which are implemented for you in [`agents.extensions.handoff_filters`][]

```python
Expand Down
3 changes: 3 additions & 0 deletions docs/running_agents.md
Original file line number Diff line number Diff line change
Expand Up @@ -51,11 +51,14 @@ The `run_config` parameter lets you configure some global settings for the agent
- [`model_settings`][agents.run.RunConfig.model_settings]: Overrides agent-specific settings. For example, you can set a global `temperature` or `top_p`.
- [`input_guardrails`][agents.run.RunConfig.input_guardrails], [`output_guardrails`][agents.run.RunConfig.output_guardrails]: A list of input or output guardrails to include on all runs.
- [`handoff_input_filter`][agents.run.RunConfig.handoff_input_filter]: A global input filter to apply to all handoffs, if the handoff doesn't already have one. The input filter allows you to edit the inputs that are sent to the new agent. See the documentation in [`Handoff.input_filter`][agents.handoffs.Handoff.input_filter] for more details.
- [`nest_handoff_history`][agents.run.RunConfig.nest_handoff_history]: When `True` (the default) the runner wraps the prior transcript in a developer-role summary message, placing the content inside a `<CONVERSATION HISTORY>` block while keeping the latest user turn separate before invoking the next agent. The block automatically appends new turns as subsequent handoffs occur. Set this to `False` or provide a custom handoff filter if you prefer to pass through the raw transcript. You can also call [`nest_handoff_history`](agents.extensions.handoff_filters.nest_handoff_history) from your own filters to reuse the default behavior. All [`Runner` methods](agents.run.Runner) automatically create a `RunConfig` when you do not pass one, so the quickstarts and examples pick up this default automatically, and any explicit [`Handoff.input_filter`][agents.handoffs.Handoff.input_filter] callbacks continue to override it.
- [`tracing_disabled`][agents.run.RunConfig.tracing_disabled]: Allows you to disable [tracing](tracing.md) for the entire run.
- [`trace_include_sensitive_data`][agents.run.RunConfig.trace_include_sensitive_data]: Configures whether traces will include potentially sensitive data, such as LLM and tool call inputs/outputs.
- [`workflow_name`][agents.run.RunConfig.workflow_name], [`trace_id`][agents.run.RunConfig.trace_id], [`group_id`][agents.run.RunConfig.group_id]: Sets the tracing workflow name, trace ID and trace group ID for the run. We recommend at least setting `workflow_name`. The group ID is an optional field that lets you link traces across multiple runs.
- [`trace_metadata`][agents.run.RunConfig.trace_metadata]: Metadata to include on all traces.

By default, the SDK now nests prior turns inside a developer summary message whenever an agent hands off to another agent. This reduces repeated assistant messages and keeps the most recent user turn explicit for the receiving agent. If you'd like to return to the legacy behavior, pass `RunConfig(nest_handoff_history=False)` or supply a `handoff_input_filter` that forwards the conversation exactly as you need.

## Conversations/chat threads

Calling any of the run methods can result in one or more agents running (and hence one or more LLM calls), but it represents a single logical turn in a chat conversation. For example:
Expand Down
17 changes: 15 additions & 2 deletions src/agents/_run_impl.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@
ToolOutputGuardrailTripwireTriggered,
UserError,
)
from .extensions.handoff_filters import nest_handoff_history
from .guardrail import InputGuardrail, InputGuardrailResult, OutputGuardrail, OutputGuardrailResult
from .handoffs import Handoff, HandoffInputData
from .items import (
Expand Down Expand Up @@ -998,8 +999,8 @@ async def execute_handoffs(
input_filter = handoff.input_filter or (
run_config.handoff_input_filter if run_config else None
)
if input_filter:
logger.debug("Filtering inputs for handoff")
handoff_input_data: HandoffInputData | None = None
if input_filter or run_config.nest_handoff_history:
handoff_input_data = HandoffInputData(
input_history=tuple(original_input)
if isinstance(original_input, list)
Expand All @@ -1008,6 +1009,9 @@ async def execute_handoffs(
new_items=tuple(new_step_items),
run_context=context_wrapper,
)

if input_filter and handoff_input_data is not None:
logger.debug("Filtering inputs for handoff")
if not callable(input_filter):
_error_tracing.attach_error_to_span(
span_handoff,
Expand Down Expand Up @@ -1037,6 +1041,15 @@ async def execute_handoffs(
)
pre_step_items = list(filtered.pre_handoff_items)
new_step_items = list(filtered.new_items)
elif run_config.nest_handoff_history and handoff_input_data is not None:
nested = nest_handoff_history(handoff_input_data)
original_input = (
nested.input_history
if isinstance(nested.input_history, str)
else list(nested.input_history)
)
pre_step_items = list(nested.pre_handoff_items)
new_step_items = list(nested.new_items)

return SingleStepResult(
original_input=original_input,
Expand Down
182 changes: 182 additions & 0 deletions src/agents/extensions/handoff_filters.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,14 @@
from __future__ import annotations

import json
from copy import deepcopy
from typing import Any, cast

from ..handoffs import HandoffInputData
from ..items import (
HandoffCallItem,
HandoffOutputItem,
ItemHelpers,
ReasoningItem,
RunItem,
ToolCallItem,
Expand Down Expand Up @@ -34,6 +39,183 @@ def remove_all_tools(handoff_input_data: HandoffInputData) -> HandoffInputData:
)


_CONVERSATION_HISTORY_START = "<CONVERSATION HISTORY>"
_CONVERSATION_HISTORY_END = "</CONVERSATION HISTORY>"


def nest_handoff_history(handoff_input_data: HandoffInputData) -> HandoffInputData:
"""Summarizes the previous transcript into a developer message for the next agent."""

normalized_history = _normalize_input_history(handoff_input_data.input_history)
flattened_history = _flatten_nested_history_messages(normalized_history)
pre_items_as_inputs = [
_run_item_to_plain_input(item) for item in handoff_input_data.pre_handoff_items
]
new_items_as_inputs = [_run_item_to_plain_input(item) for item in handoff_input_data.new_items]
transcript = flattened_history + pre_items_as_inputs + new_items_as_inputs

developer_message = _build_developer_message(transcript)
latest_user = _find_latest_user_turn(transcript)
history_items: list[TResponseInputItem] = [developer_message]
if latest_user is not None:
history_items.append(latest_user)

filtered_pre_items = tuple(
item
for item in handoff_input_data.pre_handoff_items
if _get_run_item_role(item) != "assistant"
)

return handoff_input_data.clone(
input_history=tuple(history_items),
pre_handoff_items=filtered_pre_items,
)


def _normalize_input_history(
input_history: str | tuple[TResponseInputItem, ...],
) -> list[TResponseInputItem]:
if isinstance(input_history, str):
return ItemHelpers.input_to_new_input_list(input_history)
return [deepcopy(item) for item in input_history]


def _run_item_to_plain_input(run_item: RunItem) -> TResponseInputItem:
return deepcopy(run_item.to_input_item())


def _build_developer_message(transcript: list[TResponseInputItem]) -> TResponseInputItem:
transcript_copy = [deepcopy(item) for item in transcript]
if transcript_copy:
summary_lines = [
f"{idx + 1}. {_format_transcript_item(item)}"
for idx, item in enumerate(transcript_copy)
]
else:
summary_lines = ["(no previous turns recorded)"]

content_lines = [_CONVERSATION_HISTORY_START, *summary_lines, _CONVERSATION_HISTORY_END]
content = "\n".join(content_lines)
developer_message: dict[str, Any] = {
"role": "developer",
"content": content,
}
return cast(TResponseInputItem, developer_message)


def _format_transcript_item(item: TResponseInputItem) -> str:
role = item.get("role")
if isinstance(role, str):
prefix = role
name = item.get("name")
if isinstance(name, str) and name:
prefix = f"{prefix} ({name})"
content_str = _stringify_content(item.get("content"))
return f"{prefix}: {content_str}" if content_str else prefix

item_type = item.get("type", "item")
rest = {k: v for k, v in item.items() if k != "type"}
try:
serialized = json.dumps(rest, ensure_ascii=False, default=str)
except TypeError:
serialized = str(rest)
return f"{item_type}: {serialized}" if serialized else str(item_type)


def _stringify_content(content: Any) -> str:
if content is None:
return ""
if isinstance(content, str):
return content
try:
return json.dumps(content, ensure_ascii=False, default=str)
except TypeError:
return str(content)


def _find_latest_user_turn(
transcript: list[TResponseInputItem],
) -> TResponseInputItem | None:
for item in reversed(transcript):
if item.get("role") == "user":
return deepcopy(item)
return None


def _flatten_nested_history_messages(
items: list[TResponseInputItem],
) -> list[TResponseInputItem]:
flattened: list[TResponseInputItem] = []
for item in items:
nested_transcript = _extract_nested_history_transcript(item)
if nested_transcript is not None:
flattened.extend(nested_transcript)
continue
flattened.append(deepcopy(item))
return flattened


def _extract_nested_history_transcript(
item: TResponseInputItem,
) -> list[TResponseInputItem] | None:
if item.get("role") != "developer":
return None
content = item.get("content")
if not isinstance(content, str):
return None
start_idx = content.find(_CONVERSATION_HISTORY_START)
end_idx = content.find(_CONVERSATION_HISTORY_END)
if start_idx == -1 or end_idx == -1 or end_idx <= start_idx:
return None
start_idx += len(_CONVERSATION_HISTORY_START)
body = content[start_idx:end_idx]
lines = [line.strip() for line in body.splitlines() if line.strip()]
parsed: list[TResponseInputItem] = []
for line in lines:
parsed_item = _parse_summary_line(line)
if parsed_item is not None:
parsed.append(parsed_item)
return parsed


def _parse_summary_line(line: str) -> TResponseInputItem | None:
stripped = line.strip()
if not stripped:
return None
dot_index = stripped.find(".")
if dot_index != -1 and stripped[:dot_index].isdigit():
stripped = stripped[dot_index + 1 :].lstrip()
role_part, sep, remainder = stripped.partition(":")
if not sep:
return None
role_text = role_part.strip()
if not role_text:
return None
role, name = _split_role_and_name(role_text)
reconstructed: dict[str, Any] = {"role": role}
if name:
reconstructed["name"] = name
content = remainder.strip()
if content:
reconstructed["content"] = content
return cast(TResponseInputItem, reconstructed)


def _split_role_and_name(role_text: str) -> tuple[str, str | None]:
if role_text.endswith(")") and "(" in role_text:
open_idx = role_text.rfind("(")
possible_name = role_text[open_idx + 1 : -1].strip()
role_candidate = role_text[:open_idx].strip()
if possible_name:
return (role_candidate or "developer", possible_name)
return (role_text or "developer", None)


def _get_run_item_role(run_item: RunItem) -> str | None:
role_candidate = run_item.to_input_item().get("role")
return role_candidate if isinstance(role_candidate, str) else None


def _remove_tools_from_items(items: tuple[RunItem, ...]) -> tuple[RunItem, ...]:
filtered_items = []
for item in items:
Expand Down
5 changes: 5 additions & 0 deletions src/agents/run.py
Original file line number Diff line number Diff line change
Expand Up @@ -196,6 +196,11 @@ class RunConfig:
agent. See the documentation in `Handoff.input_filter` for more details.
"""

nest_handoff_history: bool = True
"""Wrap prior run history in a developer message before handing off when no custom input
filter is set. Set to False to preserve the raw transcript behavior from previous releases.
"""

input_guardrails: list[InputGuardrail[Any]] | None = None
"""A list of input guardrails to run on the initial run input."""

Expand Down
Loading