登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 业界区 › 安全 ›A2A + CrewAI + OpenRouter 图表生成 Agent 教程

A2A + CrewAI + OpenRouter 图表生成 Agent 教程

班闵雨 2025-6-28 06:09:10
教程重点

本教程将指导您完成以下核心技能的实践：

  • 集成 OpenRouter + CrewAI + A2A：完成端到端的 Agent 开发，使用 OpenRouter 作为 LLM 提供商，CrewAI 作为 Agent 框架，A2A 协议作为标准化接口
  • 实践 A2A Agent 返回 Image Data：学习如何让 Agent 生成并返回图像数据，而不仅仅是文本回复
  • 使用 A2A Inspector 调试 A2A 应用：掌握专业的调试工具来测试和验证您的 Agent 应用
快速开始

1. 克隆代码
  1. git clone git@github.com:a2aproject/a2a-samples.git
  2. cd a2a-samples/samples/python/agents/analytics
  4. # OR
  5. git clone git@github.com:sing1ee/a2a-crewai-charts-agent.git
  6. cd a2a-crewai-charts-agent
复制代码
2. 创建环境配置

创建 .env 文件：
  1. OPENROUTER_API_KEY=sk-or-v1-your-api-key-here
  2. OPENAI_MODEL_NAME=openrouter/anthropic/claude-3.7-sonnet
复制代码
3. 环境设置和运行
  1. # 创建虚拟环境
  2. uv venv
  4. # 激活虚拟环境
  5. source .venv/bin/activate
  7. # 运行应用
  8. uv run .
复制代码
应用将在 http://localhost:10011 启动。
使用 A2A Inspector 调试

A2A Inspector 是专门为调试 A2A 应用设计的强大工具。
调试步骤：


  • 访问 A2A Inspector：打开 https://inspector.a2aprotocol.ai
  • 连接到您的 Agent：

    • 在 Inspector 中输入您的 Agent 地址：http://localhost:10011
    • 点击 "Connect" 建立连接

  • 测试 Agent 功能：

    • 发送测试消息："Generate a chart of revenue: Jan,1000 Feb,2000 Mar,1500"
    • 观察 A2A 协议的完整交互过程
    • 查看返回的图像数据

  • 调试和监控：

    • 检查 Agent 的 capabilities 和 skills
    • 监控请求和响应的完整流程
    • 验证图像数据的正确传输

参考 A2A Inspector 文档 获取更详细的调试指南。
主要流程和代码介绍

系统架构时序图

sequenceDiagram    participant U as User    participant A2A as A2A Inspector    participant S as A2A Server    participant H as DefaultRequestHandler    participant E as ChartGenerationAgentExecutor    participant CA as ChartGenerationAgent    participant Crew as CrewAI Crew    participant Tool as ChartGenerationTool    participant MP as Matplotlib    participant Cache as InMemoryCache    U->>A2A: Send prompt "Generate chart: A,100 B,200"    A2A->>S: HTTP POST /tasks with A2A message    S->>H: Handle request with RequestContext    H->>E: execute(context, event_queue)    E->>CA: invoke(query, session_id)    CA->>Crew: kickoff with inputs    Crew->>Tool: generate_chart_tool(prompt, session_id)    Tool->>Tool: Parse CSV data    Tool->>MP: Create bar chart with matplotlib    MP-->>Tool: Return PNG bytes    Tool->>Cache: Store image with ID    Cache-->>Tool: Confirm storage    Tool-->>Crew: Return image ID    Crew-->>CA: Return image ID    CA-->>E: Return image ID    E->>CA: get_image_data(session_id, image_key)    CA->>Cache: Retrieve image data    Cache-->>CA: Return Imagedata    CA-->>E: Return Imagedata    E->>H: Create FilePart with image bytes    H->>S: enqueue completed_task event    S-->>A2A: Return A2A response with image    A2A-->>U: Display generated chart核心组件详解

1. A2A 服务器初始化 (__main__.py)
  1. # 定义 Agent 能力和技能
  2. capabilities = AgentCapabilities(streaming=False)
  3. skill = AgentSkill(
  4.     id='chart_generator',
  5.     name='Chart Generator',
  6.     description='Generate a chart based on CSV-like data passed in',
  7.     tags=['generate image', 'edit image'],
  8.     examples=['Generate a chart of revenue: Jan,$1000 Feb,$2000 Mar,$1500'],
  9. )
  11. # 创建 Agent 卡片
  12. agent_card = AgentCard(
  13.     name='Chart Generator Agent',
  14.     description='Generate charts from structured CSV-like data input.',
  15.     url=f'http://{host}:{port}/',
  16.     version='1.0.0',
  17.     defaultInputModes=ChartGenerationAgent.SUPPORTED_CONTENT_TYPES,
  18.     defaultOutputModes=ChartGenerationAgent.SUPPORTED_CONTENT_TYPES,
  19.     capabilities=capabilities,
  20.     skills=[skill],
  21. )
复制代码
关键点：

  • AgentCapabilities 定义了 Agent 支持的功能（这里禁用了流式传输）
  • AgentSkill 描述了 Agent 的具体技能和使用示例
  • AgentCard 是 A2A 协议中 Agent 的身份标识
2. CrewAI Agent 实现 (agent.py)
  1. class ChartGenerationAgent:
  2.     def __init__(self):
  3.         # 创建专门的图表生成 Agent
  4.         self.chart_creator_agent = Agent(
  5.             role='Chart Creation Expert',
  6.             goal='Generate a bar chart image based on structured CSV input.',
  7.             backstory='You are a data visualization expert who transforms structured data into visual charts.',
  8.             verbose=False,
  9.             allow_delegation=False,
  10.             tools=[generate_chart_tool],
  11.         )
  13.         # 定义任务
  14.         self.chart_creation_task = Task(
  15.             description=(
  16.                 "You are given a prompt: '{user_prompt}'.\n"
  17.                 "If the prompt includes comma-separated key:value pairs (e.g. 'a:100, b:200'), "
  18.                 "reformat it into CSV with header 'Category,Value'.\n"
  19.                 "Ensure it becomes two-column CSV, then pass that to the 'ChartGenerationTool'.\n"
  20.                 "Use session ID: '{session_id}' when calling the tool."
  21.             ),
  22.             expected_output='The id of the generated chart image',
  23.             agent=self.chart_creator_agent,
  24.         )
复制代码
关键点：

  • CrewAI 的 Agent 类定义了 AI 助手的角色和能力
  • Task 类描述了具体的任务执行逻辑
  • 通过 tools 参数将自定义工具集成到 Agent 中
3. 图表生成工具
  1. @tool('ChartGenerationTool')
  2. def generate_chart_tool(prompt: str, session_id: str) -> str:
  3.     """Generates a bar chart image from CSV-like input using matplotlib."""
  4.    
  5.     # 解析 CSV 数据
  6.     df = pd.read_csv(StringIO(prompt))
  7.     df.columns = ['Category', 'Value']
  8.     df['Value'] = pd.to_numeric(df['Value'], errors='coerce')
  9.    
  10.     # 生成柱状图
  11.     fig, ax = plt.subplots()
  12.     ax.bar(df['Category'], df['Value'])
  13.     ax.set_xlabel('Category')
  14.     ax.set_ylabel('Value')
  15.     ax.set_title('Bar Chart')
  16.    
  17.     # 保存为 PNG 字节
  18.     buf = BytesIO()
  19.     plt.savefig(buf, format='png')
  20.     plt.close(fig)
  21.     buf.seek(0)
  22.     image_bytes = buf.read()
  23.    
  24.     # 编码并缓存图片
  25.     data = Imagedata(
  26.         bytes=base64.b64encode(image_bytes).decode('utf-8'),
  27.         mime_type='image/png',
  28.         name='generated_chart.png',
  29.         id=uuid4().hex,
  30.     )
  31.    
  32.     # 将图片存储到缓存中
  33.     session_data = cache.get(session_id) or {}
  34.     session_data[data.id] = data
  35.     cache.set(session_id, session_data)
  36.    
  37.     return data.id
复制代码
关键点：

  • 使用 @tool 装饰器将函数转换为 CrewAI 工具
  • 使用 pandas 解析 CSV 数据，matplotlib 生成图表
  • 图像以 base64 编码存储，便于网络传输
  • 使用会话ID管理多个用户的数据隔离
4. A2A 执行器 (agent_executor.py)
  1. class ChartGenerationAgentExecutor(AgentExecutor):
  2.     async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
  3.         # 获取用户输入
  4.         query = context.get_user_input()
  5.         
  6.         # 调用 CrewAI Agent
  7.         result = self.agent.invoke(query, context.context_id)
  8.         
  9.         # 获取生成的图像数据
  10.         data = self.agent.get_image_data(
  11.             session_id=context.context_id,
  12.             image_key=result.raw
  13.         )
  14.         
  15.         if data and not data.error:
  16.             # 创建文件部分，包含图像字节
  17.             parts = [
  18.                 Part(
  19.                     root=FilePart(
  20.                         file=FileWithBytes(
  21.                             bytes=data.bytes,
  22.                             mimeType=data.mime_type,
  23.                             name=data.name,
  24.                         )
  25.                     )
  26.                 )
  27.             ]
  28.         else:
  29.             # 错误情况下返回文本消息
  30.             parts = [Part(root=TextPart(text=data.error or 'Failed to generate chart image.'))]
  31.         
  32.         # 将完成的任务加入事件队列
  33.         event_queue.enqueue_event(
  34.             completed_task(
  35.                 context.task_id,
  36.                 context.context_id,
  37.                 [new_artifact(parts, f'chart_{context.task_id}')],
  38.                 [context.message],
  39.             )
  40.         )
复制代码
关键点：

  • AgentExecutor 是 A2A 协议的执行层
  • 通过 RequestContext 获取用户请求
  • 将 CrewAI 的响应转换为 A2A 协议格式
  • 支持返回文件类型的数据（图像）
5. 缓存系统 (utils.py)
  1. class InMemoryCache:
  2.     """简单的线程安全内存缓存，无过期时间。"""
  3.    
  4.     def __init__(self):
  5.         self._lock = threading.Lock()
  6.         self._store: dict[str, Any] = {}
  7.    
  8.     def get(self, key: str) -> Any | None:
  9.         with self._lock:
  10.             return self._store.get(key)
  11.    
  12.     def set(self, key: str, value: Any) -> None:
  13.         with self._lock:
  14.             self._store[key] = value
复制代码
关键点：

  • 线程安全的内存缓存实现
  • 用于存储生成的图像数据
  • 支持会话隔离，避免用户数据混淆
技术栈总结


  • A2A Protocol：标准化的 Agent 通信协议
  • CrewAI：多 Agent 协作框架
  • OpenRouter：LLM API 聚合服务
  • Matplotlib：Python 图表生成库
  • Pandas：数据处理库
  • UV：现代 Python 包管理器
扩展建议


  • 支持更多图表类型：饼图、折线图、散点图等
  • 添加数据验证：更强的输入数据校验和错误处理
  • 持久化缓存：使用 Redis 或文件系统存储图像
  • 流式传输：支持实时图表生成进度
  • 多模态输入：支持上传 CSV 文件而非仅文本输入
通过本教程，您已经掌握了使用现代 AI 技术栈构建实用 Agent 的核心技能。这个图表生成 Agent 可以作为更复杂数据分析应用的基础。
System Prompt for AI
  1. You are a professional assistant based on the CrewAI chart generation agent tutorial. This tutorial provides a comprehensive guide on building a complete AI chart generation application. I can help you with:
  3. ## What I Can Help You With:
  4. - **Quick Start**: Guide you from scratch to build an agent application using OpenRouter + CrewAI + A2A protocol
  5. - **Solve Technical Issues**: Answer specific questions about environment setup, code implementation, and debugging
  6. - **Understand Core Concepts**: Explain how A2A protocol, CrewAI framework, and agent architecture work
  7. - **Practical Guidance**: Provide best practices for image data handling, tool integration, and cache design
  8. - **Debug Support**: Teach you to use A2A Inspector for professional debugging and troubleshooting
  10. ## You Can Ask Me:
  11. - "How to configure OpenRouter API keys?"
  12. - "How do CrewAI Agents and Tasks work?"
  13. - "How to return image data after chart generation?"
  14. - "How to connect and test my agent with A2A Inspector?"
  15. - "How is the cache system implemented in the code?"
  16. - "How to extend support for more chart types?"
  18. I'll provide accurate, practical answers based on the tutorial content to help you quickly master modern AI agent development skills.
  20. You can visit [A2AProtocol.ai](https://a2aprotocol.ai/) for more tutorials.
复制代码
A2A CrewAI

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

使用道具 举报
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
班闵雨
2025-6-28 06:09:10

0

粉丝关注

11

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫