ai-agents-for-beginners

Изучение Microsoft Agent Framework

Agent Framework

Введение

В этом уроке будут рассмотрены:

Цели обучения

После завершения этого урока вы сможете:

Примеры кода

Примеры кода для Microsoft Agent Framework (MAF) можно найти в этом репозитории в файлах xx-python-agent-framework и xx-dotnet-agent-framework.

Понимание Microsoft Agent Framework

Framework Intro

Microsoft Agent Framework (MAF) — это единая платформа Microsoft для создания ИИ-агентов. Она предлагает гибкость для решения широкого спектра агентских сценариев, встречающихся как в производственной среде, так и в научных исследованиях, включая:

Для внедрения ИИ-агентов в производство MAF также включает функции:

Microsoft Agent Framework также ориентирован на обеспечение совместимости:

Рассмотрим, как эти функции применяются к основным концепциям Microsoft Agent Framework.

Ключевые концепции Microsoft Agent Framework

Агенты

Agent Framework

Создание агентов

Создание агента осуществляется путём определения службы вывода (поставщика LLM), набора инструкций для выполнения ИИ-агентом и присвоения name:

agent = AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent( instructions="You are good at recommending trips to customers based on their preferences.", name="TripRecommender" )

В приведённом примере используется Azure OpenAI, но агенты могут создаваться с использованием различных служб, включая Microsoft Foundry Agent Service:

AzureAIAgentClient(async_credential=credential).create_agent( name="HelperAgent", instructions="You are a helpful assistant." ) as agent

OpenAI API Responses, ChatCompletion

agent = OpenAIResponsesClient().create_agent( name="WeatherBot", instructions="You are a helpful weather assistant.", )

agent = OpenAIChatClient().create_agent( name="HelpfulAssistant", instructions="You are a helpful assistant.", )

или MiniMax, который предоставляет совместимый с OpenAI API с большими контекстными окнами (до 204K токенов):

agent = OpenAIChatClient(base_url="https://api.minimax.io/v1", api_key=os.environ["MINIMAX_API_KEY"], model_id="MiniMax-M3").create_agent( name="HelpfulAssistant", instructions="You are a helpful assistant.", )

или удалённые агенты с использованием протокола A2A:

agent = A2AAgent( name=agent_card.name, description=agent_card.description, agent_card=agent_card, url="https://your-a2a-agent-host" )

Запуск агентов

Агенты запускаются с помощью методов .run или .run_stream для получения либо неряделённых, либо потоковых ответов.

result = await agent.run("What are good places to visit in Amsterdam?")
print(result.text)

async for update in agent.run_stream("What are the good places to visit in Amsterdam?"):
    if update.text:
        print(update.text, end="", flush=True)

Каждый запуск агента может иметь параметры для настройки, такие как max_tokens для ограничения числа токенов, tools — инструменты, доступные агенту, и даже используемая model.

Это полезно, когда для выполнения задачи пользователя требуются конкретные модели или инструменты.

Инструменты

Инструменты можно определять как при создании агента:

def get_attractions( location: Annotated[str, Field(description="The location to get the top tourist attractions for")], ) -> str: """Get the top tourist attractions for a given location.""" return f"The top attractions for {location} are." 


# При прямом создании ChatAgent

agent = ChatAgent( chat_client=OpenAIChatClient(), instructions="You are a helpful assistant", tools=[get_attractions]

так и при запуске агента:


result1 = await agent.run( "What's the best place to visit in Seattle?", tools=[get_attractions] # Инструмент предоставлен только для этого запуска )

Потоки агентов

Потоки агентов используются для обработки многоходовых диалогов. Потоки могут создаваться:

Для создания потока код выглядит так:

# Создать новый поток.
thread = agent.get_new_thread() # Запустить агента с этим потоком.
response = await agent.run("Hello, I am here to help you book travel. Where would you like to go?", thread=thread)

Поток можно сериализовать и сохранить для последующего использования:

# Создать новый поток.
thread = agent.get_new_thread() 

# Запустить агента с потоком.

response = await agent.run("Hello, how are you?", thread=thread) 

# Сериализовать поток для хранения.

serialized_thread = await thread.serialize() 

# Десериализовать состояние потока после загрузки из хранилища.

resumed_thread = await agent.deserialize_thread(serialized_thread)

Промежуточное ПО агента

Агенты взаимодействуют с инструментами и LLM для выполнения задач пользователя. В некоторых сценариях мы хотим выполнять действия или отслеживать события между этими взаимодействиями. Промежуточное ПО агента позволяет это сделать следующим образом:

Функциональное промежуточное ПО

Это промежуточное ПО позволяет выполнять действия между агентом и функцией/инструментом, которые вызываются. Пример — логирование вызова функции.

В коде ниже next определяет, вызывается ли следующее промежуточное ПО или сама функция.

async def logging_function_middleware(
    context: FunctionInvocationContext,
    next: Callable[[FunctionInvocationContext], Awaitable[None]],
) -> None:
    """Function middleware that logs function execution."""
    # Предобработка: Логирование перед выполнением функции
    print(f"[Function] Calling {context.function.name}")

    # Продолжить к следующему промежуточному ПО или выполнению функции
    await next(context)

    # Постобработка: Логирование после выполнения функции
    print(f"[Function] {context.function.name} completed")

Промежуточное ПО чата

Это промежуточное ПО позволяет выполнять действия или логировать запросы между агентом и LLM.

Оно содержит важную информацию, такую как messages, отправляемые в AI-сервис.

async def logging_chat_middleware(
    context: ChatContext,
    next: Callable[[ChatContext], Awaitable[None]],
) -> None:
    """Chat middleware that logs AI interactions."""
    # Предварительная обработка: логирование перед вызовом ИИ
    print(f"[Chat] Sending {len(context.messages)} messages to AI")

    # Продолжить к следующему промежуточному ПО или AI-сервису
    await next(context)

    # Последующая обработка: логирование после ответа ИИ
    print("[Chat] AI response received")

Память агента

Как описано в уроке Agentic Memory, память — важный элемент, позволяющий агенту работать в разных контекстах. MAF предлагает несколько типов памяти:

В оперативной памяти (In-Memory Storage)

Эта память хранится в потоках во время выполнения приложения.

# Создайте новый поток.
thread = agent.get_new_thread() # Запустите агента с этим потоком.
response = await agent.run("Hello, I am here to help you book travel. Where would you like to go?", thread=thread)

Постоянные сообщения (Persistent Messages)

Эта память используется для хранения истории общения между сессиями. Она определяется с помощью chat_message_store_factory:

from agent_framework import ChatMessageStore

# Создать пользовательское хранилище сообщений
def create_message_store():
    return ChatMessageStore()

agent = ChatAgent(
    chat_client=OpenAIChatClient(),
    instructions="You are a Travel assistant.",
    chat_message_store_factory=create_message_store
)

Динамическая память (Dynamic Memory)

Эта память добавляется к контексту перед запуском агентов. Она может храниться во внешних сервисах, таких как mem0:

from agent_framework.mem0 import Mem0Provider

# Использование Mem0 для расширенных возможностей памяти
memory_provider = Mem0Provider(
    api_key="your-mem0-api-key",
    user_id="user_123",
    application_id="my_app"
)

agent = ChatAgent(
    chat_client=OpenAIChatClient(),
    instructions="You are a helpful assistant with memory.",
    context_providers=memory_provider
)

Наблюдаемость агента

Наблюдаемость важна для создания надёжных и удобных в сопровождении агентских систем. MAF интегрируется с OpenTelemetry для трассировки и метрик, что улучшает наблюдаемость.

from agent_framework.observability import get_tracer, get_meter

tracer = get_tracer()
meter = get_meter()
with tracer.start_as_current_span("my_custom_span"):
    # сделать что-то
    pass
counter = meter.create_counter("my_custom_counter")
counter.add(1, {"key": "value"})

Рабочие процессы

MAF предлагает рабочие процессы — предопределённые шаги для выполнения задачи, включающие ИИ-агентов в качестве компонентов.

Рабочие процессы состоят из различных компонентов, обеспечивающих лучший контроль потока. Они также поддерживают оркестрацию нескольких агентов и контрольные точки для сохранения состояний рабочих процессов.

Основные компоненты рабочего процесса:

Исполнители (Executors)

Исполнители получают входящие сообщения, выполняют назначенные задачи и затем создают выходящее сообщение. Это продвигает рабочий процесс к выполнению основной задачи. Исполнителем может быть ИИ-агент или индивидуальная логика.

Ребра (Edges)

Ребра определяют поток сообщений в рабочем процессе. Они бывают:

Прямые ребра — простые однонаправленные соединения между исполнителями:

from agent_framework import WorkflowBuilder

builder = WorkflowBuilder()
builder.add_edge(source_executor, target_executor)
builder.set_start_executor(source_executor)
workflow = builder.build()

Условные ребра — активируются при выполнении определённого условия. Например, если номера в отеле недоступны, исполнитель может предложить другие варианты.

Ребра switch-case — маршрутизируют сообщения к разным исполнителям в зависимости от условий. Например, если у путешественника есть приоритетный доступ, его задачи обрабатываются другим рабочим процессом.

Ребра Fan-out — отправляют одно сообщение сразу нескольким получателям.

Ребра Fan-in — собирают несколько сообщений от разных исполнителей и отправляют одному получателю.

События

Для улучшения наблюдаемости рабочих процессов MAF предлагает встроенные события выполнения, включая:

Расширенные паттерны MAF

Выше были рассмотрены основные концепции Microsoft Agent Framework. При создании более сложных агентов учтите следующие расширенные паттерны:

Размещение агентов LangChain / LangGraph на Microsoft Foundry

Microsoft Agent Framework является фреймворк-совместимым — вы не ограничены агентами, написанными с помощью MAF. Если у вас уже есть агент, созданный с помощью LangChain или LangGraph, вы можете запустить его как хостинг агент Foundry, чтобы Foundry управляла временем выполнения, сессиями, масштабированием, идентификацией и протокольными конечными точками, при этом логика вашего агента остаётся в LangGraph.

Это реализуется с помощью пакета langchain_azure_ai.agents.hosting, который предоставляет скомпилированный граф LangGraph по тем же протоколам, что и агенты, размещённые в Foundry.

1. Установите extra для хостинга:

pip install -U "langchain-azure-ai[hosting]>=1.2.4" azure-identity

Пакет hosting устанавливает библиотеки протокола Foundry: azure-ai-agentserver-responses (конечная точка /responses, совместимая с OpenAI) и azure-ai-agentserver-invocations (универсальная конечная точка /invocations).

2. Выберите протокол хостинга:

Протокол Класс хоста Конечная точка Используйте, если
Responses ResponsesHostServer /responses Вам нужен чат, потоковая обработка, история ответов и многоходовые диалоги, совместимые с OpenAI — рекомендуемый вариант для разговорных агентов.
Invocations InvocationsHostServer /invocations Требуется настраиваемый JSON, точка webhook или неконверсационная обработка.

Поскольку Responses API является основным API для разработки агентов в Foundry, начните с ResponsesHostServer для большинства агентов.

3. Настройте переменные окружения (сначала выполните az login, чтобы DefaultAzureCredential мог аутентифицироваться):

export FOUNDRY_PROJECT_ENDPOINT="https://<resource>.services.ai.azure.com/api/projects/<project>"
export FOUNDRY_MODEL_NAME="gpt-4.1"

При том, что агент позже запускается как хостинг агент в Foundry, платформа автоматически добавит FOUNDRY_PROJECT_ENDPOINT.

4. Предоставьте агент LangGraph через протокол Responses:

import os

from azure.ai.projects import AIProjectClient
from azure.identity import DefaultAzureCredential, get_bearer_token_provider
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_azure_ai.agents.hosting import ResponsesHostServer

_AZURE_AI_SCOPE = "https://ai.azure.com/.default"


def build_chat_model() -> ChatOpenAI:
    project_endpoint = os.environ["FOUNDRY_PROJECT_ENDPOINT"].rstrip("/")
    deployment = os.environ.get("FOUNDRY_MODEL_NAME", "gpt-4.1")
    credential = DefaultAzureCredential()
    project = AIProjectClient(endpoint=project_endpoint, credential=credential)
    openai_client = project.get_openai_client()
    token_provider = get_bearer_token_provider(credential, _AZURE_AI_SCOPE)

    # ChatOpenAI здесь нацелен на совместимую с OpenAI конечную точку (Responses) проекта Foundry.
    return ChatOpenAI(
        model=deployment,
        base_url=str(openai_client.base_url),
        api_key=token_provider,
    )


def main() -> None:
    graph = create_agent(build_chat_model(), tools=[])
    port = int(os.environ.get("PORT", "8088"))
    ResponsesHostServer(graph).run(port=port)


if __name__ == "__main__":
    main()

Запустите локально с помощью python main.py, затем отправьте запрос Responses на http://localhost:8088/responses.

Ключевые особенности:

Рабочий пример этого можно найти в code-samples/14-langchain-hosted-agent.py. Полное руководство (протокол Invocations, кастомные схемы запросов и устранение неполадок) смотрите в Host LangGraph agents as Foundry hosted agents.

Примеры кода

Примеры кода для Microsoft Agent Framework доступны в этом репозитории в файлах xx-python-agent-framework и xx-dotnet-agent-framework.

Есть вопросы по Microsoft Agent Framework?

Присоединяйтесь к Microsoft Foundry Discord, чтобы общаться с другими учащимися, посещать часы консультаций и получать ответы на вопросы по AI-агентам.

Предыдущий урок

Память для AI-агентов

Следующий урок

Создание агентов для использования ПК (CUA)

Отказ от ответственности: Этот документ был переведен с использованием сервиса машинного перевода Co-op Translator. Несмотря на наши усилия по обеспечению точности, имейте в виду, что автоматический перевод может содержать ошибки или неточности. Оригинальный документ на его исходном языке следует считать авторитетным источником. Для получения критически важной информации рекомендуется обратиться к профессиональному человеческому переводу. Мы не несем ответственности за любые недоразумения или неправильные толкования, возникшие в результате использования этого перевода.

This site is open source. Improve this page.