콘텐츠로 이동
CrewAI 문서

에이전트 직접 가져오기

상호 운용성은 CrewAI의 핵심 개념입니다. 이 가이드에서는 Crew 내에서 작동하는 여러분만의 에이전트를 어떻게 도입할 수 있는지 보여줍니다.

에이전트 직접 가져오기 어댑터 가이드 (Langgraph Agents, OpenAI Agents 등…)

섹션 제목: “에이전트 직접 가져오기 어댑터 가이드 (Langgraph Agents, OpenAI Agents 등…)”

다양한 프레임워크의 에이전트를 crew에서 작동하도록 하려면 3가지 어댑터가 필요합니다.

  1. BaseAgentAdapter
  2. BaseToolAdapter
  3. BaseConverter

BaseAgentAdapter

섹션 제목: “BaseAgentAdapter”

이 추상 클래스는 모든 agent adapter가 구현해야 하는 공통 인터페이스와 기능을 정의합니다. BaseAgent를 확장하여 CrewAI 프레임워크와의 호환성을 유지하면서 adapter별 요구 사항을 추가합니다.

필수 메서드:

  1. def configure_tools
  2. def configure_structured_output

자신만의 Adapter 생성하기

섹션 제목: “자신만의 Adapter 생성하기”

다른 프레임워크(예: LangGraph, Autogen, OpenAI Assistants)의 agent를 CrewAI에 통합하려면, BaseAgentAdapter를 상속하여 커스텀 adapter를 생성해야 합니다. 이 adapter는 호환성 계층 역할을 하며 CrewAI 인터페이스와 외부 agent의 특정 요구사항 사이를 변환합니다.

커스텀 adapter를 구현하는 방법은 다음과 같습니다:

  1. 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 ...

  2. __init__ 구현하기: 생성자는 부모 클래스 생성자 super().__init__(**kwargs)를 호출하고, 외부 agent에 특화된 초기화를 수행해야 합니다. CrewAI의 Agent 초기화 시 사용할 수 있는 선택적 agent_config 딕셔너리를 이용하여 adapter와 하위 agent를 구성할 수 있습니다.

    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}")

  3. configure_tools 구현하기: 이 추상 메서드는 매우 중요합니다. CrewAI BaseTool 인스턴스 리스트를 받습니다. 구현 시, 이 도구들을 외부 agent 프레임워크에서 기대하는 형식으로 변환 또는 적응시켜야 합니다. 래핑하거나, 특정 속성 추출, 혹은 외부 agent 인스턴스에 등록하는 작업이 필요할 수 있습니다.

    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.")

  4. configure_structured_output 구현하기: CrewAI Agent가 구조화된 출력 요구사항(예: output_json 또는 output_pydantic)으로 구성될 때 이 메서드가 호출됩니다. adapter에서 외부 agent가 이러한 요구사항을 준수하도록 설정해야 합니다. 이는 외부 agent에 특정 파라미터를 설정하거나, 해당 모델이 요청된 형식을 지원하는지 확인하는 것이 포함될 수 있습니다. 외부 agent가 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는 커스텀 agent 구현이 CrewAI crew 내에서 올바로 동작할 수 있도록 하여, task 및 도구들과 매끄럽게 상호작용할 수 있게 됩니다. 예시 주석 및 print문은 실제로 통합하려는 외부 agent 프레임워크에 맞춘 로직으로 교체해야 한다는 점을 기억하세요.

BaseToolAdapter 구현

섹션 제목: “BaseToolAdapter 구현”

BaseToolAdapter 클래스는 CrewAI의 기본 BaseTool 객체를 외부 에이전트 프레임워크가 이해하고 활용할 수 있는 형식으로 변환하는 역할을 합니다. 각각의 에이전트 프레임워크(LangGraph, OpenAI Assistants 등)는 도구를 정의하고 처리하는 고유한 방식을 가지고 있으며, BaseToolAdapter는 이들 간의 변환자 역할을 합니다.

사용자 정의 툴 어댑터를 구현하는 방법은 다음과 같습니다:

  1. 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 ...

  2. 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

  3. 어댑터 사용하기: 일반적으로, 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

섹션 제목: “BaseConverter”

BaseConverterAdapter는 CrewAI의 Task에서 에이전트가 JSON이나 Pydantic 모델과 같이 특정 구조화된 포맷으로 최종 출력을 반환해야 할 때 중요한 역할을 합니다. 이 어댑터는 CrewAI의 구조화된 출력 요구사항과 외부 에이전트의 기능 사이를 이어주는 다리 역할을 합니다.

주요 책임은 다음과 같습니다:

  1. 에이전트의 구조화된 출력 구성: Task의 요구사항(output_json 또는 output_pydantic)에 따라 연결된 BaseAgentAdapter(그리고 간접적으로 외부 에이전트)에게 어떤 포맷이 요구되는지 지시합니다.
  2. 시스템 프롬프트 확장: 에이전트의 시스템 프롬프트를 수정하여 필요한 구조로 출력물을 생성하는 방법에 대한 명확한 지침을 추가합니다.
  3. 결과 후처리: 에이전트로부터 받은 원시 출력을 받아, 요구되는 구조에 따라 파싱, 검증 및 포맷팅을 시도한 후, 최종적으로 문자열(예: JSON 문자열) 형태로 반환합니다.

사용자 지정 컨버터 어댑터를 구현하는 방법은 다음과 같습니다:

  1. 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 ...

  2. __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__}")

  3. configure_structured_output 구현: 이 메서드는 CrewAI Task 객체를 받습니다. 작업의 output_json 및 output_pydantic 속성을 확인하여 요구되는 출력 구조를 결정해야 합니다. 해당 정보(예: _output_type 및 _output_schema)를 저장하고, 필요하다면 외부 에이전트가 구조화된 출력에 대해 별도의 설정이 필요한 경우 self.agent_adapter에 구성 메서드를 호출할 수 있습니다(일부는 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)

  4. 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

    참고: 실제 프롬프트 엔지니어링은 사용하는 에이전트/LLM에 따라 조정이 필요할 수 있습니다.

  5. 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

이러한 메서드를 구현함으로써, MyCustomConverterAdapter는 CrewAI 작업의 구조화된 출력 요청이 통합된 외부 에이전트에서 올바르게 처리될 수 있게 하여, 사용자가 CrewAI 프레임워크 내에서 맞춤형 에이전트를 더욱 신뢰성 있고 유용하게 사용할 수 있도록 합니다.

기본 제공 어댑터

섹션 제목: “기본 제공 어댑터”

다음 프레임워크에 대해 기본 제공 어댑터를 제공합니다:

  1. LangGraph
  2. OpenAI Agents

적응형 에이전트로 crew 시작하기:

섹션 제목: “적응형 에이전트로 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)