接入你自己的智能体

互操作性是 CrewAI 的核心概念之一。本指南将演示如何把你自己的智能体接入 Crew 中运行。

适配器指南：接入你自己的智能体（LangGraph Agents、OpenAI Agents 等）

我们需要 3 个适配器，才能让来自不同框架的任意智能体在 Crew 中工作。

BaseAgentAdapter
BaseToolAdapter
BaseConverter

BaseAgentAdapter

这个抽象类定义了所有智能体适配器都必须实现的通用接口和功能。它继承自 BaseAgent，在保持与 CrewAI 框架兼容的同时，增加了适配器特定的要求。

必需方法：

def configure_tools
def configure_structured_output

创建你自己的适配器

要把来自其他框架的智能体（例如 LangGraph、Autogen、OpenAI Assistants）集成到 CrewAI 中，你需要通过继承 BaseAgentAdapter 创建自定义适配器。这个适配器充当兼容层，在 CrewAI 接口和外部智能体的特定要求之间进行转换。

下面是实现自定义适配器的方式：

继承 BaseAgentAdapter：

from crewai.agents.agent_adapters.base_agent_adapter import BaseAgentAdapter
from crewai.tools import BaseTool
from typing import List, Optional, Any, Dict

class MyCustomAgentAdapter(BaseAgentAdapter):
    # ... implementation details ...

实现 __init__：构造函数应调用父类构造函数 super().__init__(**kwargs)，并执行外部智能体所需的初始化。你可以使用 CrewAI 的 Agent 初始化过程中传入的可选 agent_config 字典来配置适配器和底层智能体。

def __init__(self, agent_config: Optional[Dict[str, Any]] = None, **kwargs: Any):
    super().__init__(agent_config=agent_config, **kwargs)
    # Initialize your external agent here, possibly using agent_config
    # Example: self.external_agent = initialize_my_agent(agent_config)
    print(f"Initializing MyCustomAgentAdapter with config: {agent_config}")

实现 configure_tools：这个抽象方法非常关键。它接收一个 CrewAI BaseTool 实例列表。你的实现必须把这些工具转换或适配成外部智能体框架所期望的格式。这可能包括包装它们、提取特定属性，或把它们注册到外部智能体实例中。

def configure_tools(self, tools: Optional[List[BaseTool]] = None) -> None:
    if tools:
        adapted_tools = []
        for tool in tools:
            # Adapt CrewAI BaseTool to the format your agent expects
            # Example: adapted_tool = adapt_to_my_framework(tool)
            # adapted_tools.append(adapted_tool)
            pass # Replace with your actual adaptation logic

        # Configure the external agent with the adapted tools
        # Example: self.external_agent.set_tools(adapted_tools)
        print(f"Configuring tools for MyCustomAgentAdapter: {adapted_tools}") # Placeholder
    else:
        # Handle the case where no tools are provided
        # Example: self.external_agent.set_tools([])
        print("No tools provided for MyCustomAgentAdapter.")

实现 configure_structured_output：当 CrewAI Agent 配置了结构化输出需求（例如 output_json 或 output_pydantic）时，会调用这个方法。你的适配器需要确保外部智能体能够满足这些要求。这可能涉及给外部智能体设置特定参数，或者确保其底层模型支持所请求的格式。如果外部智能体无法以与 CrewAI 预期兼容的方式支持结构化输出，你可能需要处理转换逻辑或抛出适当的错误。
```
def configure_structured_output(self, structured_output: Any) -> None:
    # Configure your external agent to produce output in the specified format
    # Example: self.external_agent.set_output_format(structured_output)
    self.adapted_structured_output = True # Signal that structured output is handled
    print(f"Configuring structured output for MyCustomAgentAdapter: {structured_output}")
```

通过实现这些方法，你的 MyCustomAgentAdapter 就能让自定义智能体在 CrewAI crew 中正常工作，并与任务和工具无缝交互。请记得把示例注释和 print 语句替换为与你所集成的外部框架相匹配的真实适配逻辑。

BaseToolAdapter 实现

BaseToolAdapter 类负责把 CrewAI 原生的 BaseTool 对象转换成你的特定外部智能体框架能够理解并使用的格式。不同的智能体框架（如 LangGraph、OpenAI Assistants 等）在工具定义和处理方式上各不相同，而 BaseToolAdapter 就充当翻译器。

下面是实现自定义工具适配器的方式：

继承 BaseToolAdapter：

from crewai.agents.agent_adapters.base_tool_adapter import BaseToolAdapter
from crewai.tools import BaseTool
from typing import List, Any

class MyCustomToolAdapter(BaseToolAdapter):
    # ... implementation details ...

实现 configure_tools：这是你必须实现的核心抽象方法。它接收一组传给智能体的 CrewAI BaseTool 实例。你的任务是遍历该列表，把每个 BaseTool 适配成外部框架期望的格式，并把转换后的工具存入基类构造函数中初始化的 self.converted_tools 列表。

def configure_tools(self, tools: List[BaseTool]) -> None:
    """Configure and convert CrewAI tools for the specific implementation."""
    self.converted_tools = [] # Reset in case it's called multiple times
    for tool in tools:
        # Sanitize the tool name if required by the target framework
        sanitized_name = self.sanitize_tool_name(tool.name)

        # --- Your Conversion Logic Goes Here ---
        # Example: Convert BaseTool to a dictionary format for LangGraph
        # converted_tool = {
        #     "name": sanitized_name,
        #     "description": tool.description,
        #     "parameters": tool.args_schema.schema() if tool.args_schema else {},
        #     # Add any other framework-specific fields
        # }

        # Example: Convert BaseTool to an OpenAI function definition
        # converted_tool = {
        #     "type": "function",
        #     "function": {
        #         "name": sanitized_name,
        #         "description": tool.description,
        #         "parameters": tool.args_schema.schema() if tool.args_schema else {"type": "object", "properties": {}},
        #     }
        # }

        # --- Replace above examples with your actual adaptation ---
        converted_tool = self.adapt_tool_to_my_framework(tool, sanitized_name) # Placeholder

        self.converted_tools.append(converted_tool)
        print(f"Adapted tool '{tool.name}' to '{sanitized_name}' for MyCustomToolAdapter") # Placeholder

    print(f"MyCustomToolAdapter finished configuring tools: {len(self.converted_tools)} adapted.") # Placeholder

# --- Helper method for adaptation (Example) ---
def adapt_tool_to_my_framework(self, tool: BaseTool, sanitized_name: str) -> Any:
    # Replace this with the actual logic to convert a CrewAI BaseTool
    # to the format needed by your specific external agent framework.
    # This will vary greatly depending on the target framework.
    adapted_representation = {
        "framework_specific_name": sanitized_name,
        "framework_specific_description": tool.description,
        "inputs": tool.args_schema.schema() if tool.args_schema else None,
        "implementation_reference": tool.run # Or however the framework needs to call it
    }
    # Also ensure the tool works both sync and async
    async def async_tool_wrapper(*args, **kwargs):
        output = tool.run(*args, **kwargs)
        if inspect.isawaitable(output):
            return await output
        else:
            return output

    adapted_tool = MyFrameworkTool(
        name=sanitized_name,
        description=tool.description,
        inputs=tool.args_schema.schema() if tool.args_schema else None,
        implementation_reference=async_tool_wrapper
    )

    return adapted_representation

使用适配器：通常，你会在 MyCustomAgentAdapter 的 configure_tools 方法中实例化 MyCustomToolAdapter，并在把工具配置给外部智能体前先处理它们。

# Inside MyCustomAgentAdapter.configure_tools
def configure_tools(self, tools: Optional[List[BaseTool]] = None) -> None:
    if tools:
        tool_adapter = MyCustomToolAdapter() # Instantiate your tool adapter
        tool_adapter.configure_tools(tools)  # Convert the tools
        adapted_tools = tool_adapter.tools() # Get the converted tools

        # Now configure your external agent with the adapted_tools
        # Example: self.external_agent.set_tools(adapted_tools)
        print(f"Configuring external agent with adapted tools: {adapted_tools}") # Placeholder
    else:
        # Handle no tools case
        print("No tools provided for MyCustomAgentAdapter.")

通过创建 BaseToolAdapter，你可以把工具转换逻辑和智能体适配逻辑解耦，使集成更清晰、更模块化。请记得把这些占位示例替换为你的外部智能体框架所需的真实转换逻辑。

BaseConverter

BaseConverterAdapter 在 CrewAI Task 需要智能体以特定结构化格式返回最终输出时发挥关键作用，例如 JSON 或 Pydantic 模型。它连接了 CrewAI 的结构化输出要求与外部智能体的能力。

它的核心职责是：

为结构化输出配置智能体： 根据 Task 的要求（output_json 或 output_pydantic），指示关联的 BaseAgentAdapter（以及间接的外部智能体）应返回什么格式。
增强系统提示词： 修改智能体的系统提示词，加入关于如何生成所需结构化输出的明确指令。
结果后处理： 获取智能体的原始输出，并尝试按照所需结构进行解析、校验和格式化，最终返回一个字符串表示（例如 JSON 字符串）。

下面是实现自定义转换适配器的方式：

继承 BaseConverterAdapter：

from crewai.agents.agent_adapters.base_converter_adapter import BaseConverterAdapter
# Assuming you have your MyCustomAgentAdapter defined
# from .my_custom_agent_adapter import MyCustomAgentAdapter
from crewai.task import Task
from typing import Any

class MyCustomConverterAdapter(BaseConverterAdapter):
    # Store the expected output type (e.g., 'json', 'pydantic', 'text')
    _output_type: str = 'text'
    _output_schema: Any = None # Store JSON schema or Pydantic model

    # ... implementation details ...

实现 __init__：构造函数必须接收与之配合工作的对应 agent_adapter 实例。

def __init__(self, agent_adapter: Any): # Use your specific AgentAdapter type hint
    self.agent_adapter = agent_adapter
    print(f"Initializing MyCustomConverterAdapter for agent adapter: {type(agent_adapter).__name__}")

实现 configure_structured_output：这个方法接收 CrewAI Task 对象。你需要检查任务的 output_json 和 output_pydantic 属性来确定所需的输出结构。将这些信息存储起来（例如，放入 _output_type 和 _output_schema），并在必要时调用 self.agent_adapter 上的配置方法，让外部智能体为结构化输出做好准备（这可能已经在智能体适配器的 configure_structured_output 中部分处理过）。

def configure_structured_output(self, task: Task) -> None:
    """Configure the expected structured output based on the task."""
    if task.output_pydantic:
        self._output_type = 'pydantic'
        self._output_schema = task.output_pydantic
        print(f"Converter: Configured for Pydantic output: {self._output_schema.__name__}")
    elif task.output_json:
        self._output_type = 'json'
        self._output_schema = task.output_json
        print(f"Converter: Configured for JSON output with schema: {self._output_schema}")
    else:
        self._output_type = 'text'
        self._output_schema = None
        print("Converter: Configured for standard text output.")

    # Optionally, inform the agent adapter if needed
    # self.agent_adapter.set_output_mode(self._output_type, self._output_schema)

实现 enhance_system_prompt：这个方法接收智能体的基础系统提示词字符串，并应追加当前 _output_type 和 _output_schema 相关的说明。目标是引导驱动智能体的 LLM 生成符合正确格式的输出。

def enhance_system_prompt(self, base_prompt: str) -> str:
    """Enhance the system prompt with structured output instructions."""
    if self._output_type == 'text':
        return base_prompt # No enhancement needed for plain text

    instructions = "\n\nYour final answer MUST be formatted as "
    if self._output_type == 'json':
        schema_str = json.dumps(self._output_schema, indent=2)
        instructions += f"a JSON object conforming to the following schema:\n```json\n{schema_str}\n```"
    elif self._output_type == 'pydantic':
        schema_str = json.dumps(self._output_schema.model_json_schema(), indent=2)
        instructions += f"a JSON object conforming to the Pydantic model '{self._output_schema.__name__}' with the following schema:\n```json\n{schema_str}\n```"

    instructions += "\nEnsure your entire response is ONLY the valid JSON object, without any introductory text, explanations, or concluding remarks."

    print(f"Converter: Enhancing prompt for {self._output_type} output.")
    return base_prompt + instructions

Note: The exact prompt engineering might need tuning based on the agent/LLM being used.

实现 post_process_result：这个方法接收来自智能体的原始字符串输出。如果请求的是结构化输出（json 或 pydantic），你应该尝试把字符串解析成预期格式。处理可能的解析错误（例如记录日志、尝试简单修复，或者抛出异常）。关键是，这个方法必须始终返回字符串，即使中间格式曾是字典或 Pydantic 对象（例如再序列化回 JSON 字符串）。

import json
from pydantic import ValidationError

def post_process_result(self, result: str) -> str:
    """Post-process the agent's result to ensure it matches the expected format."""
    print(f"Converter: Post-processing result for {self._output_type} output.")
    if self._output_type == 'json':
        try:
            # Attempt to parse and re-serialize to ensure validity and consistent format
            parsed_json = json.loads(result)
            # Optional: Validate against self._output_schema if it's a JSON schema dictionary
            # from jsonschema import validate
            # validate(instance=parsed_json, schema=self._output_schema)
            return json.dumps(parsed_json)
        except json.JSONDecodeError as e:
            print(f"Error: Failed to parse JSON output: {e}\nRaw output:\n{result}")
            # Handle error: return raw, raise exception, or try to fix
            return result # Example: return raw output on failure
        # except Exception as e: # Catch validation errors if using jsonschema
        #     print(f"Error: JSON output failed schema validation: {e}\nRaw output:\n{result}")
        #     return result
    elif self._output_type == 'pydantic':
        try:
            # Attempt to parse into the Pydantic model
            model_instance = self._output_schema.model_validate_json(result)
            # Return the model serialized back to JSON
            return model_instance.model_dump_json()
        except ValidationError as e:
            print(f"Error: Failed to validate Pydantic output: {e}\nRaw output:\n{result}")
            # Handle error
            return result # Example: return raw output on failure
        except json.JSONDecodeError as e:
             print(f"Error: Failed to parse JSON for Pydantic model: {e}\nRaw output:\n{result}")
             return result
    else: # 'text'
        return result # No processing needed for plain text

By implementing these methods, your MyCustomConverterAdapter ensures that structured output requests from CrewAI tasks are correctly handled by your integrated external agent, improving the reliability and usability of your custom agent within the CrewAI framework.

开箱即用的适配器

我们提供以下框架的现成适配器：

LangGraph
OpenAI Agents

使用适配后的智能体启动 crew

import json
import os
from typing import List

from crewai_tools import SerperDevTool
from src.crewai import Agent, Crew, Task
from langchain_openai import ChatOpenAI
from pydantic import BaseModel

from crewai.agents.agent_adapters.langgraph.langgraph_adapter import (
    LangGraphAgentAdapter,
)
from crewai.agents.agent_adapters.openai_agents.openai_adapter import OpenAIAgentAdapter

# CrewAI Agent
code_helper_agent = Agent(
    role="Code Helper",
    goal="Help users solve coding problems effectively and provide clear explanations.",
    backstory="You are an experienced programmer with deep knowledge across multiple programming languages and frameworks. You specialize in solving complex coding challenges and explaining solutions clearly.",
    allow_delegation=False,
    verbose=True,
)
# OpenAI Agent Adapter
link_finder_agent = OpenAIAgentAdapter(
    role="Link Finder",
    goal="Find the most relevant and high-quality resources for coding tasks.",
    backstory="You are a research specialist with a talent for finding the most helpful resources. You're skilled at using search tools to discover documentation, tutorials, and examples that directly address the user's coding needs.",
    tools=[SerperDevTool()],
    allow_delegation=False,
    verbose=True,
)

# LangGraph Agent Adapter
reporter_agent = LangGraphAgentAdapter(
    role="Reporter",
    goal="Report the results of the tasks.",
    backstory="You are a reporter who reports the results of the other tasks",
    llm=ChatOpenAI(model="gpt-4o"),
    allow_delegation=True,
    verbose=True,
)


class Code(BaseModel):
    code: str


task = Task(
    description="Give an answer to the coding question: {task}",
    expected_output="A thorough answer to the coding question: {task}",
    agent=code_helper_agent,
    output_json=Code,
)
task2 = Task(
    description="Find links to resources that can help with coding tasks. Use the serper tool to find resources that can help.",
    expected_output="A list of links to resources that can help with coding tasks",
    agent=link_finder_agent,
)


class Report(BaseModel):
    code: str
    links: List[str]


task3 = Task(
    description="Report the results of the tasks.",
    expected_output="A report of the results of the tasks. this is the code produced and then the links to the resources that can help with the coding task.",
    agent=reporter_agent,
    output_json=Report,
)
# Use in CrewAI
crew = Crew(
    agents=[code_helper_agent, link_finder_agent, reporter_agent],
    tasks=[task, task2, task3],
    verbose=True,
)

result = crew.kickoff(
    inputs={"task": "How do you implement an abstract class in python?"}
)

# Print raw result first
print("Raw result:", result)

# Handle result based on its type
if hasattr(result, "json_dict") and result.json_dict:
    json_result = result.json_dict
    print("\nStructured JSON result:")
    print(f"{json.dumps(json_result, indent=2)}")

    # Access fields safely
    if isinstance(json_result, dict):
        if "code" in json_result:
            print("\nCode:")
            print(
                json_result["code"][:200] + "..."
                if len(json_result["code"]) > 200
                else json_result["code"]
            )

        if "links" in json_result:
            print("\nLinks:")
            for link in json_result["links"][:5]:  # Print first 5 links
                print(f"- {link}")
            if len(json_result["links"]) > 5:
                print(f"...and {len(json_result['links']) - 5} more links")
elif hasattr(result, "pydantic") and result.pydantic:
    print("\nPydantic model result:")
    print(result.pydantic.model_dump_json(indent=2))
else:
    # Fallback to raw output
    print("\nNo structured result available, using raw output:")
    print(result.raw[:500] + "..." if len(result.raw) > 500 else result.raw)