# 创建虚拟环境（推荐）python -m venv crewai-envsource crewai-env/bin/activate  # Linux/Mac# crewai-env\Scripts\activate   # Windows# 安装 CrewAIpip install crewai crewai-tools# 配置 API Keyexport OPENAI_API_KEY="sk-your-api-key-here"

from crewai import Agent, Task, Crew, Process# 定义 Agent：研究员researcher = Agent(    role="行业研究员",    goal="搜集并整理 AI 行业最新趋势和关键数据",    backstory="你是一位资深行业分析师，擅长从多个来源搜集数据并提炼洞察",    verbose=True,    allow_delegation=False)# 定义 Agent：撰稿人writer = Agent(    role="技术撰稿人",    goal="将研究成果写成一篇清晰易懂的中文分析文章",    backstory="你是一位优秀的科技作者，擅长把复杂技术话题讲得通俗有趣",    verbose=True)# 定义任务：研究任务research_task = Task(    description="调研 2026 年 AI Agent 领域的三大技术趋势，列出关键数据和代表项目",    expected_output="一份包含三大趋势、数据点和代表项目的中文研究摘要",    agent=researcher)# 定义任务：写作任务（依赖研究任务输出）write_task = Task(    description="根据研究摘要，写一篇 800 字的中文分析文章，面向技术从业者",    expected_output="一篇结构完整、有观点、有数据的中文分析文章",    agent=writer,    context=[research_task])# 组建 Crew 并执行crew = Crew(    agents=[researcher, writer],    tasks=[research_task, write_task],    process=Process.sequential,    verbose=True)result = crew.kickoff()print(result)

from langchain_ollama import ChatOllama# 定义本地 LLMlocal_llm = ChatOllama(    model="qwen2.5:7b",    base_url="http://localhost:11434")# Agent 中指定 llmresearcher = Agent(    role="行业研究员",    goal="搜集并整理 AI 行业最新趋势",    backstory="你是资深行业分析师",    llm=local_llm)

from crewai import Agent, Task, Crew, Processfrom crewai_tools import SerperDevTool, WebsiteSearchTool# 工具初始化search_tool = SerperDevTool()web_tool = WebsiteSearchTool()# Agent 1：竞品搜索专家research_agent = Agent(    role="竞品搜索专家",    goal="搜集目标 AI 产品的最新信息、定价策略和用户评价",    backstory="你擅长网络信息检索，能快速从多个来源定位关键数据",    tools=[search_tool, web_tool],    verbose=True)# Agent 2：竞品分析师analyst_agent = Agent(    role="竞品分析师",    goal="对比分析竞品的功能差异、定价策略和市场定位",    backstory="你是经验丰富的产品分析师，擅长从数据中提炼战略洞察",    verbose=True)# Agent 3：报告撰写人report_agent = Agent(    role="市场报告撰写人",    goal="将分析结果整理成结构清晰、数据详实的中文调研报告",    backstory="你是专业的商业报告撰稿人，逻辑严谨、表达简洁",    verbose=True)# 任务 1：搜集竞品信息research_task = Task(    description="调研 Notion AI、飞书智能伙伴、钉钉 AI 助理三款产品的最新功能、定价和用户评价",    expected_output="一份包含三款产品功能清单、定价对比表和主要优缺点的中文摘要",    agent=research_agent,    tools=[search_tool])# 任务 2：分析对比analysis_task = Task(    description="基于搜集到的信息，分析三款产品在功能、价格、目标用户上的差异化定位",    expected_output="一份结构化的竞品对比分析，包含优势矩阵和差异化洞察",    agent=analyst_agent,    context=[research_task])# 任务 3：撰写报告report_task = Task(    description="将分析结果写成一份完整的市场调研报告，包含执行摘要、竞品对比表、趋势判断",    expected_output="一份 1500 字左右的中文市场调研报告 Markdown 文档",    agent=report_agent,    context=[analysis_task])# 组建团队crew = Crew(    agents=[research_agent, analyst_agent, report_agent],    tasks=[research_task, analysis_task, report_task],    process=Process.sequential,    verbose=True)# 执行result = crew.kickoff()print(result)

from crewai import Agent, Task, Crew, Flowfrom crewai.flow.flow import start, listen, routerfrom pydantic import BaseModel# 定义 Flow 状态classDevFlowState(BaseModel):    requirements: str = ""    code: str = ""    review_result: str = ""    approved: bool = False# Agent 定义pm_agent = Agent(    role="产品经理",    goal="将用户需求拆解为清晰的技术开发任务",    backstory="你是严谨的技术产品经理，擅长把模糊需求转化为明确的开发规格",    verbose=True)dev_agent = Agent(    role="Python 开发工程师",    goal="根据开发规格编写高质量、可运行的 Python 代码",    backstory="你是资深 Python 开发，代码规范、注释完整、注重边界处理",    verbose=True)reviewer_agent = Agent(    role="代码审查员",    goal="审查代码的正确性、安全性和规范性，给出通过或打回结论",    backstory="你是严格的代码审查员，关注边界条件、错误处理和最佳实践",    verbose=True)classDevFlow(Flow[DevFlowState]):    @start()defparse_requirements(self):"""步骤1：解析需求"""        task = Task(            description=f"将以下需求拆解为开发规格：{self.state.requirements}",            expected_output="一份结构化的开发规格，包含功能点、输入输出和边界条件",            agent=pm_agent        )        crew = Crew(agents=[pm_agent], tasks=[task])        spec = crew.kickoff()return {"requirements": spec}    @listen(parse_requirements)defwrite_code(self):"""步骤2：编写代码"""        task = Task(            description=f"根据开发规格编写 Python 代码：{self.state.requirements}",            expected_output="完整的可运行 Python 代码，包含注释和错误处理",            agent=dev_agent        )        crew = Crew(agents=[dev_agent], tasks=[task])        code = crew.kickoff()return {"code": code}    @listen(write_code)defreview_code(self):"""步骤3：代码审查"""        task = Task(            description=f"审查以下 Python 代码的质量：{self.state.code}",            expected_output='JSON 格式：{"approved": true/false, "issues": "问题描述"}',            agent=reviewer_agent        )        crew = Crew(agents=[reviewer_agent], tasks=[task])        result = crew.kickoff()import json        parsed = json.loads(result)return {"review_result": result, "approved": parsed["approved"]}    @router(review_code)defquality_gate(self):"""质量门禁：通过则交付，否则打回重写"""ifself.state.approved:return"deliver"return"rewrite"    @listen("deliver")defdeliver_code(self):print("代码审查通过，交付最终版本：")print(self.state.code)    @listen("rewrite")defrewrite_code(self):print("代码审查未通过，打回开发团队重写...")# 运行流程flow = DevFlow()flow.state.requirements = "写一个 Python 函数，输入数字列表，返回其中的质数"flow.kickoff()