Обзор фреймворков для AI-агентов

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict, Annotated
import operator, anthropic

client = anthropic.Anthropic()

# ── Состояние агента ─────────────────────────────────────────
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]   # += слияние списков
    iterations: int

# ── Узлы графа ───────────────────────────────────────────────
def node_agent(state: AgentState) -> dict:
    """Вызов LLM — ядро агента."""
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        messages=state["messages"],
    )
    return {
        "messages": [{"role": "assistant", "content": response.content}],
        "iterations": state["iterations"] + 1,
    }

def node_tools(state: AgentState) -> dict:
    """Выполнение всех запрошенных инструментов."""
    last_msg = state["messages"][-1]["content"]
    results = []
    for block in last_msg:
        if hasattr(block, "type") and block.type == "tool_use":
            result = dispatch_tool(block.name, block.input)
            results.append({"type": "tool_result", "tool_use_id": block.id, "content": result})
    return {"messages": [{"role": "user", "content": results}]}

# ── Роутер (conditional edge) ────────────────────────────────
def router(state: AgentState) -> str:
    last = state["messages"][-1]["content"]
    if state["iterations"] >= 10:
        return "end"
    for block in (last if isinstance(last, list) else []):
        if hasattr(block, "type") and block.type == "tool_use":
            return "tools"
    return "end"

# ── Сборка графа ─────────────────────────────────────────────
builder = StateGraph(AgentState)
builder.add_node("agent", node_agent)
builder.add_node("tools", node_tools)

builder.set_entry_point("agent")
builder.add_conditional_edges("agent", router, {"tools": "tools", "end": END})
builder.add_edge("tools", "agent")   # после инструментов — снова к агенту

graph = builder.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["agent"],       # Human-in-the-loop: пауза перед каждым шагом
)

# ── Запуск ───────────────────────────────────────────────────
config = {"configurable": {"thread_id": "session-1"}}
initial_state = {"messages": [{"role": "user", "content": "Найди последние новости"}], "iterations": 0}

for chunk in graph.stream(initial_state, config=config):
    print(chunk)   # стриминг каждого шага

from crewai import Agent, Task, Crew, Process
from crewai.tools import tool

# ── Инструмент ───────────────────────────────────────────────
@tool("Search Web")
def search_web(query: str) -> str:
    """Поиск актуальной информации в интернете."""
    return f"Результаты поиска по '{query}': ..."

# ── Агенты (роли с автоматическим system prompt) ─────────────
researcher = Agent(
    role="Senior AI Researcher",
    goal="Найти актуальные и достоверные данные по теме {topic}",
    backstory=(
        "Опытный исследователь с 10-летним стажем. "
        "Специализируется на быстром поиске и верификации источников."
    ),
    tools=[search_web],
    verbose=True,
)

writer = Agent(
    role="Technical Writer",
    goal="Написать понятный технический отчёт на основе исследования",
    backstory="Технический автор, умеющий превращать сложные данные в читаемые тексты.",
    tools=[],
)

# ── Задачи (привязаны к агентам) ─────────────────────────────
research_task = Task(
    description="Изучи тему '{topic}': найди последние события, статистику, экспертные мнения.",
    expected_output="Структурированный список из 5-7 ключевых фактов с источниками.",
    agent=researcher,
)

write_task = Task(
    description="На основе результатов исследования напиши отчёт на русском языке.",
    expected_output="Markdown-отчёт с заголовками, тезисами и выводом.",
    agent=writer,
    context=[research_task],   # автоматически получает результат research_task
)

# ── Команда ──────────────────────────────────────────────────
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff(inputs={"topic": "AI Agents в 2025"})
print(result.raw)

import autogen

# ── Конфигурация LLM ─────────────────────────────────────────
llm_config = {
    "config_list": [{"model": "gpt-4o", "api_key": "sk-..."}],
    "temperature": 0,
}

# ── Агент: генерирует и исправляет код ───────────────────────
assistant = autogen.AssistantAgent(
    name="Coder",
    system_message=(
        "Ты опытный Python-разработчик. "
        "Пиши чистый, рабочий код. "
        "Если код не прошёл тесты — исправь ошибку."
    ),
    llm_config=llm_config,
)

# ── Прокси: выполняет код, может спрашивать пользователя ─────
user_proxy = autogen.UserProxyAgent(
    name="Executor",
    human_input_mode="TERMINATE",       # спросить человека только при завершении
    max_consecutive_auto_reply=10,
    is_termination_msg=lambda x: "TERMINATE" in x.get("content", ""),
    code_execution_config={
        "work_dir": "coding",
        "use_docker": False,             # True в prod — изоляция через Docker
    },
)

# ── Запуск диалога ───────────────────────────────────────────
user_proxy.initiate_chat(
    assistant,
    message=(
        "Напиши Python-скрипт для анализа данных из файла sales.csv: "
        "загрузи данные, посчитай топ-10 продуктов по выручке, "
        "построй график и сохрани в report.png."
    ),
)

# ── GroupChat с несколькими агентами ────────────────────────
coder    = autogen.AssistantAgent("Coder",    llm_config=llm_config, system_message="Пишешь код.")
reviewer = autogen.AssistantAgent("Reviewer", llm_config=llm_config, system_message="Проверяешь код. Говоришь APPROVED или объясняешь что исправить.")
executor = autogen.UserProxyAgent("Executor", human_input_mode="NEVER", code_execution_config={"work_dir": "coding"})

groupchat = autogen.GroupChat(
    agents=[executor, coder, reviewer],
    messages=[],
    max_round=20,
    speaker_selection_method="auto",   # LLM выбирает следующего спикера
)
manager = autogen.GroupChatManager(groupchat=groupchat, llm_config=llm_config)

executor.initiate_chat(manager, message="Реши задачу сортировки массива за O(n log n)")

import asyncio
from agents import Agent, Runner, function_tool

# ── Инструменты ──────────────────────────────────────────────
@function_tool
def get_order_status(order_id: str) -> str:
    """Получить статус заказа по его ID."""
    orders = {"ORD-123": "доставлен", "ORD-456": "в пути"}
    return orders.get(order_id, "заказ не найден")

@function_tool
def process_refund(order_id: str, reason: str) -> str:
    """Оформить возврат средств по заказу."""
    return f"Возврат по заказу {order_id} оформлен. Причина: {reason}. Срок: 3-5 дней."

@function_tool
def search_faq(query: str) -> str:
    """Поиск по базе знаний и FAQ."""
    return f"Статья по теме '{query}': [ссылка на документ]"

# ── Специализированные агенты ────────────────────────────────
billing_agent = Agent(
    name="Billing Specialist",
    instructions=(
        "Ты специалист по расчётам и возвратам. "
        "Помогай с вопросами о заказах, счетах и возвратах средств."
    ),
    tools=[get_order_status, process_refund],
)

support_agent = Agent(
    name="Support Specialist",
    instructions=(
        "Ты агент поддержки. "
        "Отвечай на общие вопросы, ищи информацию в базе знаний."
    ),
    tools=[search_faq],
)

# ── Агент-роутер ─────────────────────────────────────────────
triage_agent = Agent(
    name="Triage Agent",
    instructions=(
        "Ты диспетчер. Определи тип запроса и передай нужному агенту:\n"
        "- Вопросы о заказах, платежах, возвратах → billing_agent\n"
        "- Все остальные вопросы → support_agent\n"
        "Не отвечай сам — только маршрутизируй."
    ),
    handoffs=[billing_agent, support_agent],
)

# ── Запуск ───────────────────────────────────────────────────
async def main():
    result = await Runner.run(triage_agent, "Хочу вернуть деньги за заказ ORD-123")
    print(result.final_output)

asyncio.run(main())