إطار عمل وكلاء مايكروسوفت: بناء سير عمل متعدد الوكلاء بلغة بايثون

mkdir support-agents && cd support-agents
python -m venv .venv
source .venv/bin/activate   # على ويندوز: .venv\Scripts\activate
 
pip install agent-framework python-dotenv

# .env
OPENAI_API_KEY=sk-...
OPENAI_CHAT_MODEL=gpt-4o-mini

# hello.py
import asyncio
from dotenv import load_dotenv
 
from agent_framework import Agent
from agent_framework.openai import OpenAIChatClient
 
load_dotenv()
 
async def main() -> None:
    agent = Agent(
        client=OpenAIChatClient(),
        name="HelloAgent",
        instructions="You are a friendly support assistant. Keep answers brief.",
    )
 
    result = await agent.run("A customer asks: where is my order?")
    print(result)
 
if __name__ == "__main__":
    asyncio.run(main())

python hello.py

print("Agent: ", end="", flush=True)
async for chunk in agent.run("Give me a one-sentence apology template.", stream=True):
    if chunk.text:
        print(chunk.text, end="", flush=True)
print()

# tools.py
from typing import Annotated
from pydantic import Field
 
def get_order_status(
    order_id: Annotated[str, Field(description="The customer's order ID, e.g. 'A-1042'")]
) -> str:
    """Look up the shipping status of an order by its ID."""
    # في الإنتاج يصل هذا إلى قاعدة بيانات الطلبات أو واجهة برمجية.
    fake_db = {
        "A-1042": "Shipped on 2026-06-14, arriving 2026-06-19.",
        "A-1099": "Processing — not yet shipped.",
    }
    return fake_db.get(order_id, "No order found with that ID.")
 
def get_refund_policy(
    region: Annotated[str, Field(description="Customer region: 'EU', 'US', or 'MENA'")]
) -> str:
    """Return the refund window and conditions for a region."""
    policies = {
        "EU": "14-day no-questions-asked returns under EU consumer law.",
        "US": "30-day returns with receipt.",
        "MENA": "14-day returns for unused items in original packaging.",
    }
    return policies.get(region, "Standard 14-day return policy applies.")

# support_agent.py
import asyncio
from dotenv import load_dotenv
 
from agent_framework import Agent
from agent_framework.openai import OpenAIChatClient
 
from tools import get_order_status, get_refund_policy
 
load_dotenv()
 
async def main() -> None:
    agent = Agent(
        client=OpenAIChatClient(),
        name="SupportAgent",
        instructions=(
            "You are a support agent. Use the tools to look up real order "
            "and policy data before answering. Never invent order details."
        ),
        tools=[get_order_status, get_refund_policy],
    )
 
    result = await agent.run(
        "Hi, where is order A-1042, and can I return it? I'm in the EU."
    )
    print(result)
 
if __name__ == "__main__":
    asyncio.run(main())

# session_demo.py
import asyncio
from dotenv import load_dotenv
 
from agent_framework import Agent
from agent_framework.openai import OpenAIChatClient
from tools import get_order_status
 
load_dotenv()
 
async def main() -> None:
    agent = Agent(
        client=OpenAIChatClient(),
        name="SupportAgent",
        instructions="You are a concise support agent.",
        tools=[get_order_status],
    )
 
    session = agent.create_session()
 
    r1 = await agent.run("Where is order A-1042?", session=session)
    print("Turn 1:", r1)
 
    # السؤال التالي بلا رقم طلب — الجلسة تحمل السياق.
    r2 = await agent.run("And will it arrive before the weekend?", session=session)
    print("Turn 2:", r2)
 
if __name__ == "__main__":
    asyncio.run(main())

# pipeline_manual.py
import asyncio
from dotenv import load_dotenv
 
from agent_framework import Agent
from agent_framework.openai import OpenAIChatClient
from tools import get_order_status, get_refund_policy
 
load_dotenv()
 
def make_agent(name: str, instructions: str, tools=None) -> Agent:
    return Agent(
        client=OpenAIChatClient(),
        name=name,
        instructions=instructions,
        tools=tools or [],
    )
 
async def main() -> None:
    triage = make_agent(
        "Triage",
        "Classify the ticket. Reply with exactly: 'CATEGORY: <billing|shipping|other> | "
        "URGENCY: <low|high>'. Nothing else.",
    )
    support = make_agent(
        "Support",
        "Draft a warm, accurate reply. Use tools for real data.",
        tools=[get_order_status, get_refund_policy],
    )
    qa = make_agent(
        "QA",
        "You review support drafts. If the tone is professional and the facts are "
        "grounded, reply with the draft unchanged. Otherwise rewrite it.",
    )
 
    ticket = "URGENT: order A-1099 still not here and I leave the country tomorrow! EU customer."
 
    label = await triage.run(ticket)
    print("Triage:", label)
 
    draft = await support.run(f"Ticket: {ticket}\nClassification: {label}")
    print("Draft:", draft)
 
    final = await qa.run(f"Review this draft:\n{draft}")
    print("Final:", final)
 
if __name__ == "__main__":
    asyncio.run(main())

# workflow_functional.py
import asyncio
from dotenv import load_dotenv
 
from agent_framework import Agent, workflow, step
from agent_framework.openai import OpenAIChatClient
from tools import get_order_status, get_refund_policy
 
load_dotenv()
 
client = OpenAIChatClient()
 
triage = Agent(client=client, name="Triage",
               instructions="Reply 'URGENCY: high' or 'URGENCY: low' and one category word.")
support = Agent(client=client, name="Support",
               instructions="Draft an accurate, friendly reply. Use tools for real data.",
               tools=[get_order_status, get_refund_policy])
qa = Agent(client=client, name="QA",
           instructions="Approve or rewrite the draft for tone and factual grounding.")
 
@step
async def classify(ticket: str) -> dict:
    label = str(await triage.run(ticket))
    return {"ticket": ticket, "label": label, "high": "high" in label.lower()}
 
@step
async def draft_reply(state: dict) -> dict:
    draft = str(await support.run(f"Ticket: {state['ticket']}\nLabel: {state['label']}"))
    state["draft"] = draft
    return state
 
@step
async def review(state: dict) -> str:
    return str(await qa.run(f"Review and finalize:\n{state['draft']}"))
 
@workflow
async def support_pipeline(ticket: str, ctx) -> str:
    state = await classify(ticket)
 
    # تفرّع بايثون الأصلي: تصعيد التذاكر العاجلة إلى إنسان.
    if state["high"]:
        decision = await ctx.request_info(
            f"High-urgency ticket needs approval before auto-reply:\n{state['label']}"
        )
        if str(decision).strip().lower().startswith("no"):
            return "Escalated to a human agent. No automated reply sent."
 
    state = await draft_reply(state)
    return await review(state)
 
async def main() -> None:
    result = await support_pipeline.run(
        "URGENT: order A-1099 not delivered, leaving tomorrow! EU."
    )
    print(result)
 
if __name__ == "__main__":
    asyncio.run(main())

# workflow_graph.py
import asyncio
from dotenv import load_dotenv
 
from agent_framework import (
    Agent, WorkflowBuilder, executor, WorkflowContext,
)
from agent_framework.openai import OpenAIChatClient
from tools import get_order_status, get_refund_policy
 
load_dotenv()
 
client = OpenAIChatClient()
 
support = Agent(client=client, name="Support",
                instructions="Draft an accurate, friendly reply. Use tools for real data.",
                tools=[get_order_status, get_refund_policy])
qa = Agent(client=client, name="QA",
           instructions="Finalize the draft for tone and accuracy.")
 
@executor(id="classify")
async def classify(ticket: str, ctx: WorkflowContext[str]) -> None:
    label = str(await Agent(
        client=client, name="Triage",
        instructions="Reply with one category word and 'high' or 'low' urgency.",
    ).run(ticket))
    # خزّن التذكرة الأصلية في الحالة المشتركة للمنفّذين اللاحقين.
    ctx.set_state("ticket", ticket)
    await ctx.send_message(label)
 
@executor(id="draft")
async def draft(label: str, ctx: WorkflowContext[str]) -> None:
    ticket = ctx.get_state("ticket")
    reply = str(await support.run(f"Ticket: {ticket}\nLabel: {label}"))
    await ctx.send_message(reply)
 
@executor(id="review")
async def review(reply: str, ctx: WorkflowContext) -> None:
    final = str(await qa.run(f"Finalize:\n{reply}"))
    await ctx.yield_output(final)
 
async def main() -> None:
    workflow = (
        WorkflowBuilder()
        .set_start_executor(classify)
        .add_edge(classify, draft)
        .add_edge(draft, review)
        .build()
    )
 
    async for event in workflow.run("Where is order A-1042? EU customer.", stream=True):
        if event.type == "output":
            print("FINAL:", event.data)
 
if __name__ == "__main__":
    asyncio.run(main())

from agent_framework.observability import setup_observability
 
# استدعِها مرة عند الإقلاع. تقرأ OTEL_EXPORTER_OTLP_ENDPOINT من البيئة.
setup_observability()

# test_tools.py
from tools import get_order_status, get_refund_policy
 
def test_known_order():
    assert "Shipped" in get_order_status("A-1042")
 
def test_unknown_order():
    assert "No order found" in get_order_status("ZZZ")
 
def test_region_policy():
    assert "14-day" in get_refund_policy("EU")
    assert "30-day" in get_refund_policy("US")