使用Groq API构建Python编码代理时触发OpenAI API认证错误求助

问题原因 #

你遇到的OpenAI API错误并非来自Groq LLM的调用，而是CrewAI默认开启的memory功能在底层使用了Chroma向量数据库，而Chroma默认采用OpenAI的Embedding模型生成向量，这部分未被配置为使用Groq或其他非OpenAI的Embedding服务，因此触发了OpenAI的API密钥验证。

解决方法 #

方法1：禁用CrewAI的Memory功能 #

如果不需要Agent间的上下文记忆，可直接关闭memory：
修改Crew初始化代码，将memory=True改为memory=False：

crew = Crew(
    agents=[coderAgent, DebuggerAgent],
    tasks=[coding_task, debug_task],
    process=Process.sequential,
    memory=False,  # 关闭默认memory模块
    cache=True,
    max_rpm=25,
    share_crew=True
)

同时，将两个Agent的memory=True也改为memory=False，避免内存模块初始化。

方法2：配置自定义Embedding模型 #

如果需要保留记忆功能，需替换Chroma默认的Embedding模型，可选择开源本地模型或Groq的Embedding服务：

示例1：使用开源Embedding模型（如BAAI的bge模型）

from langchain_community.embeddings import HuggingFaceEmbeddings
from crewai.memory import ContextualMemory
from crewai.memory.storage import RAGStorage

# 初始化开源Embedding模型
embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-small-en-v1.5")

# 创建自定义内存存储
storage = RAGStorage(embedding_function=embeddings)
contextual_memory = ContextualMemory(storage=storage)

# 初始化Crew时使用自定义memory
crew = Crew(
    agents=[coderAgent, DebuggerAgent],
    tasks=[coding_task, debug_task],
    process=Process.sequential,
    memory=contextual_memory,  # 使用自定义内存配置
    cache=True,
    max_rpm=25,
    share_crew=True
)

示例2：使用Groq的Embedding服务

若Groq提供Embedding API，可使用对应的LangChain集成：

from langchain_groq import GroqEmbeddings
from crewai.memory import ContextualMemory
from crewai.memory.storage import RAGStorage

# 初始化Groq Embedding
embeddings = GroqEmbeddings(api_key="YOUR_GROQ_API_KEY", model="your-groq-embedding-model")

# 创建自定义内存存储并传入Crew
storage = RAGStorage(embedding_function=embeddings)
contextual_memory = ContextualMemory(storage=storage)

crew = Crew(
    agents=[coderAgent, DebuggerAgent],
    tasks=[coding_task, debug_task],
    process=Process.sequential,
    memory=contextual_memory,
    cache=True,
    max_rpm=25,
    share_crew=True
)

LangGraph实现类似代理的示例结构 #

如果想用LangGraph替代CrewAI实现编码+调试的多Agent流程，可参考以下基础结构：

from langgraph.graph import StateGraph, END
from langchain_groq import ChatGroq
from langchain.tools import PythonREPLTool, DuckDuckGoSearchRun
from typing import TypedDict

# 定义状态结构
class AgentState(TypedDict):
    topic: str
    code: str
    debug_feedback: str

# 初始化工具与LLM
llm = ChatGroq(temperature=0, api_key="YOUR_GROQ_API_KEY", model="llama3-70b-8192")
python_repl = PythonREPLTool()
search_tool = DuckDuckGoSearchRun()

# 编码Agent节点
def coder_node(state: AgentState):
    prompt = f"Write bug-free pure Python code for: {state['topic']}. Only return code."
    response = llm.invoke(prompt)
    return {"code": response.content}

# 调试Agent节点
def debugger_node(state: AgentState):
    try:
        result = python_repl.run(state['code'])
        return {"debug_feedback": f"Code executed successfully: {result}"}
    except Exception as e:
        search_result = search_tool.run(f"Python error: {str(e)}")
        return {"debug_feedback": f"Error: {str(e)}. Fix suggestion: {search_result}"}

# 判断是否需要重新编码
def should_recode(state: AgentState):
    if "error" in state["debug_feedback"].lower():
        return "coder_node"
    return END

# 构建并运行流程
workflow = StateGraph(AgentState)
workflow.add_node("coder_node", coder_node)
workflow.add_node("debugger_node", debugger_node)
workflow.set_entry_point("coder_node")
workflow.add_edge("coder_node", "debugger_node")
workflow.add_conditional_edges("debugger_node", should_recode)

app = workflow.compile()
result = app.invoke({"topic": "Write code for maximum subarray sum in an Array using pure Python"})
print(result)

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