DeepSeek AI驱动的ReAct代理

无论你对DeepSeek有何看法，它现在已经在游戏中了，并且是一个值得评估的实用选项！为了做出明智的决策，在特定需求背景下评估DeepSeek与其他LLM。

在这篇文章（以及相应的GitHub仓库）中，我将引导你如何在Vertex AI端点上部署DeepSeek模型并使用Langchain构建ReAct Agent，以便你可以评估其性能。

构建自定义LLM包装器

为了连接Vertex AI上的DeepSeek和LangChain框架，我们首先创建一个自定义LLM包装器：

class CustomLLM(LLM):  
    model_name: str = Field(default="vertex-ai")  
    generation_config: Dict[str, Any] = Field(default_factory=dict)  
  
    @property  
    def _llm_type(self) -> str:  
        return "custom-vertex-ai"  
  
    def _call(self, prompt: str, stop: Optional[List[str]] = None) -> str:  
        prediction_request = {  
            "instances": [  
                {  
                    "@requestFormat": "textGeneration",  
                    "prompt": prompt,  
                    "maxOutputTokens": self.generation_config.get("max_output_tokens", 2048),  
                    "temperature": self.generation_config.get("temperature", 0.8),  
                    "candidateCount": self.generation_config.get("candidate_count", 1),  
                }  
            ]  
        }  
        response = endpoint.predict(instances=prediction_request["instances"])  
        return response.predictions[0] if response.predictions else ""

这个CustomLLM类封装了与Vertex AI端点的交互，允许我们配置参数如max_output_tokens、temperature和candidate_count。

实现ReAct Agent执行器

接下来，我们创建一个ReActAgentExecutor来协调代理的行为：

class ReActAgentExecutor:  
    """  
    一个用于运行具有指定配置和工具的ReAct代理的类。  
    """  
    def __init__(  
        self,  
        model: str,  
        generation_config: Dict,  
        max_iterations: int,  
        max_execution_time: int,  
        google_api_key: str=GOOGLE_API_KEY,  
        cse_id: str=CSE_ID,  
    ):  
        self.model = model  
        self.generation_config = generation_config  
        self.max_iterations = max_iterations  
        self.max_execution_time = max_execution_time  
        self.google_api_key = google_api_key  
        self.cse_id = cse_id  
        self.llm = None  
        self.tools = None  
        self.agent = None  
        self.agent_executor = None  
        self.token_callback = None  
  
        self._setup_llm()  
        self._setup_tools()  
        self._setup_agent()  
  
    def _setup_llm(self):  
        """初始化自定义LLM."""  
        self.llm = CustomLLM(model=self.model, generation_config=self.generation_config)  
  
    def _setup_tools(self):  
        """设置代理的工具."""  
        search = GoogleSearchAPIWrapper(  
            google_api_key=self.google_api_key, google_cse_id=self.cse_id  
        )  
        self.tools = [  
            Tool(  
                name="Google Search",  
                func=search.run,  
                description="用于查找当前事件、比较或不同观点的信息。",  
            ),  
        ]  
  
    def _setup_agent(self):  
        """设置ReAct代理和执行器."""  
        prompt = hub.pull("hwchase17/react")  
        system_instruction = "一旦找到答案，请只返回Yes或No"  
        prompt.template = system_instruction + "\n" + prompt.template  
  
        self.agent = create_react_agent(self.llm, self.tools, prompt)  
        self.token_callback = TokenCountingCallbackHandler(self.model)  
        self.agent_executor = AgentExecutor(  
            agent=self.agent,  
            tools=self.tools,  
            verbose=False,  
            handle_parsing_errors=True,  
            max_iterations=self.max_iterations,  
            max_execution_time=self.max_execution_time,  
            callbacks=[self.token_callback],  
        )  
  
    def run(self, input_data: Union[Dict, str]) -> Dict:  
        """  
        使用给定输入数据运行代理。  
        """  
        if isinstance(input_data, str):  
            input_data = {"input": input_data}  
  
        start_time = time.time()  
        try:  
            result = self.agent_executor.invoke(input_data)  
            result["total_token"] = self.token_callback.total_token  
            self.token_callback.reset()  
        except Exception as e:  
            print(f"发生错误：{e}")  
            result = {"error": str(e)}  
        end_time = time.time()  
        result["wall_time"] = end_time - start_time  
  
        return result

令牌计数回调处理器

class TokenCountingCallbackHandler(BaseCallbackHandler):  
    """语言模型使用的令牌计数回调处理器。"""  
    def __init__(self, model_name: str):  
        self.model_name = model_name  
        self.total_token = 0  
  
    def reset(self):  
        """重置下一个链运行的计数器。"""  
        self.total_token = 0

此回调处理器提供了代理执行期间令牌消耗的洞察，这对于成本优化非常有价值。

整合所有内容

通过结合DeepSeek模型、Vertex AI和ReAct代理框架，我们可以构建强大的应用程序，能够处理复杂任务。本教程提供了一个进一步探索和实验的基础。

原文链接：ReAct Agent with DeepSeek on Vertex AI

汇智网翻译整理，转载请标明出处