Files
Evan Mattson 4aa737eee5 Python: [BREAKING] Require building functional workflow instances (#7521)
* Harden functional workflow continuation authority

Use a versioned opaque single-use token on WorkflowRunResult, validate it before request correlation, consume it immediately before replayed user code, and rotate it on each pause. Carry the same explicit authority through streaming and non-streaming FunctionalWorkflowAgent responses.

Files changed: functional workflow/runtime result APIs, functional HITL regression tests, core agent guidance, and the functional HITL sample.

Next iteration: enforce pending-state overlap and token-authorized abandonment, then document and test checkpoint authorization boundaries.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Enforce one pending functional continuation

Reject fresh messages and checkpoint restores while an in-memory continuation is pending. Add token-authorized abandonment on FunctionalWorkflow and FunctionalWorkflowAgent, and clear retained replay state atomically when authority is consumed while preserving the active message for token rotation and checkpoints.

Files changed: functional workflow runtime and agent adapter, functional lifecycle regression tests, and core workflow guidance.

Next iteration: preserve and document authorized checkpoint continuation boundaries.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Preserve authorized functional checkpoint continuation

Treat checkpoint restore as a host- and storage-authorized path independent of process-local continuation tokens, and issue fresh authority whenever restored execution pauses again. Cover default and per-run storage, deterministic and custom request IDs, token rotation, and checkpoint-plus-response restore.

Files changed: functional workflow and checkpoint interface guidance, functional checkpoint lifecycle tests, the functional HITL sample, and core workflow guidance.

Next iteration: run the final repository-wide Python validation gates.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Validate Python continuation hardening

Run the complete Python workspace checks, aggregate coverage suite, repository hooks, and core package build from the final combined worktree. Keep the validation iteration code-neutral because all gates pass without corrective changes.

Files changed: none; this commit records the final validation gate.

Blockers: none. Next iteration: no remaining AFK tasks.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Handle functional checkpoint continuation failures

Publish retained continuation state only after checkpoint persistence succeeds, and cover reuse after a transient save failure.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Copilot-Session: a8f47743-1cdc-4924-8e1b-667d0d790b78

* Address functional continuation review findings

Add owner recovery for lost tokens, harden malformed token validation, preserve consistent failure surfaces, and keep agent pending state aligned with resumable workflow state.

Document process-local single-use continuation semantics and extend regression coverage across direct, streaming, checkpoint, and agent paths.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Copilot-Session: a8f47743-1cdc-4924-8e1b-667d0d790b78

* Handle functional continuation cancellation

Release the workflow run guard when cancellation interrupts resumed user code while keeping the single-use continuation token consumed.

Replace sample assertions with explicit runtime checks and add cancellation regression coverage.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Copilot-Session: a8f47743-1cdc-4924-8e1b-667d0d790b78

* Simplify functional workflow instance isolation

Remove continuation-token handling and align functional workflows with the graph workflow ownership model: one stateful instance per logical caller or session.

Add create_instance() for independent callers, document the ownership contract, and cover pending-state isolation between instances.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Copilot-Session: a8f47743-1cdc-4924-8e1b-667d0d790b78

* Scope functional workflow checkpoint storage

Do not inherit checkpoint storage when creating an independent workflow instance. Allow hosts to provide an explicitly caller-scoped storage adapter and document that shared checkpoint access requires host authorization and tenant isolation.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Copilot-Session: a8f47743-1cdc-4924-8e1b-667d0d790b78

* Require building functional workflow instances

Make @workflow return a stateless FunctionalWorkflowDefinition and require build() before run() or as_agent(). This aligns functional workflows with the graph definition/build lifecycle and prevents module-level decorated definitions from retaining caller state.

Move checkpoint configuration to build(), export the definition type, migrate samples, and cover isolated built instances.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Copilot-Session: a8f47743-1cdc-4924-8e1b-667d0d790b78

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: a8f47743-1cdc-4924-8e1b-667d0d790b78
2026-08-14 00:31:52 +00:00

84 lines
2.8 KiB
Python

# Copyright (c) Microsoft. All rights reserved.
"""Naive group chat using the functional workflow API.
A simple round-robin group chat where agents take turns responding.
Because it's just a function, you control the loop, the turn order,
and the termination condition with plain Python — no framework abstractions.
Compare this with the graph-based GroupChat orchestration to see how the
functional API lets you start simple and add complexity only when needed.
"""
import asyncio
from agent_framework import Agent, Message, workflow
from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential
# ---------------------------------------------------------------------------
# Create agents
# ---------------------------------------------------------------------------
client = FoundryChatClient(credential=AzureCliCredential())
expert = Agent(
name="PythonExpert",
instructions=(
"You are a Python expert in a group discussion. "
"Answer questions about Python and refine your answer based on feedback. "
"Keep responses concise (2-3 sentences)."
),
client=client,
)
critic = Agent(
name="Critic",
instructions=(
"You are a constructive critic in a group discussion. "
"Point out edge cases, gotchas, or missing nuances in the previous answer. "
"If the answer is solid, say so briefly."
),
client=client,
)
summarizer = Agent(
name="Summarizer",
instructions=(
"You are a summarizer in a group discussion. "
"After the discussion, provide a final concise summary that incorporates "
"the expert's answer and the critic's feedback. Keep it to 2-3 sentences."
),
client=client,
)
# ---------------------------------------------------------------------------
# A naive group chat is just a loop — no special framework needed
# ---------------------------------------------------------------------------
@workflow
async def group_chat(question: str) -> str:
"""Round-robin group chat: expert answers, critic reviews, summarizer wraps up."""
participants = [expert, critic, summarizer]
# Passing list[Message] keeps roles/authorship intact between turns,
# instead of stringifying everything into a single prompt.
conversation: list[Message] = [Message("user", [question])]
# Simple round-robin: each agent sees the full conversation so far
for agent in participants:
response = await agent.run(conversation)
conversation.extend(response.messages)
return "\n\n".join(f"{m.author_name or m.role}: {m.text}" for m in conversation)
async def main():
workflow_instance = group_chat.build()
result = await workflow_instance.run("What's the difference between a list and a tuple in Python?")
print(result.get_outputs()[0])
if __name__ == "__main__":
asyncio.run(main())