Repository navigation
Expand file tree
/
Copy pathazure_ai_agents_with_shared_session.py
More file actions
113 lines (90 loc) · 3.96 KB
/
Copy pathazure_ai_agents_with_shared_session.py
File metadata and controls
113 lines (90 loc) · 3.96 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from agent_framework import (
Agent,
AgentExecutor,
AgentExecutorRequest,
AgentExecutorResponse,
InMemoryHistoryProvider,
WorkflowBuilder,
WorkflowContext,
WorkflowRunState,
executor,
)
from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
"""
Sample: Agents with a shared thread in a workflow
A Writer agent generates content, then a Reviewer agent critiques it, sharing a common message thread.
Purpose:
Show how to use a shared thread between multiple agents in a workflow.
By default, agents have individual threads, but sharing a thread allows them to share all messages.
Notes:
- Not all agents can share threads; usually only the same type of agents can share threads.
Demonstrate:
- Creating multiple agents with FoundryChatClient.
- Setting up a shared thread between agents.
Prerequisites:
- FOUNDRY_PROJECT_ENDPOINT must be your Microsoft Foundry Agent Service (V2) project endpoint.
- FOUNDRY_MODEL must be set to your Azure OpenAI model deployment name.
- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample.
- Basic familiarity with agents, workflows, and executors in the agent framework.
"""
@executor(id="intercept_agent_response")
async def intercept_agent_response(
agent_response: AgentExecutorResponse, ctx: WorkflowContext[AgentExecutorRequest]
) -> None:
"""This executor intercepts the agent response and sends a request without messages.
This essentially prevents duplication of messages in the shared thread. Without this
executor, the response will be added to the thread as input of the next agent call.
"""
await ctx.send_message(AgentExecutorRequest(messages=[]))
async def main() -> None:
client = FoundryChatClient(
project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
model=os.environ["FOUNDRY_MODEL"],
credential=AzureCliCredential(),
)
# set the same context provider (same default source_id) for both agents to share the thread
writer = Agent(
client=client,
instructions=("You are a concise copywriter. Provide a single, punchy marketing sentence based on the prompt."),
name="writer",
context_providers=[InMemoryHistoryProvider()],
)
reviewer = Agent(
client=client,
instructions=("You are a thoughtful reviewer. Give brief feedback on the previous assistant message."),
name="reviewer",
context_providers=[InMemoryHistoryProvider()],
)
# Create the shared session
shared_session = writer.create_session()
writer_executor = AgentExecutor(writer, session=shared_session)
reviewer_executor = AgentExecutor(reviewer, session=shared_session)
workflow = (
WorkflowBuilder(start_executor=writer_executor)
.add_chain([writer_executor, intercept_agent_response, reviewer_executor])
.build()
)
result = await workflow.run(
"Write a tagline for a budget-friendly eBike.",
# client_kwargs are forwarded to each underlying chat client call.
# store=False tells the model API not to persist messages server-side
# for this example.
client_kwargs={"store": False},
)
# The final state should be IDLE since the workflow no longer has messages to
# process after the reviewer agent responds.
assert result.get_final_state() == WorkflowRunState.IDLE
# The shared session now contains the conversation between the writer and reviewer. Print it out.
print("=== Shared Session Conversation ===")
memory_state = shared_session.state.get(InMemoryHistoryProvider.DEFAULT_SOURCE_ID, {})
for message in memory_state.get("messages", []):
print(f"{message.author_name or message.role}: {message.text}")
if __name__ == "__main__":
asyncio.run(main())