Agent 开发框架（二）CrewAI

一、CrewAI 开发框架简介1、简介CrewAI 是一个开源的多智能体框架支持工具集成用于构建一组协作完成任务的智能体Agents。它的设计理念是每个 Agent 都具备特定角色、工具和目标通过任务分工与信息共享共同解决复杂问题。CrewAI 能够提供类人团队的任务执行方式不但可以构建单智能体助手更可以构建“项目经理 设计师 执行者”这样的模拟团队。CrewAI 的另一个优势是可扩展性和生产级部署因此其实可以广泛应用于你的数据分析、内容创作、自动化流程等场景。demo1、简单demofrom crewai import Agent, Crew, Task, LLM import os # 你的 gateway key os.environ[OPENAI_API_KEY] ****** llm LLM( modelgpt-5.1, base_urlhttps://llm-gatewa.xxx.xxx/v1, api_keyos.getenv(OPENAI_API_KEY), ) researcher Agent( role研究员, goal搜索并总结最新 AI 趋势, backstory你是一位热衷于探索 AI 技术的专家, llmllm, verboseTrue ) task Task( description查找并总结 2025 年 AI 领域的最新发展, agentresearcher, expected_output一份简短的 AI 趋势总结 ) crew Crew( agents[researcher], tasks[task], verboseTrue ) result crew.kickoff() print(result)输出......省略很长的一段.2、单机版import base64 import logging import os from io import BytesIO from uuid import uuid4 import requests from PIL import Image from crewai import Agent, Crew, Task from crewai.process import Process from crewai.tools import tool from crewai.llms.base_llm import BaseLLM from pydantic import BaseModel # 配置 os.environ[LLM_GATEWAY_API_KEY] ***** logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) # 自定义 LLM class MyLLM(BaseLLM): def call(self, messages, **kwargs): url https://llm-gatewayxxx.xxx.xxx/v1/chat/completions headers { Content-Type: application/json, X-Api-Key: os.getenv(LLM_GATEWAY_API_KEY), } data { model: gpt-5.1, messages: messages, } res requests.post(url, headersheaders, jsondata, timeout120) res.raise_for_status() return res.json()[choices][0][message][content] def get_context_window_size(self): return 128000 # 数据结构 class ImageData(BaseModel): id: str | None None name: str | None None mime_type: str | None None bytes: str | None None error: str | None None class SimpleImageCache: def __init__(self): self._cache {} def get(self, session_id: str): return self._cache.get(session_id, {}) def set(self, session_id: str, data: dict): self._cache[session_id] data image_cache SimpleImageCache() # Tool tool(ImageGenerationTool) def generate_image_tool(prompt: str, session_id: str, artifact_file_id: str ) - str: 根据提示词生成图像并返回图像ID if not prompt: raise ValueError(提示词不能为空) try: client genai.Client() contents prompt response client.models.generate_content( modelgemini-2.0-flash-exp, contentscontents, configtypes.GenerateContentConfig( response_modalities[Text, Image] ), ) for part in response.candidates[0].content.parts: if part.inline_data is not None: image_data ImageData( bytesbase64.b64encode(part.inline_data.data).decode(utf-8), mime_typepart.inline_data.mime_type, namegenerated_image.png, iduuid4().hex, ) session_data image_cache.get(session_id) or {} session_data[image_data.id] image_data image_cache.set(session_id, session_data) return image_data.id return 生成失败 except Exception as e: return f错误: {str(e)} # Agent class SimpleCrewAIAgent: def __init__(self): self.image_creator_agent Agent( role图像创作专家, goal根据提示词生成图像, backstory你是一个AI艺术家, verboseTrue, allow_delegationFalse, tools[generate_image_tool], llmMyLLM(modelgpt-5.1), ) self.image_creation_task Task( description( 用户提示词{user_prompt}\n 必须调用 ImageGenerationTool 工具生成图像\n 传入参数prompt{user_prompt}, session_id{session_id}, artifact_file_id{artifact_file_id} ), expected_output图像ID, agentself.image_creator_agent, ) self.image_crew Crew( agents[self.image_creator_agent], tasks[self.image_creation_task], processProcess.sequential, verboseTrue, ) def generate_image(self, prompt: str, session_id: str None) - str: if not session_id: session_id uuid4().hex inputs { user_prompt: prompt, session_id: session_id, artifact_file_id: , } result self.image_crew.kickoff(inputs) return str(result.raw) if hasattr(result, raw) else str(result) # main def main(): agent SimpleCrewAIAgent() prompt 一只赛博朋克风格的猫霓虹灯背景 session_id test_123 result agent.generate_image(prompt, session_id) print(\n生成结果:, result) if __name__ __main__: main()解读这段代码看起来很长但并不难理解它的内核是主要 Agent 类 (SimpleCrewAIAgent)这个类中的关键元素包括LLM 模型图像创作 Agent专门负责图像生成的 AI 角色。任务定义描述如何处理用户提示词。Crew协调 Agent 和任务的执行。此外ImageData 复杂创建图像的数据模型 而 SimpleImageCache 类则创建了一个简单的内存缓存系统复制存储生成的图像数据但程序重启后数据会丢失优点是快速访问不占用磁盘空间。在聊天机器人的场景中有这个缓存机制就足够了。如有存盘需要可以通过 Agent 的 save_image_to_file 功能保存它。输出3、a2a版self.image_creator_agent Agent( roleImage Creation Expert, goalGenerate an image based on the user\s text prompt..., backstoryYou are a digital artist powered by AI..., tools[generate_image_tool], llmself.model, ) self.image_creation_task Task( descriptionReceive a user prompt: {user_prompt}..., expected_outputThe id of the generated image, agentself.image_creator_agent, ) 有了 Agent 和 Task就可以通过 Crew 把 Agent 组装起来 self.image_crew Crew( agents[self.image_creator_agent], tasks[self.image_creation_task], processProcess.sequential, verboseFalse, )