from crewai import Agent, Task, Crew, Process
# 定义智能体角色
researcher = Agent(
    role='Research Specialist',
    goal='Conduct comprehensive research on given topics',
    backstory='You are an expert researcher with 10 years of experience',
    verbose=True,
    allow_delegation=False,
    tools=[search_tool, scrape_tool]
)
writer = Agent(
    role='Content Writer',
    goal='Create engaging and informative content',
    backstory='You are a skilled writer specializing in technical content',
    verbose=True,
    allow_delegation=True
)
# 定义任务流程
research_task = Task(
    description='Research the latest trends in AI agent orchestration',
    agent=researcher,
    expected_output='Detailed research report with key findings'
)
writing_task = Task(
    description='Write a comprehensive article based on research findings',
    agent=writer,
    context=[research_task],  # 依赖关系定义
    expected_output='Well-structured technical article'
)
# 创建团队并执行
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,  # 顺序执行模式
    verbose=2
)
result = crew.kickoff()

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolExecutor
import operator
# 定义状态结构
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    next_action: str
    intermediate_steps: list
# 定义节点函数
def research_node(state: AgentState):
    """研究节点：执行信息收集任务"""
    messages = state["messages"]
    # 调用研究智能体
    research_agent = create_research_agent()
    response = research_agent.invoke(messages)
    
    return {
        "messages": [response],
        "next_action": "analyze",
        "intermediate_steps": state["intermediate_steps"] + ["research_completed"]
    }
def analyze_node(state: AgentState):
    """分析节点：处理收集的信息"""
    messages = state["messages"]
    analysis_agent = create_analysis_agent()
    response = analysis_agent.invoke(messages)
    
    return {
        "messages": [response],
        "next_action": "write" if len(messages) > 3 else "research",
        "intermediate_steps": state["intermediate_steps"] + ["analysis_completed"]
    }
def should_continue(state: AgentState):
    """条件判断：决定下一步执行路径"""
    if state["next_action"] == "end":
        return END
    return state["next_action"]
# 构建状态图
workflow = StateGraph(AgentState)
workflow.add_node("research", research_node)
workflow.add_node("analyze", analyze_node)
workflow.add_node("write", write_node)
# 定义边和条件
workflow.add_edge("research", "analyze")
workflow.add_conditional_edges("analyze", should_continue)
workflow.set_entry_point("research")
# 编译并执行
app = workflow.compile()
result = app.invoke({"messages": [], "next_action": "research", "intermediate_steps": []})

# 适合CrewAI的场景示例：内容创作流水线
content_crew = Crew(
    agents=[
        seo_specialist,    # SEO专家
        content_writer,    # 内容创作者
        editor,           # 编辑
        publisher         # 发布者
    ],
    tasks=[seo_analysis, content_creation, editing, publishing],
    process=Process.sequential
)

# 适合LangGraph的场景示例：智能客服系统
customer_service_graph = StateGraph(CustomerState)
customer_service_graph.add_node("intent_recognition", recognize_intent)
customer_service_graph.add_node("knowledge_query", query_knowledge)
customer_service_graph.add_node("human_escalation", escalate_to_human)
customer_service_graph.add_conditional_edges(
    "intent_recognition", 
    route_based_on_confidence,
    {
        "high_confidence": "knowledge_query",
        "low_confidence": "human_escalation",
        "ambiguous": "clarification"
    }
)

from crewai import Agent, Task, Crew
from crewai.tools import BaseTool
class RobustAgent(Agent):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.max_retries = 3
        self.retry_count = 0
    
    def execute_task(self, task):
        try:
            return super().execute_task(task)
        except Exception as e:
            if self.retry_count < self.max_retries:
                self.retry_count += 1
                print(f"Retrying task... Attempt {self.retry_count}")
                return self.execute_task(task)
            else:
                # 失败处理：委托给其他智能体或人工干预
                return self.handle_failure(task, e)
    
    def handle_failure(self, task, error):
        # 实现失败恢复逻辑
        return {"status": "failed", "error": str(error), "fallback": True}
# 使用示例
robust_researcher = RobustAgent(
    role='Robust Researcher',
    goal='Research with error recovery',
    backstory='Expert researcher with fallback mechanisms',
    tools=[search_tool, backup_search_tool]
)

from langgraph.graph import StateGraph
from typing import TypedDict
import json
class RobustAgentState(TypedDict):
    messages: list
    current_step: str
    error_count: int
    backup_plan: dict
    checkpoint: dict  # 检查点机制
def error_recovery_node(state: RobustAgentState):
    """错误恢复节点"""
    if state["error_count"] > 0:
        # 从检查点恢复
        if "checkpoint" in state and state["checkpoint"]:
            print(f"Recovering from checkpoint: {state['checkpoint']['step']}")
            return {
                **state,
                "current_step": state["checkpoint"]["step"],
                "messages": state["checkpoint"]["messages"],
                "error_count": 0
            }
        
        # 执行备用方案
        backup_plan = state.get("backup_plan", {})
        if backup_plan:
            return execute_backup_plan(state, backup_plan)
    
    return state
def create_checkpoint(state: RobustAgentState):
    """创建检查点"""
    return {
        **state,
        "checkpoint": {
            "step": state["current_step"],
            "messages": state["messages"].copy(),
            "timestamp": time.time()
        }
    }
# 构建带错误恢复的工作流
robust_workflow = StateGraph(RobustAgentState)
robust_workflow.add_node("checkpoint", create_checkpoint)
robust_workflow.add_node("error_recovery", error_recovery_node)
robust_workflow.add_node("main_task", main_task_node)
robust_workflow.add_edge("checkpoint", "main_task")
robust_workflow.add_conditional_edges(
    "main_task", 
    lambda state: "error_recovery" if state["error_count"] > 0 else "checkpoint"
)

from crewai import Agent, Tool
from langchain.tools import DuckDuckGoSearchRun
from langchain.agents import initialize_agent, AgentType
from langchain.llms import OpenAI
# 使用LangChain工具
search_tool = Tool(
    name="Web Search",
    description="Search the web for information",
    func=DuckDuckGoSearchRun().run
)
# 创建混合智能体
hybrid_agent = Agent(
    role='Hybrid Research Agent',
    goal='Leverage both CrewAI and LangChain capabilities',
    backstory='Expert in using multiple AI frameworks',
    tools=[search_tool],
    llm=OpenAI(temperature=0.1)  # 使用LangChain的LLM
)

from langgraph.graph import StateGraph
import autogen
class AutoGenLangGraphState(TypedDict):
    autogen_chat: dict
    langgraph_state: dict
    conversation_history: list
def autogen_chat_node(state: AutoGenLangGraphState):
    """集成AutoGen的多智能体对话"""
    config_list = [{"model": "gpt-4", "api_key": "your-key"}]
    
    assistant = autogen.AssistantAgent(
        name="assistant",
        llm_config={"config_list": config_list}
    )
    
    user_proxy = autogen.UserProxyAgent(
        name="user_proxy",
        human_input_mode="NEVER",
        code_execution_config={"work_dir": "coding"}
    )
    
    # 执行AutoGen对话
    chat_result = user_proxy.initiate_chat(
        assistant, 
        message=state["langgraph_state"]["current_message"]
    )
    
    return {
        **state,
        "autogen_chat": chat_result,
        "conversation_history": state["conversation_history"] + [chat_result]
    }
# 集成工作流
integrated_workflow = StateGraph(AutoGenLangGraphState)
integrated_workflow.add_node("autogen_chat", autogen_chat_node)
integrated_workflow.add_node("langgraph_processing", langgraph_processing_node)