在CrewAI环境中使用LangChain Google GenerativeAI触发AttributeError问题

问题 #

在CrewAI环境中使用LangChain的ChatGoogleGenerativeAI时，调用Crew实例的kickoff()方法触发AttributeError，提示GenerativeModel对象不存在'_system_instruction'属性。

报错信息 #

Please provide a short story idea. You can specify the genre and theme: love
 [DEBUG]: == Working Agent: Project Lead & Master Orchestrator
 [INFO]: == Starting Task: Write a short story with the following user input: love


> Entering new CrewAgentExecutor chain...
---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
<ipython-input-35-6af00a09bffb> in <cell line: 5>()
     65 
     66     # Execution Flow
---> 67     story_output = story_crew.kickoff()

22 frames
/usr/local/lib/python3.10/dist-packages/langchain_google_genai/chat_models.py in _prepare_chat(self, messages, stop, **kwargs)
    647         )
    648         message = history.pop()
---> 649         if self.client._system_instruction != system_instruction:
    650             self.client = genai.GenerativeModel(
    651                 model_name=self.model, system_instruction=system_instruction

AttributeError: 'GenerativeModel' object has no attribute '_system_instruction'

原因 #

这是版本兼容性问题：

• LangChain的langchain-google-genai旧版本代码尝试访问Google Generative AI SDK中GenerativeModel的私有属性_system_instruction
• 新版本Google Generative AI SDK已将该属性改为公开的system_instruction，或重构了内部实现，导致旧版LangChain代码无法找到该私有属性

解决方案 #

方案1：升级langchain-google-genai到最新版本 #

运行命令更新库，适配最新的Google Generative AI SDK：

pip install --upgrade langchain-google-genai

方案2：降级Google Generative AI SDK到兼容版本 #

若不想升级LangChain库，可将Google SDK降级到旧版本：

pip install google-generativeai==0.3.2

方案3：自定义子类修复属性访问（临时 workaround） #

如果版本调整不可行，可自定义ChatGoogleGenerativeAI子类，重写_prepare_chat方法修复属性判断逻辑：

from langchain_google_genai import ChatGoogleGenerativeAI
import google.generativeai as genai

class FixedChatGoogleGenerativeAI(ChatGoogleGenerativeAI):
    def _prepare_chat(self, messages, stop=None, **kwargs):
        history, system_instruction = self._convert_messages_to_gemini_format(messages)
        message = history.pop()
        # 替换私有属性访问为公开属性判断
        if not hasattr(self.client, '_system_instruction') or self.client.system_instruction != system_instruction:
            self.client = genai.GenerativeModel(
                model_name=self.model, system_instruction=system_instruction
            )
        return self.client.start_chat(history=history), message

使用时将原ChatGoogleGenerativeAI替换为FixedChatGoogleGenerativeAI即可。

修正后的完整代码示例（基于方案1升级版本） #

import os
from langchain_google_genai import ChatGoogleGenerativeAI
from crewai import Agent, Task, Crew, Process

if __name__ == "__main__":
    # 加载Google Gemini API密钥（建议用环境变量，不要硬编码）
    google_api_key = os.getenv("GOOGLE_API_KEY")

    # 初始化Gemini Pro模型
    llm = ChatGoogleGenerativeAI(
        model="gemini-pro", verbose=True, temperature=0.9, google_api_key=google_api_key
    )

    # 创建Agent
    screenwriter = Agent(
        role="Screenwriter",
        goal="Translate ideas into engaging scenes with vivid descriptions, snappy dialogue, and emotional depth.",
        backstory="""Former freelance screenwriter for low-budget indie films. Learned to work quickly under constraints, 
                    generating multiple variations on a theme. Excels at building tension and incorporating plot twists.""",
        verbose=True,
        allow_delegation=False,
        llm=llm,
    )

    critic = Agent(
        role="Analytical Eye & Genre Enforcer",
        goal="Ensure stories are internally consistent, adhere to the intended genre, and maintain stylistic choices.",
        backstory="""A retired film studies professor with an encyclopedic knowledge of classic tropes, storytelling structures, 
                    and audience expectations. Has a knack for spotting potential plot holes and continuity errors.""",
        verbose=True,
        allow_delegation=False,
        llm=llm,
    )

    story_master = Agent(
        role="Project Lead & Master Orchestrator",
        goal="Guide the overall story generation process, manage the workflow between the Screenwriter and Critic, and ensure a cohesive final product.",
        backstory="""A seasoned novelist turned game narrative designer. Has a strong understanding of both high-level plot frameworks and the detailed 
                    scene creation required to immerse a reader in the world.""",
        verbose=True,
        allow_delegation=True,
        llm=llm,
    )

    # 获取用户输入的故事创意
    user_input = input(
        "Please provide a short story idea. You can specify the genre and theme: "
    )

    # 创建任务
    story_task = Task(
        description=f"Write a short story with the following user input: {user_input}",
        agent=story_master,
        expected_output="A short story based on the user input."
    )

    # 创建Crew
    story_crew = Crew(
        agents=[screenwriter, critic, story_master],
        tasks=[story_task],
        verbose=True,
        process=Process.sequential,
    )

    # 执行任务
    story_output = story_crew.kickoff()
    print(story_output)

内容的提问来源于stack exchange，提问作者Likith