第七章：CrewAI 角色扮演与 Flow 事件驱动

from crewai import Agent

senior_researcher = Agent(
    role="Senior Research Analyst",
    goal="Uncover cutting-edge developments in AI agent frameworks",
    backstory="""You are a veteran technology researcher with 15 years 
    of experience tracking AI/ML trends. You've published extensively 
    in top conferences and have a knack for identifying which technologies 
    will become mainstream. You're known for your rigorous methodology: 
    always citing sources, cross-referencing claims, and distinguishing 
    hype from substance.""",
)

# 坏的 backstory（装饰性，不提供信息）
backstory_bad = "You are a smart and helpful AI assistant."

# 好的 backstory（结构化，功能性）
backstory_good = """You are a senior backend engineer at a fintech company, 
specializing in distributed systems and payment processing. You've handled 
3 major incidents involving race conditions in transaction processing. 
Your debugging approach: reproduce first, then bisect, then fix. 
You never guess—you always verify with logs and metrics."""

from crewai import Agent, LLM

agent = Agent(
    # === 核心三元组 ===
    role="Senior Data Analyst",
    goal="Provide accurate, actionable insights from complex datasets",
    backstory="...",
    
    # === LLM 配置 ===
    llm=LLM(model="gpt-4", temperature=0.1),  # 指定模型
    
    # === 工具能力 ===
    tools=[search_tool, calculator_tool],       # 可用工具列表
    
    # === MCP 服务器（2025新增）===
    mcps=[
        MCPServer(url="http://localhost:3000"),  # 连接 MCP 服务
    ],
    
    # === 技能（Skills，2025新增）===
    skills=[data_analysis_skill, visualization_skill],
    
    # === 知识源 ===
    knowledge_sources=[                         # RAG 知识库
        PDFKnowledgeSource(path="company_data.pdf"),
        StringKnowledgeSource(content="公司规则..."),
    ],
    
    # === 行为控制 ===
    max_iter=15,                # 最大思考迭代次数（防止无限循环）
    allow_delegation=True,      # 是否允许将任务委托给其他 Agent
    verbose=True,               # 是否打印思考过程
    
    # === 推理能力（2025新增）===
    reasoning=True,             # 启用 Chain-of-Thought 增强推理
    
    # === 内存 ===
    memory=True,                # 启用长期记忆
    
    # === 缓存 ===
    cache=True,                 # 缓存工具调用结果
    
    # === 执行控制 ===
    max_rpm=10,                 # 每分钟最大请求数（限流）
    max_retry_limit=2,          # 失败重试次数
)

from crewai import Crew, Process

crew = Crew(
    agents=[researcher, writer, editor],  # 团队成员
    tasks=[research_task, write_task, edit_task],  # 任务列表
    process=Process.sequential,  # 执行模式
    verbose=True,
)

# 启动执行
result = crew.kickoff(inputs={"topic": "AI Agent 2025 趋势"})

crew = Crew(
    # === 核心配置 ===
    agents=[...],           # Agent 列表
    tasks=[...],            # Task 列表
    process=Process.sequential,  # 执行模式
    
    # === 管理者配置（仅 Hierarchical 模式）===
    manager_llm=LLM(model="gpt-4"),  # Manager Agent 的 LLM
    manager_agent=None,              # 或直接提供自定义 Manager Agent
    
    # === 执行配置 ===
    verbose=True,           # 打印执行过程
    memory=True,            # 启用团队共享记忆
    cache=True,             # 缓存工具调用
    max_rpm=25,             # 全团队请求限流
    
    # === 输出配置 ===
    output_log_file="crew_log.txt",  # 日志文件
    
    # === 高级配置 ===
    planning=True,          # 启用任务前自动规划
    planning_llm=LLM(model="gpt-4"),  # 规划用的 LLM
)

from crewai import Crew, Task, Process

# 任务按顺序定义
research_task = Task(
    description="Research the topic: {topic}",
    expected_output="A comprehensive research report with sources",
    agent=researcher,
)

write_task = Task(
    description="Write an article based on the research",
    expected_output="A 2000-word article ready for editing",
    agent=writer,
    context=[research_task],  # 显式声明依赖上一个任务的输出
)

edit_task = Task(
    description="Edit and polish the article",
    expected_output="A publication-ready article",
    agent=editor,
    context=[write_task],
)

crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, write_task, edit_task],
    process=Process.sequential,  # 流水线模式
)

crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, write_task, edit_task],
    process=Process.hierarchical,  # 层级模式
    manager_llm=LLM(model="gpt-4"),  # Manager 使用的 LLM
)

from crewai import Task

task = Task(
    # === 核心定义 ===
    description="分析竞品 X 的技术架构，重点关注其 Agent 编排层的实现方式",
    expected_output="一份 2000 字的技术分析报告，包含架构图、核心设计决策、优缺点分析",
    agent=researcher,  # 指定负责的 Agent
    
    # === 上下文依赖 ===
    context=[previous_task_1, previous_task_2],  # 依赖哪些前置任务的输出
    
    # === 输出配置 ===
    output_file="reports/competitor_analysis.md",  # 将输出保存到文件
    output_json=AnalysisReport,  # 输出为结构化 JSON（Pydantic model）
    output_pydantic=AnalysisReport,  # 同上，用 Pydantic 验证
    
    # === 执行配置 ===
    async_execution=False,  # 是否异步执行
    
    # === Human-in-the-Loop ===
    human_input=True,  # 执行前请求人工确认/补充
    
    # === 回调 ===
    callback=on_task_complete,  # 完成后的回调函数
)

# 坏的 expected_output（太模糊）
expected_output_bad = "A good report"

# 好的 expected_output（具体、可验证）
expected_output_good = """一份结构化的竞品分析报告，包含：
1. 架构概览图（文字描述）
2. 核心组件列表及其职责
3. 关键设计决策（至少 3 个）及其 trade-off 分析
4. 与我们方案的对比表格
5. 总结：3 个可借鉴的优点 + 2 个应避免的缺点"""

# Task 1 的输出
research_output = "AI Agent 框架趋势报告..."

# Task 2 声明依赖 Task 1
write_task = Task(
    description="基于研究报告写一篇博客",
    context=[research_task],  # ← 这里
    agent=writer,
)

# 实际执行时，writer 看到的 prompt 是：
# [System] You are a Writer...
# [Context from previous tasks]
# Task: Research Analyst's output: "AI Agent 框架趋势报告..."
# [Current Task]
# 基于研究报告写一篇博客

from crewai.flow.flow import Flow, listen, start, router, or_, and_

class ContentPipeline(Flow):
    
    @start()
    def begin(self):
        """标记为起始节点——Flow 从这里开始执行"""
        print("Pipeline started!")
        return {"topic": self.state.topic}
    
    @listen(begin)
    def research(self, event):
        """监听 begin 事件——begin 完成后自动触发 research"""
        research_crew = Crew(agents=[researcher], tasks=[research_task])
        result = research_crew.kickoff(inputs=event)
        return {"research": result.raw}
    
    @listen(research)
    def write_draft(self, event):
        """监听 research 事件——research 完成后触发"""
        writing_crew = Crew(agents=[writer], tasks=[write_task])
        result = writing_crew.kickoff(inputs=event)
        return {"draft": result.raw}
    
    @router(write_draft)
    def review_router(self, event):
        """路由节点——根据条件决定下一步去哪"""
        if self.state.quality_score > 0.8:
            return "publish"   # 质量合格 → 发布
        elif self.state.iterations >= 3:
            return "publish"   # 达到上限 → 强制发布
        else:
            return "revise"    # 需要修改 → 返回修改
    
    @listen("publish")
    def publish(self, event):
        """监听路由的 'publish' 输出"""
        print(f"Published: {self.state.draft[:100]}")
    
    @listen("revise")
    def revise(self, event):
        """监听路由的 'revise' 输出——形成循环"""
        self.state.iterations += 1
        # 修改后重新触发 write_draft
        return self.write_draft(event)

@start()
def initialize(self):
    """Flow 的入口点。一个 Flow 可以有多个 @start 方法（并行启动）"""
    pass

@listen(source_method)  # 监听另一个方法的完成事件
def handler(self, event):
    pass

@listen("route_name")   # 监听 router 返回的字符串
def handler(self, event):
    pass

@router(source_method)
def my_router(self, event):
    """返回字符串，决定触发哪个下游 @listen"""
    if condition_a:
        return "path_a"
    else:
        return "path_b"

from crewai.flow.flow import human_feedback

class ReviewFlow(Flow):
    
    @listen(generate_draft)
    @human_feedback("请审核以下草稿，输入修改意见或输入 'approve' 通过：")
    def human_review(self, event, feedback):
        """等待人工输入"""
        if feedback.lower() == "approve":
            self.state.approved = True
        else:
            self.state.feedback = feedback
            self.state.approved = False
        return {"approved": self.state.approved}

from crewai.flow.flow import Flow
from pydantic import BaseModel

class ArticleState(BaseModel):
    """Flow 的状态模型——用 Pydantic 定义"""
    topic: str = ""
    draft: str = ""
    feedback: list[str] = []
    quality_score: float = 0.0
    iterations: int = 0
    approved: bool = False

class ArticleFlow(Flow[ArticleState]):
    """泛型参数指定 State 类型"""
    
    @start()
    def begin(self):
        # 通过 self.state 访问和修改状态
        self.state.topic = "AI Agent 框架对比"
        print(f"Starting with topic: {self.state.topic}")

class StateProxy:
    """线程安全的状态代理"""
    
    def __init__(self, state: BaseModel):
        self._state = state
        self._lock = threading.RLock()
    
    def __getattr__(self, name):
        with self._lock:
            return getattr(self._state, name)
    
    def __setattr__(self, name, value):
        if name.startswith('_'):
            super().__setattr__(name, value)
        else:
            with self._lock:
                setattr(self._state, name, value)

# 不安全的复合操作（可能产生竞态条件）
self.state.count = self.state.count + 1  
# 线程 A 读到 count=5，线程 B 也读到 count=5
# 两者都写入 6，实际应该是 7

# 安全做法：使用 list append（原子操作）
self.state.feedback.append("新反馈")
# 或者使用 Flow 提供的原子操作方法

from crewai.flow.flow import or_

class ParallelFlow(Flow):
    
    @start()
    def search_google(self):
        return {"source": "google", "results": [...]}
    
    @start()
    def search_bing(self):
        return {"source": "bing", "results": [...]}
    
    @listen(or_(search_google, search_bing))
    def process_first_result(self, event):
        """只要 google 或 bing 任一个返回结果，立即开始处理"""
        print(f"Got first result from: {event['source']}")

from crewai.flow.flow import and_

class GatherFlow(Flow):
    
    @start()
    def research_market(self):
        return {"market_data": [...]}
    
    @start()
    def research_technology(self):
        return {"tech_data": [...]}
    
    @start()
    def research_competition(self):
        return {"competition_data": [...]}
    
    @listen(and_(research_market, research_technology, research_competition))
    def synthesize(self, events):
        """三个研究全部完成后，综合分析"""
        # events 是一个列表，包含所有上游的返回值
        all_data = merge_research(events)
        return {"synthesis": all_data}

@listen(and_(
    or_(fast_search, slow_search),  # 搜索：任一个返回即可
    mandatory_validation,            # 验证：必须完成
))
def proceed(self, events):
    """搜索（快的或慢的）完成 AND 验证完成 → 继续"""
    pass

class ProductLaunchFlow(Flow[LaunchState]):
    """产品发布流程：研究 → 内容制作 → 审核 → 发布"""
    
    @start()
    def market_research(self):
        """阶段 1：市场研究（一组 Agent 协作）"""
        research_crew = Crew(
            agents=[
                Agent(role="Market Analyst", ...),
                Agent(role="Competitor Researcher", ...),
                Agent(role="User Researcher", ...),
            ],
            tasks=[
                Task(description="分析目标市场规模和趋势", ...),
                Task(description="调研主要竞品的定位和策略", ...),
                Task(description="总结目标用户的痛点和需求", ...),
            ],
            process=Process.sequential,
        )
        result = research_crew.kickoff()
        self.state.research_report = result.raw
        return result.raw
    
    @listen(market_research)
    def content_creation(self, research):
        """阶段 2：内容制作（另一组 Agent 协作）"""
        content_crew = Crew(
            agents=[
                Agent(role="Copywriter", ...),
                Agent(role="Designer", ...),
                Agent(role="SEO Specialist", ...),
            ],
            tasks=[
                Task(description="撰写产品文案", context_from=research, ...),
                Task(description="设计视觉素材方案", ...),
                Task(description="优化 SEO 关键词", ...),
            ],
            process=Process.hierarchical,  # 由 Manager 动态分配
            manager_llm=LLM(model="gpt-4"),
        )
        result = content_crew.kickoff()
        self.state.content_package = result.raw
        return result.raw
    
    @router(content_creation)
    def quality_gate(self, content):
        """阶段 3：质量门禁"""
        # 可以用 LLM 做自动评分
        score = self._evaluate_quality(content)
        self.state.quality_score = score
        
        if score >= 0.85:
            return "approved"
        elif self.state.revision_count >= 2:
            return "force_publish"
        else:
            return "needs_revision"
    
    @listen("needs_revision")
    def revise(self, event):
        """修改循环"""
        self.state.revision_count += 1
        return self.content_creation(self.state.research_report)
    
    @listen(or_("approved", "force_publish"))
    def publish(self, event):
        """发布"""
        publish_crew = Crew(
            agents=[Agent(role="Publisher", ...)],
            tasks=[Task(description="执行发布流程", ...)],
        )
        publish_crew.kickoff()
        print("Product launched!")

# 启动整个流程
flow = ProductLaunchFlow()
flow.kickoff(inputs={"product_name": "AI Writer Pro"})

class ArticleFlow(Flow):
    @start()
    def research(self):
        crew = Crew(agents=[researcher], tasks=[research_task])
        return crew.kickoff().raw
    
    @listen(research)
    def write(self, data):
        crew = Crew(agents=[writer], tasks=[write_task])
        return crew.kickoff(inputs={"research": data}).raw
    
    @router(write)
    def review(self, draft):
        return "publish" if quality_ok(draft) else "revise"
    
    @listen("revise")
    def do_revise(self, _):
        return self.write(self.state.research)
    
    @listen("publish")
    def publish(self, _):
        print("Done!")

graph = StateGraph(ArticleState)
graph.add_node("research", research_node)
graph.add_node("write", write_node)
graph.add_node("review", review_node)
graph.add_node("publish", publish_node)
graph.add_edge("research", "write")
graph.add_edge("write", "review")
graph.add_conditional_edges("review", review_route, {
    "pass": "publish", "revise": "write"
})
graph.set_entry_point("research")
graph.set_finish_point("publish")
app = graph.compile()

from crewai.memory import ShortTermMemory, LongTermMemory, EntityMemory

crew = Crew(
    agents=[...],
    tasks=[...],
    memory=True,  # 启用记忆系统
    
    # 可自定义记忆存储
    short_term_memory=ShortTermMemory(),   # 当前会话记忆
    long_term_memory=LongTermMemory(),     # 跨会话持久记忆
    entity_memory=EntityMemory(),          # 实体知识图谱
)

from crewai.knowledge.source import (
    PDFKnowledgeSource,
    TextFileKnowledgeSource,
    CSVKnowledgeSource,
    JSONKnowledgeSource,
)

researcher = Agent(
    role="Research Analyst",
    knowledge_sources=[
        PDFKnowledgeSource(
            file_paths=["papers/agent_survey_2025.pdf"],
            chunk_size=1000,
            chunk_overlap=200,
        ),
        TextFileKnowledgeSource(
            file_paths=["docs/internal_guidelines.md"],
        ),
    ],
)

from pydantic import BaseModel

class CompetitorReport(BaseModel):
    company_name: str
    strengths: list[str]
    weaknesses: list[str]
    market_share: float
    recommendation: str

task = Task(
    description="分析竞品 X 的市场表现",
    expected_output="结构化的竞品分析报告",
    output_pydantic=CompetitorReport,  # 强制输出为此结构
    agent=analyst,
)

# result.pydantic 就是一个 CompetitorReport 实例
result = crew.kickoff()
report: CompetitorReport = result.pydantic
print(report.company_name)  # 类型安全访问

crew = Crew(
    agents=[...],
    tasks=[...],
    planning=True,           # 启用自动规划
    planning_llm=LLM(model="gpt-4"),  # 规划用的 LLM
)

from crewai import Agent, Task, Crew, Process, LLM
from crewai.flow.flow import Flow, listen, start, router
from crewai.tools import SerperDevTool
from pydantic import BaseModel

# ===== 状态定义 =====
class BlogState(BaseModel):
    topic: str = ""
    research_data: str = ""
    draft: str = ""
    review_feedback: str = ""
    seo_optimized: str = ""
    quality_score: float = 0.0
    revision_count: int = 0
    published: bool = False

# ===== Agent 定义 =====
researcher = Agent(
    role="Senior Tech Researcher",
    goal="Find comprehensive, accurate, and up-to-date information on technical topics",
    backstory="""You are a veteran technology journalist with 10+ years of experience 
    covering AI/ML, distributed systems, and developer tools. You have a PhD in Computer 
    Science and maintain active connections with researchers at top labs. Your research 
    methodology: start with primary sources (papers, official docs), cross-reference 
    with multiple sources, and always note the recency and reliability of each source.""",
    tools=[SerperDevTool()],
    llm=LLM(model="gpt-4"),
    max_iter=10,
)

writer = Agent(
    role="Technical Content Writer",
    goal="Produce clear, engaging, and technically accurate blog posts",
    backstory="""You are a technical writer who combines deep engineering knowledge 
    with excellent storytelling ability. You've written for major tech publications 
    and your posts consistently rank in the top 10% for reader engagement. Your 
    writing philosophy: lead with the 'why', use concrete examples before abstract 
    definitions, and always include runnable code snippets. You target an audience 
    of intermediate developers (2-5 years experience).""",
    llm=LLM(model="gpt-4"),
    max_iter=5,
)

reviewer = Agent(
    role="Technical Editor & Fact-Checker",
    goal="Ensure technical accuracy, logical flow, and readability",
    backstory="""You are a senior technical editor at a top-tier engineering blog. 
    You've caught hundreds of subtle technical errors that would have embarrassed 
    authors. Your review checklist: factual accuracy, code correctness, logical flow, 
    jargon explanation, missing context, and readability score. You give constructive 
    feedback with specific suggestions, not vague complaints.""",
    llm=LLM(model="gpt-4"),
    max_iter=3,
)

seo_specialist = Agent(
    role="Technical SEO Specialist",
    goal="Optimize content for search engines without sacrificing technical quality",
    backstory="""You specialize in SEO for technical content. You understand that 
    developer audiences hate keyword-stuffed content, so your approach is subtle: 
    optimize headings, meta descriptions, and internal linking while preserving 
    the author's voice. You focus on semantic SEO and topic clustering.""",
    llm=LLM(model="gpt-4"),
    max_iter=3,
)

# ===== Flow 定义 =====
class BlogProductionFlow(Flow[BlogState]):
    
    @start()
    def kickoff_research(self):
        """阶段 1: 深度研究"""
        research_task = Task(
            description=f"""Research the topic: '{self.state.topic}'
            Gather: key concepts, recent developments (2024-2025), 
            code examples, expert opinions, and common misconceptions.
            Minimum 5 distinct sources required.""",
            expected_output="""A structured research brief with:
            - Executive summary (3 sentences)
            - Key findings (5-8 bullet points with source citations)
            - Code examples or architecture diagrams
            - Contrarian viewpoints or limitations
            - Suggested article structure""",
            agent=researcher,
        )
        
        crew = Crew(agents=[researcher], tasks=[research_task], verbose=True)
        result = crew.kickoff()
        self.state.research_data = result.raw
        return result.raw
    
    @listen(kickoff_research)
    def write_article(self, research_data):
        """阶段 2: 撰写文章"""
        write_task = Task(
            description=f"""Write a technical blog post about '{self.state.topic}'.
            Use the following research: {research_data}
            
            Requirements:
            - 2000-3000 words
            - Start with a compelling hook
            - Include at least 2 code examples
            - End with actionable takeaways
            - Target audience: intermediate developers""",
            expected_output="A complete, publication-ready blog post in markdown format",
            agent=writer,
        )
        
        crew = Crew(agents=[writer], tasks=[write_task], verbose=True)
        result = crew.kickoff()
        self.state.draft = result.raw
        return result.raw
    
    @listen(write_article)
    def technical_review(self, draft):
        """阶段 3: 技术审核"""
        review_task = Task(
            description=f"""Review this technical article for accuracy and quality:
            
            {draft}
            
            Check for: factual errors, code bugs, logical gaps, 
            unclear explanations, missing context.
            
            Rate overall quality 0-1 and provide specific feedback.""",
            expected_output="""Review report with:
            - Quality score (0-1)
            - List of issues found (if any)
            - Specific improvement suggestions
            - Final verdict: APPROVE or NEEDS_REVISION""",
            agent=reviewer,
        )
        
        crew = Crew(agents=[reviewer], tasks=[review_task], verbose=True)
        result = crew.kickoff()
        
        # 解析审核结果
        self.state.review_feedback = result.raw
        self.state.quality_score = self._extract_score(result.raw)
        return result.raw
    
    @router(technical_review)
    def review_decision(self, review_result):
        """路由：根据审核结果决定下一步"""
        if self.state.quality_score >= 0.8:
            return "approved"
        elif self.state.revision_count >= 2:
            return "force_approve"  # 防止无限循环
        else:
            return "needs_revision"
    
    @listen("needs_revision")
    def revise_article(self, _):
        """修改文章（循环）"""
        self.state.revision_count += 1
        
        revision_task = Task(
            description=f"""Revise this article based on reviewer feedback:
            
            Original article: {self.state.draft}
            Reviewer feedback: {self.state.review_feedback}
            
            Address all issues raised by the reviewer.""",
            expected_output="A revised article addressing all feedback points",
            agent=writer,
        )
        
        crew = Crew(agents=[writer], tasks=[revision_task])
        result = crew.kickoff()
        self.state.draft = result.raw
        
        # 重新送审
        return self.technical_review(result.raw)
    
    @listen(or_("approved", "force_approve"))
    def seo_optimization(self, _):
        """阶段 4: SEO 优化"""
        seo_task = Task(
            description=f"""Optimize this article for SEO:
            
            {self.state.draft}
            
            Focus on: heading optimization, meta description, 
            internal linking suggestions, keyword placement.""",
            expected_output="SEO-optimized version of the article with meta tags",
            agent=seo_specialist,
        )
        
        crew = Crew(agents=[seo_specialist], tasks=[seo_task])
        result = crew.kickoff()
        self.state.seo_optimized = result.raw
        return result.raw
    
    @listen(seo_optimization)
    def publish(self, final_article):
        """阶段 5: 发布"""
        self.state.published = True
        print(f"\n{'='*60}")
        print(f"Article Published!")
        print(f"Topic: {self.state.topic}")
        print(f"Revisions: {self.state.revision_count}")
        print(f"Quality Score: {self.state.quality_score}")
        print(f"{'='*60}\n")
        return final_article
    
    def _extract_score(self, review_text: str) -> float:
        """从审核文本中提取分数"""
        import re
        match = re.search(r'(\d+\.?\d*)\s*/\s*1|score[:\s]*(\d+\.?\d*)', 
                         review_text.lower())
        if match:
            score = float(match.group(1) or match.group(2))
            return min(score, 1.0)
        return 0.5  # 默认中等分数

# ===== 运行 =====
if __name__ == "__main__":
    flow = BlogProductionFlow()
    result = flow.kickoff(inputs={"topic": "Building AI Agents with LangGraph in 2025"})
    print(f"\nFinal article length: {len(flow.state.seo_optimized)} chars")