CrewAI智能体开发：CrewAI 运行自动化工具

使 CrewAI 代理能够在您的工作流中调用 CrewAI 平台自动化并利用外部团队服务。

InvokeCrewAIAutomationTool

InvokeCrewAIAutomationTool提供 CrewAI 平台 API 与外部团队服务的集成。此工具允许您在 CrewAI 代理中调用并与 CrewAI 平台自动化进行交互，从而实现不同团队工作流之间的无缝集成。

安装

uv pip install 'crewai[tools]'

要求

• CrewAI 平台 API 访问
• 用于身份验证的有效 Bearer 令牌
• 对 CrewAI 平台自动化端点的网络访问

用法以下是如何在 CrewAI 智能体中使用此工具

from crewai import Agent, Task, Crew from crewai_tools import InvokeCrewAIAutomationTool # Initialize the tool automation_tool = InvokeCrewAIAutomationTool( crew_api_url="https://data-analysis-crew-[...].crewai.com", crew_bearer_token="your_bearer_token_here", crew_name="Data Analysis Crew", crew_description="Analyzes data and generates insights" ) # Create a CrewAI agent that uses the tool automation_coordinator = Agent( role='Automation Coordinator', goal='Coordinate and execute automated crew tasks', backstory='I am an expert at leveraging automation tools to execute complex workflows.', tools=[automation_tool], verbose=True ) # Create a task for the agent analysis_task = Task( description="Execute data analysis automation and provide insights", agent=automation_coordinator, expected_output="Comprehensive data analysis report" ) # Create a crew with the agent crew = Crew( agents=[automation_coordinator], tasks=[analysis_task], verbose=2 ) # Run the crew result = crew.kickoff() print(result)

工具参数

环境变量

CREWAI_API_URL=https://your-crew-automation.crewai.com # Alternative to passing crew_api_url CREWAI_BEARER_TOKEN=your_bearer_token_here # Alternative to passing crew_bearer_token

高级用法

带动态参数的自定义输入模式

from crewai import Agent, Task, Crew from crewai_tools import InvokeCrewAIAutomationTool from pydantic import Field # Define custom input schema custom_inputs = { "year": Field(..., description="Year to retrieve the report for (integer)"), "region": Field(default="global", description="Geographic region for analysis"), "format": Field(default="summary", description="Report format (summary, detailed, raw)") } # Create tool with custom inputs market_research_tool = InvokeCrewAIAutomationTool( crew_api_url="https://state-of-ai-report-crew-[...].crewai.com", crew_bearer_token="your_bearer_token_here", crew_name="State of AI Report", crew_description="Retrieves a comprehensive report on state of AI for a given year and region", crew_inputs=custom_inputs, max_polling_time=15 * 60 # 15 minutes timeout ) # Create an agent with the tool research_agent = Agent( role="Research Coordinator", goal="Coordinate and execute market research tasks", backstory="You are an expert at coordinating research tasks and leveraging automation tools.", tools=[market_research_tool], verbose=True ) # Create and execute a task with custom parameters research_task = Task( description="Conduct market research on AI tools market for 2024 in North America with detailed format", agent=research_agent, expected_output="Comprehensive market research report" ) crew = Crew( agents=[research_agent], tasks=[research_task] ) result = crew.kickoff()

多阶段自动化工作流

from crewai import Agent, Task, Crew, Process from crewai_tools import InvokeCrewAIAutomationTool # Initialize different automation tools data_collection_tool = InvokeCrewAIAutomationTool( crew_api_url="https://data-collection-crew-[...].crewai.com", crew_bearer_token="your_bearer_token_here", crew_name="Data Collection Automation", crew_description="Collects and preprocesses raw data" ) analysis_tool = InvokeCrewAIAutomationTool( crew_api_url="https://analysis-crew-[...].crewai.com", crew_bearer_token="your_bearer_token_here", crew_name="Analysis Automation", crew_description="Performs advanced data analysis and modeling" ) reporting_tool = InvokeCrewAIAutomationTool( crew_api_url="https://reporting-crew-[...].crewai.com", crew_bearer_token="your_bearer_token_here", crew_name="Reporting Automation", crew_description="Generates comprehensive reports and visualizations" ) # Create specialized agents data_collector = Agent( role='Data Collection Specialist', goal='Gather and preprocess data from various sources', backstory='I specialize in collecting and cleaning data from multiple sources.', tools=[data_collection_tool] ) data_analyst = Agent( role='Data Analysis Expert', goal='Perform advanced analysis on collected data', backstory='I am an expert in statistical analysis and machine learning.', tools=[analysis_tool] ) report_generator = Agent( role='Report Generation Specialist', goal='Create comprehensive reports and visualizations', backstory='I excel at creating clear, actionable reports from complex data.', tools=[reporting_tool] ) # Create sequential tasks collection_task = Task( description="Collect market data for Q4 2024 analysis", agent=data_collector ) analysis_task = Task( description="Analyze collected data to identify trends and patterns", agent=data_analyst ) reporting_task = Task( description="Generate executive summary report with key insights and recommendations", agent=report_generator ) # Create a crew with sequential processing crew = Crew( agents=[data_collector, data_analyst, report_generator], tasks=[collection_task, analysis_task, reporting_task], process=Process.sequential, verbose=2 ) result = crew.kickoff()

用例

分布式团队编排

• 协调多个专业团队自动化以处理复杂的、多阶段的工作流
• 实现不同自动化服务之间的无缝交接，以实现全面的任务执行
• 通过将工作负载分布到多个 CrewAI 平台自动化中来扩展处理能力

跨平台集成

• 连接 CrewAI 代理与 CrewAI 平台自动化，实现混合本地-云工作流
• 利用专业自动化，同时保持本地控制和编排
• 实现本地代理与基于云的自动化服务之间的安全协作

企业自动化管道

• 创建结合本地智能和云处理能力的企业级自动化管道
• 实施跨多个自动化服务的复杂业务工作流
• 为数据分析、报告和决策制定实现可扩展、可重复的流程

动态工作流组合

• 根据任务要求通过链接不同的自动化服务来动态组合工作流
• 实现自适应处理，其中自动化选择取决于数据特征或业务规则
• 创建灵活、可重用的自动化组件，可以通过多种方式组合

专业领域处理

• 从通用代理访问特定领域的自动化（财务分析、法律研究、技术文档）
• 利用预构建的专业团队自动化，而无需重新构建复杂的领域逻辑
• 使代理能够通过目标自动化服务访问专家级功能

自定义输入模式定义crew_inputs时，使用 Pydantic Field 对象来指定输入参数

from pydantic import Field crew_inputs = { "required_param": Field(..., description="This parameter is required"), "optional_param": Field(default="default_value", description="This parameter is optional"), "typed_param": Field(..., description="Integer parameter", ge=1, le=100) # With validation }

错误处理该工具为常见场景提供全面的错误处理

• API 连接错误：与 CrewAI 平台的网络连接问题
• 身份验证错误：无效或过期的 Bearer 令牌
• 超时错误：任务超出最大轮询时间
• 任务失败：Crew 自动化在执行过程中失败
• 输入验证错误：向自动化端点传递了无效参数

API 端点该工具与两个主要的 API 端点交互

• POST {crew_api_url}/kickoff：启动一个新的团队自动化任务
• GET {crew_api_url}/status/{crew_id}：检查正在运行任务的状态

注意

• 该工具会自动每秒轮询状态端点，直到完成或超时
• 成功的任务直接返回结果，而失败的任务返回错误信息
• Bearer 令牌应妥善保管，不要在生产环境中硬编码
• 考虑使用环境变量来存储敏感配置，例如 Bearer 令牌
• 自定义输入模式必须与目标团队自动化所需的参数兼容

《AI提示工程必知必会》为读者提供了丰富的AI提示工程知识与实战技能。《AI提示工程必知必会》主要内容包括各类提示词的应用，如问答式、指令式、状态类、建议式、安全类和感谢类提示词，以及如何通过实战演练掌握提示词的使用技巧；使用提示词进行文本摘要、改写重述、语法纠错、机器翻译等语言处理任务，以及在数据挖掘、程序开发等领域的应用；AI在绘画创作上的应用，百度文心一言和阿里通义大模型这两大智能平台的特性与功能，以及市场调研中提示词的实战应用。通过阅读《AI提示工程必知必会》，读者可掌握如何有效利用AI提示工程提升工作效率，创新工作流程，并在职场中脱颖而出。