mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
Python: fix tool call content not showing up in workflow events (#1290)
* fix tool call content not showing up in workflow events * fix lint * touch up * add another sample to show function bridge * missing sample
This commit is contained in:
Eric Zhu
committed by
GitHub
Unverified
GitHub
parent
334d52f300
commit
1c5e607a1f
@@ -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))
|
||||
|
||||
|
||||
@@ -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]:
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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())
|
||||
+273
@@ -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())
|
||||
Reference in New Issue
Block a user