Сравнение фреймворков AI-агентов CrewAI, LangGraph и BeeAI

from crewai import Agent, Task, Crew
 
# Создаем исследовательского агента
researcher = Agent(
    role="Исследователь рынка",
    goal="Собрать всю релевантную информацию о рынке криптовалют",
    backstory="Ты опытный аналитик с 10-летним стажем в финансах",
    verbose=True,
    allow_delegation=True
)
 
# Создаем аналитика
analyst = Agent(
    role="Финансовый аналитик",
    goal="Проанализировать данные и найти инвестиционные возможности",
    backstory="Ты скрупулезный аналитик, который видит паттерны в данных",
    verbose=True
)
 
# Задача для исследователя
research_task = Task(
    description="Изучи текущую ситуацию на рынке криптовалют",
    expected_output="Отчет о текущих трендах и данных",
    agent=researcher
)
 
# Задача для аналитика
analysis_task = Task(
    description="Проанализируй отчет и найди перспективные возможности",
    expected_output="Список из 3 перспективных криптовалют с обоснованием",
    agent=analyst
)
 
# Создаем команду и запускаем процес
crew = Crew(
    agents=[researcher, analyst],
    tasks=[research_task, analysis_task],
    verbose=2
)
 
result = crew.kickoff()

# Создаем команду с иерархическим процессом
crew = Crew(
    agents=[senior_researcher, junior_researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.hierarchical,
    verbose=2
)

# Создаем инструмент для поиска
search_tool = SearchTool()
 
# Добавляем инструмент к агенту
researcher = Agent(
    role="Исследователь",
    goal="Найти информацию",
    tools=[search_tool],
    verbose=True
)

# Включение расширенного логирования
import logging
logging.basicConfig(level=logging.INFO)
 
# Создание агента с активированным трейсингом
researcher = Agent(
    role="Исследователь",
    goal="Собрать данные",
    backstory="Опытный исследователь с большим стажем",
    verbose=True,
    tracing=True  # Включение трейсинга для отладки
)

from crewai.llm import OLLAMA, OpenAI
 
# Создание агентов с разными моделями
researcher = Agent(
    role="Исследователь",
    llm=OLLAMA(model="llama3:70b"),
    verbose=True
)
 
writer = Agent(
    role="Писатель",
    llm=OpenAI(model="gpt-4"),
    verbose=True
)

# Пример реализации паттерна Master-Slave в CrewAI
master_agent = Agent(
    role="Руководитель проекта",
    goal="Координировать работу команды для создания бизнес-плана",
    backstory="Опытный менеджер проектов с MBA из Гарварда",
    allow_delegation=True  # Ключевой параметр для мастер-агента
)
 
slave_agents = [
    Agent(role="Маркетолог", goal="Разработать маркетинговую стратегию"),
    Agent(role="Финансист", goal="Подготовить финансовые прогнозы"),
    Agent(role="Аналитик", goal="Исследовать конкурентов")
]
 
crew = Crew(
    agents=[master_agent] + slave_agents,
    process=Process.hierarchical
)

# Пример рефлексивного паттерна в LangGraph
def improve_code(state):
    code = state["code"]
    critique = "Код работает, но не оптимально. Можно улучшить..."
    improved_code = "# Улучшенная версия\ndef better_solution()..."
    return {"code": improved_code, "critique": critique}
 
builder = StateGraph(...)
builder.add_node("generate_code", generate_code_function)
builder.add_node("improve_code", improve_code)
# Создаем цикл для улучшения
builder.add_edge("generate_code", "improve_code")
builder.add_edge("improve_code", "improve_code", condition=needs_improvement)
builder.add_edge("improve_code", "END", condition=is_good_enough)

from langgraph.graph import StateGraph
from typing import TypedDict, Literal
 
# Определяем структуру состояния
class AgentState(TypedDict):
    messages: list
    current_step: str
    context: dict
    
# Создаем функции-обработчики для узлов графа
def search_information(state: AgentState) -> AgentState:
    # Поиск информации
    return {"messages": state["messages"] + ["Найдена информация..."], 
            "current_step": "search", 
            "context": {"search_results": "..."}}
 
def analyze_data(state: AgentState) -> AgentState:
    # Анализ полученных данных
    return {"messages": state["messages"] + ["Анализ завершен..."], 
            "current_step": "analyze", 
            "context": state["context"] | {"analysis": "..."}}
 
# Создаем функцию-маршрутизатор
def router(state: AgentState) -> Literal["search", "analyze", "summarize"]:
    if "search_results" not in state["context"]:
        return "search"
    if "analysis" not in state["context"]:
        return "analyze"
    return "summarize"
 
# Строим граф
builder = StateGraph(AgentState)
builder.add_node("search", search_information)
builder.add_node("analyze", analyze_data)
builder.add_node("summarize", summarize_function)
 
# Определяем маршрутизацию
builder.add_conditional_edges("router", router)
builder.add_edge("search", "router")
builder.add_edge("analyze", "router")
builder.add_edge("summarize", "END")
 
# Компилируем граф
graph = builder.compile()

# Сохраняем состояние между вызовами
state = graph.invoke({"messages": [], "current_step": "", "context": {}})
[H2]Где-то в хранилище сохраняем state...[/H2]
 
# В следующем запросе восстанавливаем и продолжаем
restored_state = get_state_from_storage()  # Получаем из БД или кеша
new_state = graph.invoke(restored_state)

def query_database(state: AgentState) -> AgentState:
    # Извлекаем параметры запроса из состояния
    query_params = state["context"].get("query_params", {})
    
    # Выполняем запрос к БД
    results = db_client.execute_query(query_params)
    
    # Обновляем состояние
    return {
        "messages": state["messages"] + [f"Получены результаты из БД: {results}"],
        "current_step": "db_query",
        "context": state["context"] | {"db_results": results}
    }
 
# Добавляем узел в граф
builder.add_node("db_query", query_database)

# Визуализация графа в формате DOT
dot_graph = builder.to_dot()
with open("agent_graph.dot", "w") as f:
    f.write(dot_graph)

def compress_context(state: AgentState) -> AgentState:
    # Берем только ключевую информацию из контекста
    full_context = state["context"]
    compressed = {
        "key_entities": extract_entities(full_context),
        "summary": summarize_context(full_context),
        "decisions": full_context.get("decisions", [])
    }
    
    return {
        "messages": state["messages"][-5:],  # Храним только последние сообщения
        "current_step": state["current_step"],
        "context": compressed
    }
 
# Добавляем узел сжатия после тяжелых операций
builder.add_node("compress", compress_context)
builder.add_edge("heavy_processing", "compress")
builder.add_edge("compress", "next_step")

import os
from langsmith import Client
 
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
 
client = Client()
# Теперь все запуски будут отслеживаться в LangSmith

def build_dynamic_graph(context):
    builder = StateGraph(AgentState)
    
    # Базовые узлы, всегда присутствующие
    builder.add_node("init", initialize_state)
    builder.add_node("finish", finalize_state)
    
    # Динамическое добавление узлов на основе контекста
    if "data_analysis" in context["requirements"]:
        builder.add_node("analyze", analyze_data)
        builder.add_edge("init", "analyze")
        builder.add_edge("analyze", "finish")
    else:
        builder.add_edge("init", "finish")
    
    return builder.compile()

# Пример создания простого агента в BeeAI
from beeai import Agent
 
agent = Agent(
    name="FinancialAnalyst",
    description="Специалист по анализу финансовых данных",
    model="ibm/granite-20b-instruct", # IBM модель
    tools=[search_tool, calculator_tool]
)
 
result = agent.run("Проанализируй финансовые показатели компании Apple за последний квартал")

# Настройка телеметрии в BeeAI
from beeai import Telemetry
 
telemetry = Telemetry(
    enabled=True,
    log_level="DEBUG",
    storage_path="./telemetry_logs"
)
 
agent = Agent(
    name="Debuggable",
    telemetry=telemetry,
    # другие параметры
)

# Настройка памяти с ключевыми моментами
from beeai import Memory, KeyPointsStrategy
 
memory = Memory(
    strategy=KeyPointsStrategy(
        window_size=10,  # Количество сообщений в окне
        key_points_count=3  # Сколько ключевых моментов сохранять
    )
)
 
agent = Agent(
    name="MemoryOptimized",
    memory=memory,
    # другие параметры
)

# Использование песочницы для выполнения кода
from beeai import CodeSandbox
 
sandbox = CodeSandbox(
    timeout=5,  # Максимальное время выполнения в секундах
    allowed_modules=["pandas", "numpy", "math"],  # Разрешенные модули
    max_memory="100MB"  # Ограничение памяти
)
 
agent = Agent(
    name="CodeExecutor",
    code_sandbox=sandbox,
    # другие параметры
)
 
result = agent.run("Напиши и выполни код для расчета медианного значения ряда чисел")

# Настройка ограничений безопасности
from beeai import SecurityConstraints
 
security = SecurityConstraints(
    data_access_level="aggregated",
    allowed_apis=["public_market_data", "internal_analytics"],
    data_exfiltration_protection=True,
    sensitive_data_detection=True
)
 
agent = Agent(
    name="RiskAnalyst",
    security_constraints=security,
    # другие параметры
)

# Настройка планировщика задач
from beeai import TaskScheduler, SchedulingStrategy
 
scheduler = TaskScheduler(
    strategy=SchedulingStrategy.PRIORITY_WITH_DEPENDENCIES,
    max_concurrent_tasks=10,
    priority_levels=3
)
 
workflow = AgentWorkflow(
    agents=[agent1, agent2, agent3],
    scheduler=scheduler
)

# Добавление функциональности обратной связи в LangGraph
def process_feedback(state):
    feedback = state["user_feedback"]
    # Анализируем обратную связь и корректируем ответ
    corrected_response = f"Спасибо за обратную связь! Исправленный ответ: {modified_answer}"
    return {"messages": state["messages"] + [corrected_response], "has_feedback": True}
 
# Добавляем новый узел
builder.add_node("feedback_handler", process_feedback)
 
# Добавляем условный переход
def has_feedback(state):
    return "feedback" if "user_feedback" in state else "no_feedback"
 
builder.add_conditional_edges("check_feedback", has_feedback,
                             {
                                 "feedback": "feedback_handler",
                                 "no_feedback": "END"
                             })
 
# Соединяем с существующим графом
builder.add_edge("generate_response", "check_feedback")
builder.add_edge("feedback_handler", "END")

# Интеграционное приложение для анализа научных статей
import os
from typing import TypedDict, List, Dict, Any
from crewai import Agent, Task, Crew, Process
from langgraph.graph import StateGraph
import beeai
 
# Структура состояния для LangGraph
class AnalysisState(TypedDict):
    research_data: Dict[str, Any]
    analysis_results: List[Dict[str, Any]]
    summary: str
    current_stage: str
 
# Шаг 1: Создаем агентов CrewAI для сбора и первичной обработки данных
def setup_crewai():
    researcher = Agent(
        role="Научный исследователь",
        goal="Собрать и структурировать данные из научных статей",
        backstory="Ты опытный исследователь с докторской степенью",
        verbose=True
    )
    
    analyst = Agent(
        role="Аналитик данных",
        goal="Проанализировать структурированные данные и выявить паттерны",
        backstory="Ты аналитик с опытом работы в научной сфере",
        verbose=True
    )
    
    research_task = Task(
        description="Собери данные из указанных научных статей по теме ИИ-агентов",
        expected_output="Структурированный набор данных",
        agent=researcher
    )
    
    analysis_task = Task(
        description="Проанализируй собранные данные и выяви ключевые паттерны",
        expected_output="Аналитический отчет с выявленными паттернами",
        agent=analyst
    )
    
    crew = Crew(
        agents=[researcher, analyst],
        tasks=[research_task, analysis_task],
        process=Process.sequential,
        verbose=2
    )
    
    return crew
 
# Шаг 2: Создаем граф состояний LangGraph для управления процессом
def setup_langgraph():
    # Функции-обработчики для узлов графа
    def process_research_data(state: AnalysisState) -> AnalysisState:
        # Обработка данных исследования
        processed_data = state["research_data"]
        # Здесь логика обработки...
        return {**state, "processed_data": processed_data, "current_stage": "data_processed"}
    
    def analyze_processed_data(state: AnalysisState) -> AnalysisState:
        # Анализ обработанных данных
        analysis = []
        # Логика анализа...
        return {**state, "analysis_results": analysis, "current_stage": "analysis_complete"}
    
    def generate_summary(state: AnalysisState) -> AnalysisState:
        # Генерация итогового отчета
        summary = "Сводка по результатам анализа: ..."
        return {**state, "summary": summary, "current_stage": "summary_complete"}
    
    def determine_next_step(state: AnalysisState):
        current = state.get("current_stage", "")
        if current == "":
            return "process_data"
        elif current == "data_processed":
            return "analyze_data"
        elif current == "analysis_complete":
            return "generate_summary"
        else:
            return "END"
    
    # Создаем граф
    builder = StateGraph(AnalysisState)
    builder.add_node("process_data", process_research_data)
    builder.add_node("analyze_data", analyze_processed_data)
    builder.add_node("generate_summary", generate_summary)
    
    # Настраиваем маршрутизацию
    builder.add_conditional_edges("router", determine_next_step, {
        "process_data": "process_data",
        "analyze_data": "analyze_data",
        "generate_summary": "generate_summary",
        "END": "END"
    })
    
    # Соединяем узлы
    builder.add_edge("process_data", "router")
    builder.add_edge("analyze_data", "router")
    builder.add_edge("generate_summary", "END")
    
    # Компилируем граф
    graph = builder.compile()
    return graph
 
# Шаг 3: Используем BeeAI для оптимизации и масштабирования
def setup_beeai():
    # Настройка BeeAI для оптимизации использования токенов
    memory = beeai.Memory(
        strategy=beeai.KeyPointsStrategy(
            window_size=10,
            key_points_count=3
        )
    )
    
    # Создаем агента BeeAI с оптимизированной памятью
    summary_agent = beeai.Agent(
        name="SummarySpe******t",
        description="Специалист по созданию итоговых отчетов на основе научных данных",
        memory=memory,
        model="ibm/granite-20b-instruct"
    )
    
    # Настраиваем телеметрию
    telemetry = beeai.Telemetry(
        enabled=True,
        log_level="INFO",
        storage_path="./telemetry_logs"
    )
    
    # Создаем рабочий процесс
    workflow = beeai.AgentWorkflow(
        agents=[summary_agent],
        telemetry=telemetry
    )
    
    return workflow
 
# Главная функция интеграции всех трех фреймворков
def integrated_analysis_system(research_papers):
    # Шаг 1: Используем CrewAI для сбора и первичного анализа
    crew = setup_crewai()
    crew_results = crew.kickoff(inputs={"papers": research_papers})
    
    # Шаг 2: Передаем результаты в LangGraph для обработки потоков данных
    graph = setup_langgraph()
    graph_state = graph.invoke({
        "research_data": crew_results,
        "analysis_results": [],
        "summary": "",
        "current_stage": ""
    })
    
    # Шаг 3: Используем BeeAI для оптимизации итогового отчета
    workflow = setup_beeai()
    final_report = workflow.run_task(
        "Создай финальный отчет на основе проведенного анализа",
        context={"analysis_state": graph_state}
    )
    
    return {
        "crew_results": crew_results,
        "graph_state": graph_state,
        "final_report": final_report
    }
 
# Пример использования
if __name__ == "__main__":
    research_papers = [
        {"title": "Advances in Multi-Agent Systems", "url": "https://example.com/paper1"},
        {"title": "Collaborative AI Frameworks", "url": "https://example.com/paper2"},
        {"title": "Performance Analysis of Agent Networks", "url": "https://example.com/paper3"}
    ]
    
    results = integrated_analysis_system(research_papers)
    print(f"Итоговый отчет:\n{results['final_report']}")

# Пример более детальной обработки перехода между CrewAI и LangGraph
def crew_to_langgraph(crew_results):
    # Преобразуем неструктурированный текст в структурированные данные
    structured_data = {
        "documents": [],
        "metadata": {"source": "CrewAI", "timestamp": datetime.now().isoformat()}
    }
    
    # Парсим результаты CrewAI
    if isinstance(crew_results, str):
        # Простой случай: текстовый результат
        try:
            # Пытаемся распарсить как JSON
            import json
            parsed = json.loads(crew_results)
            structured_data["documents"] = parsed.get("documents", [])
        except:
            # Если не JSON, разбираем текст по маркерам
            sections = crew_results.split("---")
            structured_data["documents"] = [{"content": s.strip()} for s in sections if s.strip()]
    elif isinstance(crew_results, dict):
        # Словарь, копируем напрямую
        structured_data = crew_results
    
    return structured_data

# Использование Redis для обмена данными между компонентами
import redis
import json
 
redis_client = redis.Redis(host='localhost', port=6379, db=0)
 
# В компоненте CrewAI
def save_crew_results(job_id, results):
    redis_client.set(f"crew:{job_id}", json.dumps(results))
    redis_client.expire(f"crew:{job_id}", 3600)  # Храним час
 
# В компоненте LangGraph
def load_crew_results(job_id):
    data = redis_client.get(f"crew:{job_id}")
    if data:
        return json.loads(data)
    return None