diff --git a/python/packages/core/agent_framework/_workflows/_agent_executor.py b/python/packages/core/agent_framework/_workflows/_agent_executor.py index 4d40eb6168..b92c845a4d 100644 --- a/python/packages/core/agent_framework/_workflows/_agent_executor.py +++ b/python/packages/core/agent_framework/_workflows/_agent_executor.py @@ -104,9 +104,6 @@ class AgentExecutor(Executor): self._cache, thread=self._agent_thread, ): - if not update.text: - # Skip empty updates (no textual or structural content) - continue updates.append(update) await ctx.add_event(AgentRunUpdateEvent(self.id, update)) diff --git a/python/packages/core/agent_framework/_workflows/_runner_context.py b/python/packages/core/agent_framework/_workflows/_runner_context.py index 5dc135d674..4de70cb591 100644 --- a/python/packages/core/agent_framework/_workflows/_runner_context.py +++ b/python/packages/core/agent_framework/_workflows/_runner_context.py @@ -12,7 +12,7 @@ from typing import Any, Protocol, TypedDict, TypeVar, cast, runtime_checkable from ._checkpoint import CheckpointStorage, WorkflowCheckpoint from ._const import DEFAULT_MAX_ITERATIONS -from ._events import AgentRunUpdateEvent, WorkflowEvent +from ._events import WorkflowEvent from ._shared_state import SharedState logger = logging.getLogger(__name__) @@ -487,28 +487,6 @@ class InProcRunnerContext: Events are enqueued so runners can stream them in real time instead of waiting for superstep boundaries. """ - # Filter out empty AgentRunUpdateEvent updates to avoid emitting None/empty chunks - try: - if isinstance(event, AgentRunUpdateEvent): - update = getattr(event, "data", None) - # Skip if no update payload - if not update: - return - # Robust emptiness check: allow either top-level text or any text-bearing content - text_val = getattr(update, "text", None) - contents = getattr(update, "contents", None) - has_text_content = False - if contents: - for c in contents: - if getattr(c, "text", None): - has_text_content = True - break - if not (text_val or has_text_content): - return - except Exception as exc: # pragma: no cover - defensive logging path - # Best-effort filtering only; never block event delivery on filtering errors - logger.debug(f"Error while filtering event {event!r}: {exc}", exc_info=True) - await self._event_queue.put(event) async def drain_events(self) -> list[WorkflowEvent]: diff --git a/python/packages/core/tests/workflow/test_agent_executor_tool_calls.py b/python/packages/core/tests/workflow/test_agent_executor_tool_calls.py new file mode 100644 index 0000000000..8124f6253d --- /dev/null +++ b/python/packages/core/tests/workflow/test_agent_executor_tool_calls.py @@ -0,0 +1,122 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Tests for AgentExecutor handling of tool calls and results in streaming mode.""" + +from collections.abc import AsyncIterable +from typing import Any + +from agent_framework import ( + AgentExecutor, + AgentRunResponse, + AgentRunResponseUpdate, + AgentRunUpdateEvent, + AgentThread, + BaseAgent, + ChatMessage, + FunctionCallContent, + FunctionResultContent, + Role, + TextContent, + WorkflowBuilder, +) + + +class _ToolCallingAgent(BaseAgent): + """Mock agent that simulates tool calls and results in streaming mode.""" + + def __init__(self, **kwargs: Any) -> None: + super().__init__(**kwargs) + + async def run( + self, + messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None, + *, + thread: AgentThread | None = None, + **kwargs: Any, + ) -> AgentRunResponse: + """Non-streaming run - not used in this test.""" + return AgentRunResponse(messages=[ChatMessage(role=Role.ASSISTANT, text="done")]) + + async def run_stream( + self, + messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None, + *, + thread: AgentThread | None = None, + **kwargs: Any, + ) -> AsyncIterable[AgentRunResponseUpdate]: + """Simulate streaming with tool calls and results.""" + # First update: some text + yield AgentRunResponseUpdate( + contents=[TextContent(text="Let me search for that...")], + role=Role.ASSISTANT, + ) + + # Second update: tool call (no text!) + yield AgentRunResponseUpdate( + contents=[ + FunctionCallContent( + call_id="call_123", + name="search", + arguments={"query": "weather"}, + ) + ], + role=Role.ASSISTANT, + ) + + # Third update: tool result (no text!) + yield AgentRunResponseUpdate( + contents=[ + FunctionResultContent( + call_id="call_123", + result={"temperature": 72, "condition": "sunny"}, + ) + ], + role=Role.TOOL, + ) + + # Fourth update: final text response + yield AgentRunResponseUpdate( + contents=[TextContent(text="The weather is sunny, 72°F.")], + role=Role.ASSISTANT, + ) + + +async def test_agent_executor_emits_tool_calls_in_streaming_mode() -> None: + """Test that AgentExecutor emits updates containing FunctionCallContent and FunctionResultContent.""" + # Arrange + agent = _ToolCallingAgent(id="tool_agent", name="ToolAgent") + agent_exec = AgentExecutor(agent, id="tool_exec") + + workflow = WorkflowBuilder().set_start_executor(agent_exec).build() + + # Act: run in streaming mode + events: list[AgentRunUpdateEvent] = [] + async for event in workflow.run_stream("What's the weather?"): + if isinstance(event, AgentRunUpdateEvent): + events.append(event) + + # Assert: we should receive 4 events (text, function call, function result, text) + assert len(events) == 4, f"Expected 4 events, got {len(events)}" + + # First event: text update + assert events[0].data is not None + assert isinstance(events[0].data.contents[0], TextContent) + assert "Let me search" in events[0].data.contents[0].text + + # Second event: function call + assert events[1].data is not None + assert isinstance(events[1].data.contents[0], FunctionCallContent) + func_call = events[1].data.contents[0] + assert func_call.call_id == "call_123" + assert func_call.name == "search" + + # Third event: function result + assert events[2].data is not None + assert isinstance(events[2].data.contents[0], FunctionResultContent) + func_result = events[2].data.contents[0] + assert func_result.call_id == "call_123" + + # Fourth event: final text + assert events[3].data is not None + assert isinstance(events[3].data.contents[0], TextContent) + assert "sunny" in events[3].data.contents[0].text diff --git a/python/samples/getting_started/workflows/README.md b/python/samples/getting_started/workflows/README.md index 66ddfd1a89..17780a7aac 100644 --- a/python/samples/getting_started/workflows/README.md +++ b/python/samples/getting_started/workflows/README.md @@ -34,10 +34,11 @@ Once comfortable with these, explore the rest of the samples below. | Sample | File | Concepts | |---|---|---| -| Azure Chat Agents (Streaming) | [agents/azure_chat_agents_streaming.py](./agents/azure_chat_agents_streaming.py) | Add Azure agents as edges and handle streaming events | -| Custom Agent Executors | [agents/custom_agent_executors.py](./agents/custom_agent_executors.py) | Create executors to handle agent run methods | -| Azure AI Chat Agents (Streaming) | [agents/azure_ai_agents_streaming.py](./agents/azure_ai_agents_streaming.py) | Add Azure AI agents as edges and handle streaming events | | Azure Chat Agents (Streaming) | [agents/azure_chat_agents_streaming.py](./agents/azure_chat_agents_streaming.py) | Add Azure Chat agents as edges and handle streaming events | +| Azure AI Chat Agents (Streaming) | [agents/azure_ai_agents_streaming.py](./agents/azure_ai_agents_streaming.py) | Add Azure AI agents as edges and handle streaming events | +| Azure Chat Agents (Function Bridge) | [agents/azure_chat_agents_function_bridge.py](./agents/azure_chat_agents_function_bridge.py) | Chain two agents with a function executor that injects external context | +| Azure Chat Agents (Tools + HITL) | [agents/azure_chat_agents_tool_calls_with_feedback.py](./agents/azure_chat_agents_tool_calls_with_feedback.py) | Tool-enabled writer/editor pipeline with human feedback gating via RequestInfoExecutor | +| Custom Agent Executors | [agents/custom_agent_executors.py](./agents/custom_agent_executors.py) | Create executors to handle agent run methods | | Workflow as Agent (Reflection Pattern) | [agents/workflow_as_agent_reflection_pattern.py](./agents/workflow_as_agent_reflection_pattern.py) | Wrap a workflow so it can behave like an agent (reflection pattern) | | Workflow as Agent + HITL | [agents/workflow_as_agent_human_in_the_loop.py](./agents/workflow_as_agent_human_in_the_loop.py) | Extend workflow-as-agent with human-in-the-loop capability | @@ -73,6 +74,7 @@ Once comfortable with these, explore the rest of the samples below. | Sample | File | Concepts | |---|---|---| | Human-In-The-Loop (Guessing Game) | [human-in-the-loop/guessing_game_with_human_input.py](./human-in-the-loop/guessing_game_with_human_input.py) | Interactive request/response prompts with a human | +| Azure Agents Tool Feedback Loop | [agents/azure_chat_agents_tool_calls_with_feedback.py](./agents/azure_chat_agents_tool_calls_with_feedback.py) | Two-agent workflow that streams tool calls and pauses for human guidance between passes | ### observability diff --git a/python/samples/getting_started/workflows/agents/azure_chat_agents_function_bridge.py b/python/samples/getting_started/workflows/agents/azure_chat_agents_function_bridge.py new file mode 100644 index 0000000000..e760f52c92 --- /dev/null +++ b/python/samples/getting_started/workflows/agents/azure_chat_agents_function_bridge.py @@ -0,0 +1,141 @@ +# Copyright (c) Microsoft. All rights reserved. + +import asyncio +from typing import Final + +from agent_framework import ( + AgentExecutorRequest, + AgentExecutorResponse, + AgentRunResponse, + AgentRunUpdateEvent, + ChatMessage, + Role, + WorkflowBuilder, + WorkflowContext, + WorkflowOutputEvent, + executor, +) +from agent_framework.azure import AzureOpenAIChatClient +from azure.identity import AzureCliCredential + +""" +Sample: Two agents connected by a function executor bridge + +Pipeline layout: +research_agent -> enrich_with_references (@executor) -> final_editor_agent + +The first agent drafts a short answer. A lightweight @executor function simulates +an external data fetch and injects a follow-up user message containing extra context. +The final agent incorporates the new note and produces the polished output. + +Demonstrates: +- Using the @executor decorator to create a function-style Workflow node. +- Consuming an AgentExecutorResponse and forwarding an AgentExecutorRequest for the next agent. +- Streaming AgentRunUpdateEvent events across agent + function + agent chain. + +Prerequisites: +- Azure OpenAI configured for AzureOpenAIChatClient with required environment variables. +- Authentication via azure-identity. Run `az login` before executing. +""" + +# Simulated external content keyed by a simple topic hint. +EXTERNAL_REFERENCES: Final[dict[str, str]] = { + "workspace": ( + "From Workspace Weekly: Adjustable monitor arms and sit-stand desks can reduce " + "neck strain by up to 30%. Consider adding a reminder to move every 45 minutes." + ), + "travel": ( + "Checklist excerpt: Always confirm baggage limits for budget airlines. " + "Keep a photocopy of your passport stored separately from the original." + ), + "wellness": ( + "Recent survey: Employees who take two 5-minute breaks per hour report 18% higher focus " + "scores. Encourage scheduling micro-breaks alongside hydration reminders." + ), +} + + +def _lookup_external_note(prompt: str) -> str | None: + """Return the first matching external note based on a keyword search.""" + lowered = prompt.lower() + for keyword, note in EXTERNAL_REFERENCES.items(): + if keyword in lowered: + return note + return None + + +@executor(id="enrich_with_references") +async def enrich_with_references( + draft: AgentExecutorResponse, + ctx: WorkflowContext[AgentExecutorRequest], +) -> None: + """Inject a follow-up user instruction that adds an external note for the next agent.""" + conversation = list(draft.full_conversation or draft.agent_run_response.messages) + original_prompt = next((message.text for message in conversation if message.role == Role.USER), "") + external_note = _lookup_external_note(original_prompt) or ( + "No additional references were found. Please refine the previous assistant response for clarity." + ) + + follow_up = ( + "External knowledge snippet:\n" + f"{external_note}\n\n" + "Please update the prior assistant answer so it weaves this note into the guidance." + ) + conversation.append(ChatMessage(role=Role.USER, text=follow_up)) + + await ctx.send_message(AgentExecutorRequest(messages=conversation)) + + +async def main() -> None: + """Run the workflow and stream combined updates from both agents.""" + chat_client = AzureOpenAIChatClient(credential=AzureCliCredential()) + + research_agent = chat_client.create_agent( + name="research_agent", + instructions=( + "Produce a short, bullet-style briefing with two actionable ideas. Label the section as 'Initial Draft'." + ), + ) + + final_editor_agent = chat_client.create_agent( + name="final_editor_agent", + instructions=( + "Use all conversation context (including external notes) to produce the final answer. " + "Merge the draft and extra note into a concise recommendation under 150 words." + ), + ) + + workflow = ( + WorkflowBuilder() + .add_agent(research_agent, id="research_agent") + .add_agent(final_editor_agent, id="final_editor_agent", output_response=True) + .add_edge(research_agent, enrich_with_references) + .add_edge(enrich_with_references, final_editor_agent) + .set_start_executor(research_agent) + .build() + ) + + events = workflow.run_stream( + "Create quick workspace wellness tips for a remote analyst working across two monitors." + ) + + last_executor: str | None = None + async for event in events: + if isinstance(event, AgentRunUpdateEvent): + if event.executor_id != last_executor: + if last_executor is not None: + print() + print(f"{event.executor_id}:", end=" ", flush=True) + last_executor = event.executor_id + print(event.data, end="", flush=True) + elif isinstance(event, WorkflowOutputEvent): + print("\n\n===== Final Output =====") + response = event.data + if isinstance(response, AgentRunResponse): + print(response.text or "(empty response)") + else: + print(response if response is not None else "No response generated.") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/python/samples/getting_started/workflows/agents/azure_chat_agents_tool_calls_with_feedback.py b/python/samples/getting_started/workflows/agents/azure_chat_agents_tool_calls_with_feedback.py new file mode 100644 index 0000000000..919db464e5 --- /dev/null +++ b/python/samples/getting_started/workflows/agents/azure_chat_agents_tool_calls_with_feedback.py @@ -0,0 +1,273 @@ +# Copyright (c) Microsoft. All rights reserved. + +import asyncio +import json +from dataclasses import dataclass, field +from typing import Annotated + +from agent_framework import ( + AgentExecutorRequest, + AgentExecutorResponse, + AgentRunUpdateEvent, + ChatMessage, + Executor, + FunctionCallContent, + FunctionResultContent, + RequestInfoEvent, + RequestInfoExecutor, + RequestInfoMessage, + RequestResponse, + Role, + ToolMode, + WorkflowBuilder, + WorkflowContext, + WorkflowOutputEvent, + handler, +) +from agent_framework.azure import AzureOpenAIChatClient +from azure.identity import AzureCliCredential +from pydantic import Field + +""" +Sample: Tool-enabled agents with human feedback + +Pipeline layout: +writer_agent (uses Azure OpenAI tools) -> DraftFeedbackCoordinator -> RequestInfoExecutor +-> DraftFeedbackCoordinator -> final_editor_agent + +The writer agent calls tools to gather product facts before drafting copy. A custom executor +packages the draft and emits a RequestInfoEvent so a human can comment, then replays the human +guidance back into the conversation before the final editor agent produces the polished output. + +Demonstrates: +- Attaching Python function tools to an agent inside a workflow. +- Capturing the writer's output and routing it through RequestInfoExecutor for human review. +- Streaming AgentRunUpdateEvent updates alongside human-in-the-loop pauses. + +Prerequisites: +- Azure OpenAI configured for AzureOpenAIChatClient with required environment variables. +- Authentication via azure-identity. Run `az login` before executing. +""" + + +def fetch_product_brief( + product_name: Annotated[str, Field(description="Product name to look up.")], +) -> str: + """Return a marketing brief for a product.""" + briefs = { + "lumenx desk lamp": ( + "Product: LumenX Desk Lamp\n" + "- Three-point adjustable arm with 270° rotation.\n" + "- Custom warm-to-neutral LED spectrum (2700K-4000K).\n" + "- USB-C charging pad integrated in the base.\n" + "- Designed for home offices and late-night study sessions." + ) + } + return briefs.get(product_name.lower(), f"No stored brief for '{product_name}'.") + + +def get_brand_voice_profile( + voice_name: Annotated[str, Field(description="Brand or campaign voice to emulate.")], +) -> str: + """Return guidance for the requested brand voice.""" + voices = { + "lumenx launch": ( + "Voice guidelines:\n" + "- Friendly and modern with concise sentences.\n" + "- Highlight practical benefits before aesthetics.\n" + "- End with an invitation to imagine the product in daily use." + ) + } + return voices.get(voice_name.lower(), f"No stored voice profile for '{voice_name}'.") + + +@dataclass +class DraftFeedbackRequest(RequestInfoMessage): + """Payload sent to RequestInfoExecutor for human review.""" + + prompt: str = "" + draft_text: str = "" + conversation: list[ChatMessage] = field(default_factory=list) # type: ignore[reportUnknownVariableType] + + +class DraftFeedbackCoordinator(Executor): + """Bridge between the writer agent, human feedback, and final editor.""" + + def __init__(self, *, id: str = "draft_feedback_coordinator") -> None: + super().__init__(id) + + @handler + async def on_writer_response( + self, + draft: AgentExecutorResponse, + ctx: WorkflowContext[DraftFeedbackRequest], + ) -> None: + # Preserve the full conversation so the final editor can see tool traces and the initial prompt. + conversation: list[ChatMessage] + if draft.full_conversation is not None: + conversation = list(draft.full_conversation) + else: + conversation = list(draft.agent_run_response.messages) + draft_text = draft.agent_run_response.text.strip() + if not draft_text: + draft_text = "No draft text was produced." + + prompt = ( + "Review the draft from the writer and provide a short directional note " + "(tone tweaks, must-have detail, target audience, etc.). " + "Keep it under 30 words." + ) + await ctx.send_message(DraftFeedbackRequest(prompt=prompt, draft_text=draft_text, conversation=conversation)) + + @handler + async def on_human_feedback( + self, + feedback: RequestResponse[DraftFeedbackRequest, str], + ctx: WorkflowContext[AgentExecutorRequest], + ) -> None: + note = (feedback.data or "").strip() + request = feedback.original_request + + conversation: list[ChatMessage] = list(request.conversation) + instruction = ( + "A human reviewer shared the following guidance:\n" + f"{note or 'No specific guidance provided.'}\n\n" + "Rewrite the draft from the previous assistant message into a polished final version. " + "Keep the response under 120 words and reflect any requested tone adjustments." + ) + conversation.append(ChatMessage(Role.USER, text=instruction)) + await ctx.send_message(AgentExecutorRequest(messages=conversation, should_respond=True)) + + +async def main() -> None: + """Run the workflow and bridge human feedback between two agents.""" + chat_client = AzureOpenAIChatClient(credential=AzureCliCredential()) + + writer_agent = chat_client.create_agent( + name="writer_agent", + instructions=( + "You are a marketing writer. Call the available tools before drafting copy so you are precise. " + "Always call both tools once before drafting. Summarize tool outputs as bullet points, then " + "produce a 3-sentence draft." + ), + tools=[fetch_product_brief, get_brand_voice_profile], + tool_choice=ToolMode.REQUIRED_ANY, + ) + + final_editor_agent = chat_client.create_agent( + name="final_editor_agent", + instructions=( + "You are an editor who polishes marketing copy using human guidance. " + "Respect factual details from the prior messages while applying the feedback." + ), + ) + + feedback_coordinator = DraftFeedbackCoordinator() + request_info_executor = RequestInfoExecutor(id="human_feedback") + + workflow = ( + WorkflowBuilder() + .add_agent(writer_agent, id="Writer") + .add_agent(final_editor_agent, id="FinalEditor", output_response=True) + .set_start_executor(writer_agent) + .add_edge(writer_agent, feedback_coordinator) + .add_edge(feedback_coordinator, request_info_executor) + .add_edge(request_info_executor, feedback_coordinator) + .add_edge(feedback_coordinator, final_editor_agent) + .build() + ) + + print( + "Interactive mode. When prompted, provide a short feedback note for the editor (type 'exit' to quit).", + flush=True, + ) + + pending_responses: dict[str, str] | None = None + completed = False + printed_tool_calls: set[str] = set() + printed_tool_results: set[str] = set() + + while not completed: + last_executor: str | None = None + stream = ( + workflow.send_responses_streaming(pending_responses) + if pending_responses is not None + else workflow.run_stream( + "Create a short launch blurb for the LumenX desk lamp. Emphasize adjustability and warm lighting." + ) + ) + pending_responses = None + requests: list[tuple[str, DraftFeedbackRequest]] = [] + + async for event in stream: + if isinstance(event, AgentRunUpdateEvent): + executor_id = event.executor_id + update = event.data + # Extract and print any new tool calls or results from the update. + function_calls = [c for c in update.contents if isinstance(c, FunctionCallContent)] # type: ignore[union-attr] + function_results = [c for c in update.contents if isinstance(c, FunctionResultContent)] # type: ignore[union-attr] + if executor_id != last_executor: + if last_executor is not None: + print() + print(f"{executor_id}:", end=" ", flush=True) + last_executor = executor_id + # Print any new tool calls before the text update. + for call in function_calls: + if call.call_id in printed_tool_calls: + continue + printed_tool_calls.add(call.call_id) + args = call.arguments + if isinstance(args, dict): + args_preview = json.dumps(args, ensure_ascii=False) + else: + args_preview = (args or "").strip() + print( + f"\n{executor_id} [tool-call] {call.name}({args_preview})", + flush=True, + ) + print(f"{executor_id}:", end=" ", flush=True) + # Print any new tool results before the text update. + for result in function_results: + if result.call_id in printed_tool_results: + continue + printed_tool_results.add(result.call_id) + result_text = result.result + if not isinstance(result_text, str): + result_text = json.dumps(result_text, ensure_ascii=False) + print( + f"\n{executor_id} [tool-result] {result.call_id}: {result_text}", + flush=True, + ) + print(f"{executor_id}:", end=" ", flush=True) + # Finally, print the text update. + print(update, end="", flush=True) + elif isinstance(event, RequestInfoEvent) and isinstance(event.data, DraftFeedbackRequest): + # Stash the request so we can prompt the human after the stream completes. + requests.append((event.request_id, event.data)) + last_executor = None + elif isinstance(event, WorkflowOutputEvent): + last_executor = None + response = event.data + print("\n===== Final output =====") + final_text = getattr(response, "text", str(response)) + print(final_text.strip()) + completed = True + + if requests and not completed: + responses: dict[str, str] = {} + for request_id, request in requests: + print("\n----- Writer draft -----") + print(request.draft_text.strip()) + print("\nProvide guidance for the editor (or press Enter to accept the draft).") + answer = input("Human feedback: ").strip() # noqa: ASYNC250 + if answer.lower() == "exit": + print("Exiting...") + return + responses[request_id] = answer + pending_responses = responses + + print("Workflow complete.") + + +if __name__ == "__main__": + asyncio.run(main())