agent/agent-framework
2
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

Python: name changes executed (#607)

Browse Source 
* name changes executed

* updated adr to accepted

* renamed openai base config

* renamed openai config to mixin

* added renames in user docs

* reverted mcperror

* fix tests

* remove sse from tests
This commit is contained in:
Eduard van Valkenburg
2025-09-04 17:00:38 +02:00
committed by GitHub
Unverified
parent 6310ca5be0
commit 40ab6e9d67
100 changed files with 1223 additions and 1100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/packages/workflow/agent_framework_workflow/_agent.py
+14 -14
View File
@@ -7,14 +7,14 @@ from datetime import datetime
from typing import TYPE_CHECKING, Any, ClassVar, TypedDict, cast
from agent_framework import (
AgentBase,
AgentRunResponse,
AgentRunResponseUpdate,
AgentThread,
BaseAgent,
ChatMessage,
ChatRole,
FunctionCallContent,
FunctionResultContent,
Role,
TextContent,
UsageDetails,
)
@@ -34,8 +34,8 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
class WorkflowAgent(AgentBase):
"""An `AIAgent` subclass that wraps a workflow and exposes it as an agent."""
class WorkflowAgent(BaseAgent):
"""An `Agent` subclass that wraps a workflow and exposes it as an agent."""
# Class variable for the request info function name
REQUEST_INFO_FUNCTION_NAME: ClassVar[str] = "request_info"
@@ -65,11 +65,11 @@ class WorkflowAgent(AgentBase):
id: Unique identifier for the agent. If None, will be generated.
name: Optional name for the agent.
description: Optional description of the agent.
**kwargs: Additional keyword arguments passed to AgentBase.
**kwargs: Additional keyword arguments passed to BaseAgent.
"""
if id is None:
id = f"WorkflowAgent_{uuid.uuid4().hex[:8]}"
# Initialize with standard AgentBase parameters first
# Initialize with standard BaseAgent parameters first
kwargs["workflow"] = workflow
# Validate the workflow's start executor can handle agent-facing message inputs
@@ -107,7 +107,7 @@ class WorkflowAgent(AgentBase):
thread = thread or self.get_new_thread()
response_id = str(uuid.uuid4())
async for update in self._run_streaming_impl(input_messages, response_id):
async for update in self._run_stream_impl(input_messages, response_id):
response_updates.append(update)
# Convert updates to final response.
@@ -119,7 +119,7 @@ class WorkflowAgent(AgentBase):
return response
async def run_streaming(
async def run_stream(
self,
messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
*,
@@ -141,7 +141,7 @@ class WorkflowAgent(AgentBase):
response_updates: list[AgentRunResponseUpdate] = []
response_id = str(uuid.uuid4())
async for update in self._run_streaming_impl(input_messages, response_id):
async for update in self._run_stream_impl(input_messages, response_id):
response_updates.append(update)
yield update
@@ -152,7 +152,7 @@ class WorkflowAgent(AgentBase):
await self._notify_thread_of_new_messages(thread, input_messages)
await self._notify_thread_of_new_messages(thread, response.messages)
async def _run_streaming_impl(
async def _run_stream_impl(
self,
input_messages: list[ChatMessage],
response_id: str,
@@ -188,7 +188,7 @@ class WorkflowAgent(AgentBase):
else:
# Execute workflow with streaming (initial run or no function responses)
# Pass the new input messages directly to the workflow
event_stream = self.workflow.run_streaming(input_messages)
event_stream = self.workflow.run_stream(input_messages)
# Process events from the stream
async for event in event_stream:
@@ -206,7 +206,7 @@ class WorkflowAgent(AgentBase):
return []
if isinstance(messages, str):
return [ChatMessage(role=ChatRole.USER, contents=[TextContent(text=messages)])]
return [ChatMessage(role=Role.USER, contents=[TextContent(text=messages)])]
if isinstance(messages, ChatMessage):
return [messages]
@@ -214,7 +214,7 @@ class WorkflowAgent(AgentBase):
normalized = []
for msg in messages:
if isinstance(msg, str):
normalized.append(ChatMessage(role=ChatRole.USER, contents=[TextContent(text=msg)]))
normalized.append(ChatMessage(role=Role.USER, contents=[TextContent(text=msg)]))
elif isinstance(msg, ChatMessage):
normalized.append(msg)
return normalized
@@ -250,7 +250,7 @@ class WorkflowAgent(AgentBase):
)
return AgentRunResponseUpdate(
contents=[function_call],
role=ChatRole.ASSISTANT,
role=Role.ASSISTANT,
author_name=self.name,
response_id=response_id,
message_id=str(uuid.uuid4()),
python/packages/workflow/agent_framework_workflow/_executor.py
+4 -4
View File
@@ -12,7 +12,7 @@ from typing import TYPE_CHECKING, Any, Generic, TypeVar, Union, get_args, get_or
if TYPE_CHECKING:
from ._workflow import Workflow
from agent_framework import AgentRunResponse, AgentRunResponseUpdate, AgentThread, AIAgent, ChatMessage
from agent_framework import AgentProtocol, AgentRunResponse, AgentRunResponseUpdate, AgentThread, ChatMessage
from agent_framework._pydantic import AFBaseModel
from pydantic import Field
@@ -789,7 +789,7 @@ class AgentExecutor(Executor):
def __init__(
self,
agent: AIAgent,
agent: AgentProtocol,
*,
agent_thread: AgentThread | None = None,
streaming: bool = False,
@@ -818,7 +818,7 @@ class AgentExecutor(Executor):
if request.should_respond:
if self._streaming:
updates: list[AgentRunResponseUpdate] = []
async for update in self._agent.run_streaming(
async for update in self._agent.run_stream(
self._cache,
thread=self._agent_thread,
):
@@ -894,7 +894,7 @@ class WorkflowExecutor(Executor):
try:
# Run the sub-workflow and collect all events
events = [event async for event in self.workflow.run_streaming(input_data)]
events = [event async for event in self.workflow.run_stream(input_data)]
# Count requests and initialize response tracking
request_count = 0
python/packages/workflow/agent_framework_workflow/_magentic.py
+60 -60
View File
@@ -14,16 +14,16 @@ from typing import Annotated, Any, Literal, Protocol, TypeVar, Union, cast
from uuid import uuid4
from agent_framework import (
AgentProtocol,
AgentRunResponse,
AgentRunResponseUpdate,
AIAgent,
ChatClient,
ChatClientProtocol,
ChatMessage,
ChatRole,
FunctionCallContent,
FunctionResultContent,
Role,
)
from agent_framework._agents import AgentBase
from agent_framework._agents import BaseAgent
from agent_framework._pydantic import AFBaseModel
from pydantic import BaseModel, ConfigDict, Field
@@ -83,7 +83,7 @@ class MagenticAgentDeltaEvent:
function_call_arguments: Any | None = None
function_result_id: str | None = None
function_result: Any | None = None
role: ChatRole | None = None
role: Role | None = None
@dataclass
@@ -289,7 +289,7 @@ class MagenticStartMessage:
Returns:
A MagenticStartMessage with the string converted to a ChatMessage.
"""
return cls(task=ChatMessage(role=ChatRole.USER, text=task_text))
return cls(task=ChatMessage(role=Role.USER, text=task_text))
@dataclass
@@ -401,7 +401,7 @@ def _team_block(participants: dict[str, str]) -> str:
def _first_assistant(messages: list[ChatMessage]) -> ChatMessage | None:
for msg in reversed(messages):
if msg.role == ChatRole.ASSISTANT:
if msg.role == Role.ASSISTANT:
return msg
return None
@@ -409,7 +409,7 @@ def _first_assistant(messages: list[ChatMessage]) -> ChatMessage | None:
def _extract_json(text: str) -> dict[str, Any]:
"""Potentially temp helper method.
Note: this method is required right now because the ChatClient, when calling
Note: this method is required right now because the ChatClientProtocol, when calling
response.text, returns duplicate JSON payloads - need to figure out why.
The `text` method is concatenating multiple text contents from diff msgs into a single string.
@@ -497,7 +497,7 @@ class MagenticManagerBase(AFBaseModel, ABC):
class StandardMagenticManager(MagenticManagerBase):
"""Standard Magentic manager that performs real LLM calls via a ChatClientAgent.
"""Standard Magentic manager that performs real LLM calls via a ChatAgent.
The manager constructs prompts that mirror the original Magentic One orchestration:
- Facts gathering
@@ -509,7 +509,7 @@ class StandardMagenticManager(MagenticManagerBase):
model_config = ConfigDict(arbitrary_types_allowed=True)
chat_client: ChatClient
chat_client: ChatClientProtocol
task_ledger: MagenticTaskLedger | None = None
instructions: str | None = None
@@ -526,7 +526,7 @@ class StandardMagenticManager(MagenticManagerBase):
def __init__(
self,
chat_client: ChatClient,
chat_client: ChatClientProtocol,
task_ledger: MagenticTaskLedger | None = None,
*,
instructions: str | None = None,
@@ -597,7 +597,7 @@ class StandardMagenticManager(MagenticManagerBase):
*,
response_format: type[BaseModel] | None = None,
) -> ChatMessage:
"""Call the underlying ChatClient directly and return the last assistant message.
"""Call the underlying ChatClientProtocol directly and return the last assistant message.
If manager instructions are provided, they are injected as a SYSTEM message
at the start of the request to guide the model consistently without needing
@@ -606,7 +606,7 @@ class StandardMagenticManager(MagenticManagerBase):
# Prepend system instructions if present
request_messages: list[ChatMessage] = []
if self.instructions:
request_messages.append(ChatMessage(role=ChatRole.SYSTEM, text=self.instructions))
request_messages.append(ChatMessage(role=Role.SYSTEM, text=self.instructions))
request_messages.extend(messages)
# Invoke the chat client non-streaming API
@@ -619,13 +619,13 @@ class StandardMagenticManager(MagenticManagerBase):
if out_messages:
last = out_messages[-1]
return ChatMessage(
role=last.role or ChatRole.ASSISTANT,
role=last.role or Role.ASSISTANT,
text=last.text or "",
author_name=last.author_name or MAGENTIC_MANAGER_NAME,
)
# Fallback if no messages
return ChatMessage(role=ChatRole.ASSISTANT, text="No output produced.", author_name=MAGENTIC_MANAGER_NAME)
return ChatMessage(role=Role.ASSISTANT, text="No output produced.", author_name=MAGENTIC_MANAGER_NAME)
async def plan(self, magentic_context: MagenticContext) -> ChatMessage:
"""Create facts and plan using the model, then render a combined task ledger as a single assistant message."""
@@ -634,14 +634,14 @@ class StandardMagenticManager(MagenticManagerBase):
# Gather facts
facts_user = ChatMessage(
role=ChatRole.USER,
role=Role.USER,
text=self.task_ledger_facts_prompt.format(task=task_text),
)
facts_msg = await self._complete([*magentic_context.chat_history, facts_user])
# Create plan
plan_user = ChatMessage(
role=ChatRole.USER,
role=Role.USER,
text=self.task_ledger_plan_prompt.format(team=team_text),
)
plan_msg = await self._complete([*magentic_context.chat_history, facts_user, facts_msg, plan_user])
@@ -659,7 +659,7 @@ class StandardMagenticManager(MagenticManagerBase):
facts=facts_msg.text,
plan=plan_msg.text,
)
return ChatMessage(role=ChatRole.ASSISTANT, text=combined, author_name=MAGENTIC_MANAGER_NAME)
return ChatMessage(role=Role.ASSISTANT, text=combined, author_name=MAGENTIC_MANAGER_NAME)
async def replan(self, magentic_context: MagenticContext) -> ChatMessage:
"""Update facts and plan when stalling or looping has been detected."""
@@ -671,14 +671,14 @@ class StandardMagenticManager(MagenticManagerBase):
# Update facts
facts_update_user = ChatMessage(
role=ChatRole.USER,
role=Role.USER,
text=self.task_ledger_facts_update_prompt.format(task=task_text, old_facts=self.task_ledger.facts.text),
)
updated_facts = await self._complete([*magentic_context.chat_history, facts_update_user])
# Update plan
plan_update_user = ChatMessage(
role=ChatRole.USER,
role=Role.USER,
text=self.task_ledger_plan_update_prompt.format(team=team_text),
)
updated_plan = await self._complete([
@@ -701,7 +701,7 @@ class StandardMagenticManager(MagenticManagerBase):
facts=updated_facts.text,
plan=updated_plan.text,
)
return ChatMessage(role=ChatRole.ASSISTANT, text=combined, author_name=MAGENTIC_MANAGER_NAME)
return ChatMessage(role=Role.ASSISTANT, text=combined, author_name=MAGENTIC_MANAGER_NAME)
async def create_progress_ledger(self, magentic_context: MagenticContext) -> MagenticProgressLedger:
"""Use the model to produce a JSON progress ledger based on the conversation so far.
@@ -721,7 +721,7 @@ class StandardMagenticManager(MagenticManagerBase):
team=team_text,
names=names_csv,
)
user_message = ChatMessage(role=ChatRole.USER, text=prompt)
user_message = ChatMessage(role=Role.USER, text=prompt)
# Include full context to help the model decide current stage, with small retry loop
attempts = 0
@@ -751,11 +751,11 @@ class StandardMagenticManager(MagenticManagerBase):
async def prepare_final_answer(self, magentic_context: MagenticContext) -> ChatMessage:
"""Ask the model to produce the final answer addressed to the user."""
prompt = self.final_answer_prompt.format(task=magentic_context.task.text)
user_message = ChatMessage(role=ChatRole.USER, text=prompt)
user_message = ChatMessage(role=Role.USER, text=prompt)
response = await self._complete([*magentic_context.chat_history, user_message])
# Ensure role is assistant
return ChatMessage(
role=ChatRole.ASSISTANT,
role=Role.ASSISTANT,
text=response.text,
author_name=response.author_name or MAGENTIC_MANAGER_NAME,
)
@@ -896,9 +896,9 @@ class MagenticOrchestratorExecutor(Executor):
logger.debug("Magentic Orchestrator: Received response from agent")
# Add transfer message if needed
if message.body.role != ChatRole.USER:
if message.body.role != Role.USER:
transfer_msg = ChatMessage(
role=ChatRole.USER,
role=Role.USER,
text=f"Transferred to {getattr(message.body, 'author_name', 'agent')}",
)
self._context.chat_history.append(transfer_msg)
@@ -945,7 +945,7 @@ class MagenticOrchestratorExecutor(Executor):
plan=human.edited_plan_text,
)
self._task_ledger = ChatMessage(
role=ChatRole.ASSISTANT,
role=Role.ASSISTANT,
text=combined,
author_name=MAGENTIC_MANAGER_NAME,
)
@@ -953,7 +953,7 @@ class MagenticOrchestratorExecutor(Executor):
elif human.comments:
# Record the human feedback for grounding
self._context.chat_history.append(
ChatMessage(role=ChatRole.USER, text=f"Human plan feedback: {human.comments}")
ChatMessage(role=Role.USER, text=f"Human plan feedback: {human.comments}")
)
# Ask the manager to replan based on comments; proceed immediately
self._task_ledger = await self._manager.replan(self._context.model_copy(deep=True))
@@ -981,7 +981,7 @@ class MagenticOrchestratorExecutor(Executor):
self._require_plan_signoff = False
# Add a clear note to the conversation so users know review is closed
notice = ChatMessage(
role=ChatRole.ASSISTANT,
role=Role.ASSISTANT,
text=(
"Plan review closed after max rounds. Proceeding with the current plan and will no longer "
"prompt for plan approval."
@@ -1015,14 +1015,14 @@ class MagenticOrchestratorExecutor(Executor):
facts=(mgr_ledger2.facts.text if mgr_ledger2 else ""),
plan=human.edited_plan_text,
)
self._task_ledger = ChatMessage(role=ChatRole.ASSISTANT, text=combined, author_name=MAGENTIC_MANAGER_NAME)
self._task_ledger = ChatMessage(role=Role.ASSISTANT, text=combined, author_name=MAGENTIC_MANAGER_NAME)
await self._send_plan_review_request(context)
return
# Else pass comments into the chat history and replan with the manager
if human.comments:
self._context.chat_history.append(
ChatMessage(role=ChatRole.USER, text=f"Human plan feedback: {human.comments}")
ChatMessage(role=Role.USER, text=f"Human plan feedback: {human.comments}")
)
# Ask the manager to replan; this only adjusts the plan stage, not a full reset
@@ -1127,7 +1127,7 @@ class MagenticOrchestratorExecutor(Executor):
# Add instruction to conversation (assistant guidance)
instruction_msg = ChatMessage(
role=ChatRole.ASSISTANT,
role=Role.ASSISTANT,
text=str(instruction),
author_name=MAGENTIC_MANAGER_NAME,
)
@@ -1215,7 +1215,7 @@ class MagenticOrchestratorExecutor(Executor):
partial_result = _first_assistant(ctx.chat_history)
if partial_result is None:
partial_result = ChatMessage(
role=ChatRole.ASSISTANT,
role=Role.ASSISTANT,
text=f"Stopped due to {limit_type} limit. No partial result available.",
author_name=MAGENTIC_MANAGER_NAME,
)
@@ -1262,7 +1262,7 @@ class MagenticAgentExecutor(Executor):
def __init__(
self,
agent: AIAgent | Executor,
agent: AgentProtocol | Executor,
agent_id: str,
agent_response_callback: Callable[[str, ChatMessage], Awaitable[None]] | None = None,
streaming_agent_response_callback: Callable[[str, AgentRunResponseUpdate, bool], Awaitable[None]] | None = None,
@@ -1288,9 +1288,9 @@ class MagenticAgentExecutor(Executor):
return
# Add transfer message if needed
if message.body.role != ChatRole.USER:
if message.body.role != Role.USER:
transfer_msg = ChatMessage(
role=ChatRole.USER,
role=Role.USER,
text=f"Transferred to {getattr(message.body, 'author_name', 'agent')}",
)
self._chat_history.append(transfer_msg)
@@ -1298,18 +1298,18 @@ class MagenticAgentExecutor(Executor):
# Add message to agent's history
self._chat_history.append(message.body)
def _get_persona_adoption_role(self) -> ChatRole:
def _get_persona_adoption_role(self) -> Role:
"""Determine the best role for persona adoption messages.
Uses SYSTEM role if the agent supports it, otherwise falls back to USER.
"""
# Only AgentBase-derived agents are assumed to support SYSTEM messages reliably.
from agent_framework import AgentBase as _AF_AgentBase # local import to avoid cycles
# Only BaseAgent-derived agents are assumed to support SYSTEM messages reliably.
from agent_framework import BaseAgent as _AF_AgentBase # local import to avoid cycles
if isinstance(self._agent, _AF_AgentBase) and hasattr(self._agent, "chat_client"):
return ChatRole.SYSTEM
return Role.SYSTEM
# For other agent types or when we can't determine support, use USER
return ChatRole.USER
return Role.USER
@handler
async def handle_request_message(
@@ -1331,14 +1331,14 @@ class MagenticAgentExecutor(Executor):
# Add the orchestrator's instruction as a USER message so the agent treats it as the prompt
if message.instruction:
self._chat_history.append(ChatMessage(role=ChatRole.USER, text=message.instruction))
self._chat_history.append(ChatMessage(role=Role.USER, text=message.instruction))
try:
# If the participant is not an invokable AgentBase, return a no-op response.
from agent_framework import AgentBase as _AF_AgentBase # local import to avoid cycles
# If the participant is not an invokable BaseAgent, return a no-op response.
from agent_framework import BaseAgent as _AF_AgentBase # local import to avoid cycles
if not isinstance(self._agent, _AF_AgentBase):
response = ChatMessage(
role=ChatRole.ASSISTANT,
role=Role.ASSISTANT,
text=f"{self._agent_id} is a workflow executor and cannot be invoked directly.",
author_name=self._agent_id,
)
@@ -1354,7 +1354,7 @@ class MagenticAgentExecutor(Executor):
logger.warning("Agent %s invoke failed: %s", self._agent_id, e)
# Fallback response
response = ChatMessage(
role=ChatRole.ASSISTANT,
role=Role.ASSISTANT,
text=f"Agent {self._agent_id}: Error processing request - {str(e)[:100]}",
)
self._chat_history.append(response)
@@ -1370,9 +1370,9 @@ class MagenticAgentExecutor(Executor):
logger.debug(f"Agent {self._agent_id}: Running with {len(self._chat_history)} messages")
updates: list[AgentRunResponseUpdate] = []
# The wrapped participant is guaranteed to be an AgentBase when this is called.
agent = cast("AIAgent", self._agent)
async for update in agent.run_streaming(messages=self._chat_history): # type: ignore[attr-defined]
# The wrapped participant is guaranteed to be an BaseAgent when this is called.
agent = cast("AgentProtocol", self._agent)
async for update in agent.run_stream(messages=self._chat_history): # type: ignore[attr-defined]
updates.append(update)
if self._streaming_agent_response_callback is not None:
with contextlib.suppress(Exception):
@@ -1394,7 +1394,7 @@ class MagenticAgentExecutor(Executor):
if messages and len(messages) > 0:
last: ChatMessage = messages[-1]
author = last.author_name or self._agent_id
role: ChatRole = last.role if last.role else ChatRole.ASSISTANT
role: Role = last.role if last.role else Role.ASSISTANT
text = last.text or str(last)
msg = ChatMessage(role=role, text=text, author_name=author)
if self._agent_response_callback is not None:
@@ -1403,7 +1403,7 @@ class MagenticAgentExecutor(Executor):
return msg
msg = ChatMessage(
role=ChatRole.ASSISTANT,
role=Role.ASSISTANT,
text=f"Agent {self._agent_id}: No output produced",
author_name=self._agent_id,
)
@@ -1422,7 +1422,7 @@ class MagenticBuilder:
"""High-level builder for creating Magentic One workflows."""
def __init__(self) -> None:
self._participants: dict[str, AIAgent | Executor] = {}
self._participants: dict[str, AgentProtocol | Executor] = {}
self._manager: MagenticManagerBase | None = None
self._exception_callback: Callable[[Exception], None] | None = None
self._result_callback: Callable[[ChatMessage], Awaitable[None]] | None = None
@@ -1435,7 +1435,7 @@ class MagenticBuilder:
self._unified_callback: CallbackSink | None = None
self._callback_mode: MagenticCallbackMode | None = None
def participants(self, **participants: AIAgent | Executor) -> Self:
def participants(self, **participants: AgentProtocol | Executor) -> Self:
"""Add participants (agents) to the workflow."""
self._participants.update(participants)
return self
@@ -1450,7 +1450,7 @@ class MagenticBuilder:
manager: MagenticManagerBase | None = None,
*,
# Constructor args for StandardMagenticManager when manager is not provided
chat_client: ChatClient | None = None,
chat_client: ChatClientProtocol | None = None,
task_ledger: MagenticTaskLedger | None = None,
instructions: str | None = None,
# Prompt overrides
@@ -1540,7 +1540,7 @@ class MagenticBuilder:
# Create participant descriptions
participant_descriptions: dict[str, str] = {}
for name, participant in self._participants.items():
if isinstance(participant, AgentBase):
if isinstance(participant, BaseAgent):
description = getattr(participant, "description", None) or f"Agent {name}"
else:
description = f"Executor {name}"
@@ -1745,7 +1745,7 @@ class MagenticWorkflow:
WorkflowEvent: The events generated during the workflow execution.
"""
start_message = MagenticStartMessage.from_string(task_text)
async for event in self._workflow.run_streaming(start_message):
async for event in self._workflow.run_stream(start_message):
yield event
async def run_streaming_with_message(self, task_message: ChatMessage) -> AsyncIterable[WorkflowEvent]:
@@ -1758,10 +1758,10 @@ class MagenticWorkflow:
WorkflowEvent: The events generated during the workflow execution.
"""
start_message = MagenticStartMessage(task=task_message)
async for event in self._workflow.run_streaming(start_message):
async for event in self._workflow.run_stream(start_message):
yield event
async def run_streaming(self, message: Any | None = None) -> AsyncIterable[WorkflowEvent]:
async def run_stream(self, message: Any | None = None) -> AsyncIterable[WorkflowEvent]:
"""Run the workflow with either a message object or the preset task string.
Args:
@@ -1780,7 +1780,7 @@ class MagenticWorkflow:
elif isinstance(message, ChatMessage):
message = MagenticStartMessage(task=message)
async for event in self._workflow.run_streaming(message):
async for event in self._workflow.run_stream(message):
yield event
async def run_with_string(self, task_text: str) -> WorkflowRunResult:
@@ -1822,7 +1822,7 @@ class MagenticWorkflow:
WorkflowRunResult: All events generated during the workflow execution.
"""
events: list[WorkflowEvent] = []
async for event in self.run_streaming(message):
async for event in self.run_stream(message):
events.append(event)
return WorkflowRunResult(events)
python/packages/workflow/agent_framework_workflow/_workflow.py
+4 -4
View File
@@ -225,7 +225,7 @@ class Workflow(AFBaseModel):
workflow_tracer.add_workflow_error_event(e)
raise
async def run_streaming(self, message: Any) -> AsyncIterable[WorkflowEvent]:
async def run_stream(self, message: Any) -> AsyncIterable[WorkflowEvent]:
"""Run the workflow with a starting message and stream events.
Args:
@@ -252,7 +252,7 @@ class Workflow(AFBaseModel):
async for event in self._run_workflow_with_tracing(initial_executor_fn=initial_execution, reset_context=True):
yield event
async def run_streaming_from_checkpoint(
async def run_stream_from_checkpoint(
self,
checkpoint_id: str,
checkpoint_storage: CheckpointStorage | None = None,
@@ -368,7 +368,7 @@ class Workflow(AFBaseModel):
Returns:
A WorkflowRunResult instance containing a list of events generated during the workflow execution.
"""
events = [event async for event in self.run_streaming(message)]
events = [event async for event in self.run_stream(message)]
return WorkflowRunResult(events)
async def run_from_checkpoint(
@@ -394,7 +394,7 @@ class Workflow(AFBaseModel):
RuntimeError: If checkpoint restoration fails.
"""
events = [
event async for event in self.run_streaming_from_checkpoint(checkpoint_id, checkpoint_storage, responses)
event async for event in self.run_stream_from_checkpoint(checkpoint_id, checkpoint_storage, responses)
]
return WorkflowRunResult(events)
python/packages/workflow/tests/test_magentic.py
+41 -43
View File
@@ -11,11 +11,11 @@ from agent_framework import (
ChatMessage,
ChatResponse,
ChatResponseUpdate,
ChatRole,
Role,
TextContent,
)
from agent_framework._agents import AgentBase
from agent_framework._clients import ChatClient as AFChatClient
from agent_framework._agents import BaseAgent
from agent_framework._clients import ChatClientProtocol as AFChatClient
from agent_framework_workflow import (
Executor,
@@ -42,7 +42,7 @@ def test_magentic_start_message_from_string():
msg = MagenticStartMessage.from_string("Do the thing")
assert isinstance(msg, MagenticStartMessage)
assert isinstance(msg.task, ChatMessage)
assert msg.task.role == ChatRole.USER
assert msg.task.role == Role.USER
assert msg.task.text == "Do the thing"
@@ -67,11 +67,11 @@ def test_plan_review_request_defaults_and_reply_variants():
def test_magentic_context_reset_behavior():
ctx = MagenticContext(
task=ChatMessage(role=ChatRole.USER, text="task"),
task=ChatMessage(role=Role.USER, text="task"),
participant_descriptions={"Alice": "Researcher"},
)
# seed context state
ctx.chat_history.append(ChatMessage(role=ChatRole.ASSISTANT, text="draft"))
ctx.chat_history.append(ChatMessage(role=Role.ASSISTANT, text="draft"))
ctx.stall_count = 2
prev_reset = ctx.reset_count
@@ -97,18 +97,18 @@ class FakeManager(MagenticManagerBase):
instruction_text: str = "Proceed with step 1"
async def plan(self, magentic_context: MagenticContext) -> ChatMessage:
facts = ChatMessage(role=ChatRole.ASSISTANT, text="GIVEN OR VERIFIED FACTS\n- A\n")
plan = ChatMessage(role=ChatRole.ASSISTANT, text="- Do X\n- Do Y\n")
facts = ChatMessage(role=Role.ASSISTANT, text="GIVEN OR VERIFIED FACTS\n- A\n")
plan = ChatMessage(role=Role.ASSISTANT, text="- Do X\n- Do Y\n")
self.task_ledger = _SimpleLedger(facts=facts, plan=plan)
combined = f"Task: {magentic_context.task.text}\n\nFacts:\n{facts.text}\n\nPlan:\n{plan.text}"
return ChatMessage(role=ChatRole.ASSISTANT, text=combined, author_name="magentic_manager")
return ChatMessage(role=Role.ASSISTANT, text=combined, author_name="magentic_manager")
async def replan(self, magentic_context: MagenticContext) -> ChatMessage:
facts = ChatMessage(role=ChatRole.ASSISTANT, text="GIVEN OR VERIFIED FACTS\n- A2\n")
plan = ChatMessage(role=ChatRole.ASSISTANT, text="- Do Z\n")
facts = ChatMessage(role=Role.ASSISTANT, text="GIVEN OR VERIFIED FACTS\n- A2\n")
plan = ChatMessage(role=Role.ASSISTANT, text="- Do Z\n")
self.task_ledger = _SimpleLedger(facts=facts, plan=plan)
combined = f"Task: {magentic_context.task.text}\n\nFacts:\n{facts.text}\n\nPlan:\n{plan.text}"
return ChatMessage(role=ChatRole.ASSISTANT, text=combined, author_name="magentic_manager")
return ChatMessage(role=Role.ASSISTANT, text=combined, author_name="magentic_manager")
async def create_progress_ledger(self, magentic_context: MagenticContext) -> MagenticProgressLedger:
is_satisfied = self.satisfied_after_signoff and len(magentic_context.chat_history) > 0
@@ -121,18 +121,18 @@ class FakeManager(MagenticManagerBase):
)
async def prepare_final_answer(self, magentic_context: MagenticContext) -> ChatMessage:
return ChatMessage(role=ChatRole.ASSISTANT, text="FINAL", author_name="magentic_manager")
return ChatMessage(role=Role.ASSISTANT, text="FINAL", author_name="magentic_manager")
async def test_standard_manager_plan_and_replan_combined_ledger():
manager = FakeManager(max_round_count=10, max_stall_count=3, max_reset_count=2)
ctx = MagenticContext(
task=ChatMessage(role=ChatRole.USER, text="demo task"),
task=ChatMessage(role=Role.USER, text="demo task"),
participant_descriptions={"agentA": "Agent A"},
)
first = await manager.plan(ctx.model_copy(deep=True))
assert first.role == ChatRole.ASSISTANT and "Facts:" in first.text and "Plan:" in first.text
assert first.role == Role.ASSISTANT and "Facts:" in first.text and "Plan:" in first.text
assert manager.task_ledger is not None
replanned = await manager.replan(ctx.model_copy(deep=True))
@@ -142,7 +142,7 @@ async def test_standard_manager_plan_and_replan_combined_ledger():
async def test_standard_manager_progress_ledger_and_fallback():
manager = FakeManager(max_round_count=10)
ctx = MagenticContext(
task=ChatMessage(role=ChatRole.USER, text="demo"),
task=ChatMessage(role=Role.USER, text="demo"),
participant_descriptions={"agentA": "Agent A"},
)
@@ -166,7 +166,7 @@ async def test_magentic_workflow_plan_review_approval_to_completion():
)
req_event: RequestInfoEvent | None = None
async for ev in wf.run_streaming("do work"):
async for ev in wf.run_stream("do work"):
if isinstance(ev, RequestInfoEvent) and ev.request_type is MagenticPlanReviewRequest:
req_event = ev
assert req_event is not None
@@ -205,7 +205,7 @@ async def test_magentic_plan_review_approve_with_comments_replans_and_proceeds()
# Wait for the initial plan review request
req_event: RequestInfoEvent | None = None
async for ev in wf.run_streaming("do work"):
async for ev in wf.run_stream("do work"):
if isinstance(ev, RequestInfoEvent) and ev.request_type is MagenticPlanReviewRequest:
req_event = ev
assert req_event is not None
@@ -242,7 +242,7 @@ async def test_magentic_orchestrator_round_limit_produces_partial_result():
from agent_framework_workflow import WorkflowEvent # type: ignore
events: list[WorkflowEvent] = []
async for ev in wf.run_streaming("round limit test"):
async for ev in wf.run_stream("round limit test"):
events.append(ev)
if len(events) > 50:
break
@@ -251,7 +251,7 @@ async def test_magentic_orchestrator_round_limit_produces_partial_result():
assert completed is not None
data = getattr(completed, "data", None)
assert isinstance(data, ChatMessage)
assert data.role == ChatRole.ASSISTANT
assert data.role == Role.ASSISTANT
class _DummyExec(Executor):
@@ -268,7 +268,7 @@ from agent_framework_workflow import StandardMagenticManager # noqa: E402
class _StubChatClient(AFChatClient):
async def get_response(self, messages, **kwargs): # type: ignore[override]
return ChatResponse(messages=[ChatMessage(role=ChatRole.ASSISTANT, text="ok")])
return ChatResponse(messages=[ChatMessage(role=Role.ASSISTANT, text="ok")])
def get_streaming_response(self, messages, **kwargs) -> AsyncIterable[ChatResponseUpdate]: # type: ignore[override]
async def _gen():
@@ -284,14 +284,14 @@ async def test_standard_manager_plan_and_replan_via_complete_monkeypatch():
async def fake_complete_plan(messages: list[ChatMessage], **kwargs: Any) -> ChatMessage:
# Return a different response depending on call order length
if any("FACTS" in (m.text or "") for m in messages):
return ChatMessage(role=ChatRole.ASSISTANT, text="- step A\n- step B")
return ChatMessage(role=ChatRole.ASSISTANT, text="GIVEN OR VERIFIED FACTS\n- fact1")
return ChatMessage(role=Role.ASSISTANT, text="- step A\n- step B")
return ChatMessage(role=Role.ASSISTANT, text="GIVEN OR VERIFIED FACTS\n- fact1")
# First, patch to produce facts then plan
mgr._complete = fake_complete_plan # type: ignore[attr-defined]
ctx = MagenticContext(
task=ChatMessage(role=ChatRole.USER, text="T"),
task=ChatMessage(role=Role.USER, text="T"),
participant_descriptions={"A": "desc"},
)
combined = await mgr.plan(ctx.model_copy(deep=True))
@@ -303,8 +303,8 @@ async def test_standard_manager_plan_and_replan_via_complete_monkeypatch():
# Now replan with new outputs
async def fake_complete_replan(messages: list[ChatMessage], **kwargs: Any) -> ChatMessage:
if any("Please briefly explain" in (m.text or "") for m in messages):
return ChatMessage(role=ChatRole.ASSISTANT, text="- new step")
return ChatMessage(role=ChatRole.ASSISTANT, text="GIVEN OR VERIFIED FACTS\n- updated")
return ChatMessage(role=Role.ASSISTANT, text="- new step")
return ChatMessage(role=Role.ASSISTANT, text="GIVEN OR VERIFIED FACTS\n- updated")
mgr._complete = fake_complete_replan # type: ignore[attr-defined]
combined2 = await mgr.replan(ctx.model_copy(deep=True))
@@ -314,7 +314,7 @@ async def test_standard_manager_plan_and_replan_via_complete_monkeypatch():
async def test_standard_manager_progress_ledger_success_and_error():
mgr = StandardMagenticManager(chat_client=_StubChatClient())
ctx = MagenticContext(
task=ChatMessage(role=ChatRole.USER, text="task"),
task=ChatMessage(role=Role.USER, text="task"),
participant_descriptions={"alice": "desc"},
)
@@ -327,7 +327,7 @@ async def test_standard_manager_progress_ledger_success_and_error():
'"next_speaker": {"reason": "r", "answer": "alice"}, '
'"instruction_or_question": {"reason": "r", "answer": "do"}}'
)
return ChatMessage(role=ChatRole.ASSISTANT, text=json_text)
return ChatMessage(role=Role.ASSISTANT, text=json_text)
mgr._complete = fake_complete_ok # type: ignore[attr-defined]
ledger = await mgr.create_progress_ledger(ctx.model_copy(deep=True))
@@ -335,7 +335,7 @@ async def test_standard_manager_progress_ledger_success_and_error():
# Error path: invalid JSON now raises to avoid emitting planner-oriented instructions to agents
async def fake_complete_bad(messages: list[ChatMessage], **kwargs: Any) -> ChatMessage:
return ChatMessage(role=ChatRole.ASSISTANT, text="not-json")
return ChatMessage(role=Role.ASSISTANT, text="not-json")
mgr._complete = fake_complete_bad # type: ignore[attr-defined]
with pytest.raises(RuntimeError):
@@ -348,10 +348,10 @@ class InvokeOnceManager(MagenticManagerBase):
self._invoked = False
async def plan(self, magentic_context: MagenticContext) -> ChatMessage:
return ChatMessage(role=ChatRole.ASSISTANT, text="ledger")
return ChatMessage(role=Role.ASSISTANT, text="ledger")
async def replan(self, magentic_context: MagenticContext) -> ChatMessage:
return ChatMessage(role=ChatRole.ASSISTANT, text="re-ledger")
return ChatMessage(role=Role.ASSISTANT, text="re-ledger")
async def create_progress_ledger(self, magentic_context: MagenticContext) -> MagenticProgressLedger:
if not self._invoked:
@@ -374,43 +374,41 @@ class InvokeOnceManager(MagenticManagerBase):
)
async def prepare_final_answer(self, magentic_context: MagenticContext) -> ChatMessage:
return ChatMessage(role=ChatRole.ASSISTANT, text="final")
return ChatMessage(role=Role.ASSISTANT, text="final")
class StubThreadAgent(AgentBase):
async def run_streaming(self, messages=None, *, thread=None, **kwargs): # type: ignore[override]
class StubThreadAgent(BaseAgent):
async def run_stream(self, messages=None, *, thread=None, **kwargs): # type: ignore[override]
yield AgentRunResponseUpdate(
contents=[TextContent(text="thread-ok")],
author_name="agentA",
role=ChatRole.ASSISTANT,
role=Role.ASSISTANT,
)
async def run(self, messages=None, *, thread=None, **kwargs): # type: ignore[override]
return AgentRunResponse(messages=[ChatMessage(role=ChatRole.ASSISTANT, text="thread-ok", author_name="agentA")])
return AgentRunResponse(messages=[ChatMessage(role=Role.ASSISTANT, text="thread-ok", author_name="agentA")])
class StubAssistantsClient:
pass # class name used for branch detection
class StubAssistantsAgent(AgentBase):
class StubAssistantsAgent(BaseAgent):
chat_client: object | None = None # allow assignment via Pydantic field
def __init__(self) -> None:
super().__init__()
self.chat_client = StubAssistantsClient() # type name contains 'AssistantsClient'
async def run_streaming(self, messages=None, *, thread=None, **kwargs): # type: ignore[override]
async def run_stream(self, messages=None, *, thread=None, **kwargs): # type: ignore[override]
yield AgentRunResponseUpdate(
contents=[TextContent(text="assistants-ok")],
author_name="agentA",
role=ChatRole.ASSISTANT,
role=Role.ASSISTANT,
)
async def run(self, messages=None, *, thread=None, **kwargs): # type: ignore[override]
return AgentRunResponse(
messages=[ChatMessage(role=ChatRole.ASSISTANT, text="assistants-ok", author_name="agentA")]
)
return AgentRunResponse(messages=[ChatMessage(role=Role.ASSISTANT, text="assistants-ok", author_name="agentA")])
async def _collect_agent_responses_setup(participant_obj: object):
@@ -432,7 +430,7 @@ async def _collect_agent_responses_setup(participant_obj: object):
# Run a bounded stream to allow one invoke and then completion
events: list[WorkflowEvent] = []
async for ev in wf.run_streaming("task"): # plan review disabled
async for ev in wf.run_stream("task"): # plan review disabled
events.append(ev)
if len(events) > 50:
break
python/packages/workflow/tests/test_tracing.py
+2 -2
View File
@@ -343,7 +343,7 @@ async def test_end_to_end_workflow_tracing(tracing_enabled: Any, span_exporter:
# Run workflow (this should create run spans)
events = []
async for event in workflow.run_streaming("test input"):
async for event in workflow.run_stream("test input"):
events.append(event)
# Verify workflow executed correctly
@@ -444,7 +444,7 @@ async def test_workflow_error_handling_in_tracing(tracing_enabled: Any, span_exp
# Run workflow and expect error
with pytest.raises(ValueError, match="Test error"):
async for _ in workflow.run_streaming("test input"):
async for _ in workflow.run_stream("test input"):
pass
spans = span_exporter.get_finished_spans()
python/packages/workflow/tests/test_workflow.py
+9 -9
View File
@@ -95,7 +95,7 @@ async def test_workflow_run_streaming():
)
result: int | None = None
async for event in workflow.run_streaming(NumberMessage(data=0)):
async for event in workflow.run_stream(NumberMessage(data=0)):
assert isinstance(event, WorkflowEvent)
if isinstance(event, WorkflowCompletedEvent):
result = event.data
@@ -118,7 +118,7 @@ async def test_workflow_run_stream_not_completed():
)
with pytest.raises(RuntimeError):
async for _ in workflow.run_streaming(NumberMessage(data=0)):
async for _ in workflow.run_stream(NumberMessage(data=0)):
pass
@@ -176,7 +176,7 @@ async def test_workflow_send_responses_streaming():
)
request_info_event: RequestInfoEvent | None = None
async for event in workflow.run_streaming(NumberMessage(data=0)):
async for event in workflow.run_stream(NumberMessage(data=0)):
if isinstance(event, RequestInfoEvent):
request_info_event = event
@@ -326,7 +326,7 @@ async def test_workflow_checkpointing_not_enabled_for_external_restore(simple_ex
# Attempt to restore from checkpoint without providing external storage should fail
try:
[event async for event in workflow.run_streaming_from_checkpoint("fake-checkpoint-id")]
[event async for event in workflow.run_stream_from_checkpoint("fake-checkpoint-id")]
raise AssertionError("Expected ValueError to be raised")
except ValueError as e:
assert "Cannot restore from checkpoint" in str(e)
@@ -344,7 +344,7 @@ async def test_workflow_run_stream_from_checkpoint_no_checkpointing_enabled(simp
# Attempt to run from checkpoint should fail
try:
async for _ in workflow.run_streaming_from_checkpoint("fake_checkpoint_id"):
async for _ in workflow.run_stream_from_checkpoint("fake_checkpoint_id"):
pass
raise AssertionError("Expected ValueError to be raised")
except ValueError as e:
@@ -368,7 +368,7 @@ async def test_workflow_run_stream_from_checkpoint_invalid_checkpoint(simple_exe
# Attempt to run from non-existent checkpoint should fail
try:
async for _ in workflow.run_streaming_from_checkpoint("nonexistent_checkpoint_id"):
async for _ in workflow.run_stream_from_checkpoint("nonexistent_checkpoint_id"):
pass
raise AssertionError("Expected RuntimeError to be raised")
except RuntimeError as e:
@@ -401,7 +401,7 @@ async def test_workflow_run_stream_from_checkpoint_with_external_storage(simple_
# Resume from checkpoint using external storage parameter
try:
events: list[WorkflowEvent] = []
async for event in workflow_without_checkpointing.run_streaming_from_checkpoint(
async for event in workflow_without_checkpointing.run_stream_from_checkpoint(
checkpoint_id, checkpoint_storage=storage
):
events.append(event)
@@ -446,7 +446,7 @@ async def test_workflow_run_from_checkpoint_non_streaming(simple_executor: Execu
async def test_workflow_run_stream_from_checkpoint_with_responses(simple_executor: Executor):
"""Test that run_streaming_from_checkpoint accepts responses parameter."""
"""Test that run_stream_from_checkpoint accepts responses parameter."""
with tempfile.TemporaryDirectory() as temp_dir:
storage = FileCheckpointStorage(temp_dir)
@@ -477,7 +477,7 @@ async def test_workflow_run_stream_from_checkpoint_with_responses(simple_executo
try:
events: list[WorkflowEvent] = []
async for event in workflow.run_streaming_from_checkpoint(checkpoint_id, responses=responses):
async for event in workflow.run_stream_from_checkpoint(checkpoint_id, responses=responses):
events.append(event)
if len(events) >= 2: # Limit to avoid infinite loops
break
python/packages/workflow/tests/test_workflow_agent.py
+15 -15
View File
@@ -8,8 +8,8 @@ from agent_framework import (
AgentRunResponse,
AgentRunResponseUpdate,
ChatMessage,
ChatRole,
FunctionResultContent,
Role,
TextContent,
UsageContent,
UsageDetails,
@@ -43,11 +43,11 @@ class SimpleExecutor(Executor):
response_text = f"{self.response_text}: {input_text}"
# Create response message for both streaming and non-streaming cases
response_message = ChatMessage(role=ChatRole.ASSISTANT, contents=[TextContent(text=response_text)])
response_message = ChatMessage(role=Role.ASSISTANT, contents=[TextContent(text=response_text)])
# Emit update event.
streaming_update = AgentRunResponseUpdate(
contents=[TextContent(text=response_text)], role=ChatRole.ASSISTANT, message_id=str(uuid.uuid4())
contents=[TextContent(text=response_text)], role=Role.ASSISTANT, message_id=str(uuid.uuid4())
)
await ctx.add_event(AgentRunUpdateEvent(executor_id=self.id, data=streaming_update))
@@ -68,7 +68,7 @@ class RequestingExecutor(Executor):
# Handle the response and emit completion response
update = AgentRunResponseUpdate(
contents=[TextContent(text="Request completed successfully")],
role=ChatRole.ASSISTANT,
role=Role.ASSISTANT,
message_id=str(uuid.uuid4()),
)
await ctx.add_event(AgentRunUpdateEvent(executor_id=self.id, data=update))
@@ -132,7 +132,7 @@ class TestWorkflowAgent:
# Execute workflow streaming to capture streaming events
updates = []
async for update in agent.run_streaming("Test input"):
async for update in agent.run_stream("Test input"):
updates.append(update)
# Should have received at least one streaming update
@@ -165,7 +165,7 @@ class TestWorkflowAgent:
# Execute workflow streaming to get request info event
updates = []
async for update in agent.run_streaming("Start request"):
async for update in agent.run_stream("Start request"):
updates.append(update)
# Should have received a function call for the request info
assert len(updates) > 0
@@ -192,7 +192,7 @@ class TestWorkflowAgent:
# Now provide a function result response to test continuation
response_message = ChatMessage(
role=ChatRole.USER,
role=Role.USER,
contents=[FunctionResultContent(call_id=function_call.call_id, result="User provided answer")],
)
@@ -252,7 +252,7 @@ class TestWorkflowAgentMergeUpdates:
# Response B, Message 2 (latest in resp B)
AgentRunResponseUpdate(
contents=[TextContent(text="RespB-Msg2")],
role=ChatRole.ASSISTANT,
role=Role.ASSISTANT,
response_id="resp-b",
message_id="msg-2",
created_at="2024-01-01T12:02:00Z",
@@ -260,7 +260,7 @@ class TestWorkflowAgentMergeUpdates:
# Response A, Message 1 (earliest overall)
AgentRunResponseUpdate(
contents=[TextContent(text="RespA-Msg1")],
role=ChatRole.ASSISTANT,
role=Role.ASSISTANT,
response_id="resp-a",
message_id="msg-1",
created_at="2024-01-01T12:00:00Z",
@@ -268,7 +268,7 @@ class TestWorkflowAgentMergeUpdates:
# Response B, Message 1 (earlier in resp B)
AgentRunResponseUpdate(
contents=[TextContent(text="RespB-Msg1")],
role=ChatRole.ASSISTANT,
role=Role.ASSISTANT,
response_id="resp-b",
message_id="msg-1",
created_at="2024-01-01T12:01:00Z",
@@ -276,7 +276,7 @@ class TestWorkflowAgentMergeUpdates:
# Response A, Message 2 (later in resp A)
AgentRunResponseUpdate(
contents=[TextContent(text="RespA-Msg2")],
role=ChatRole.ASSISTANT,
role=Role.ASSISTANT,
response_id="resp-a",
message_id="msg-2",
created_at="2024-01-01T12:00:30Z",
@@ -284,7 +284,7 @@ class TestWorkflowAgentMergeUpdates:
# Global dangling update (no response_id) - should go at end
AgentRunResponseUpdate(
contents=[TextContent(text="Global-Dangling")],
role=ChatRole.ASSISTANT,
role=Role.ASSISTANT,
response_id=None,
message_id="msg-global",
created_at="2024-01-01T11:59:00Z", # Earliest timestamp but should be last
@@ -360,7 +360,7 @@ class TestWorkflowAgentMergeUpdates:
details=UsageDetails(input_token_count=10, output_token_count=5, total_token_count=15)
),
],
role=ChatRole.ASSISTANT,
role=Role.ASSISTANT,
response_id="resp-1",
message_id="msg-1",
created_at="2024-01-01T12:00:00Z",
@@ -373,7 +373,7 @@ class TestWorkflowAgentMergeUpdates:
details=UsageDetails(input_token_count=20, output_token_count=8, total_token_count=28)
),
],
role=ChatRole.ASSISTANT,
role=Role.ASSISTANT,
response_id="resp-2",
message_id="msg-2",
created_at="2024-01-01T12:01:00Z", # Later timestamp
@@ -384,7 +384,7 @@ class TestWorkflowAgentMergeUpdates:
TextContent(text="Third"),
UsageContent(details=UsageDetails(input_token_count=5, output_token_count=3, total_token_count=8)),
],
role=ChatRole.ASSISTANT,
role=Role.ASSISTANT,
response_id="resp-1", # Same response_id as first
message_id="msg-3",
created_at="2024-01-01T11:59:00Z", # Earlier timestamp