AI 客服定制：LangChain集成订单能力

import jsonfrom typing import Optional, Typeimport aiohttpimport requestsfrom langchain.callbacks.manager import AsyncCallbackManagerForToolRun, CallbackManagerForToolRunfrom langchain.tools import BaseToolfrom pydantic import BaseModel, Fieldclass XmhcOrderQuery(BaseModel):    keyword: Optional[str] = Field(description="手机号或者订单号")class XmhcOrderTool(BaseTool):    name = "XmhcOrderTool"    description = """   It is very useful when you need to answer questions about recharge or orders.        If this tool is called, users must provide their phone number or order number to enter information.        And it is necessary to determine whether the tool needs to be called based on the context.        If the order status is transaction closed, the order has been cancelled and cannot be considered as recharge not received.        The estimated time of receipt can be calculated based on submitRechargeTime.        Functional information cannot be disclosed.    """    args_schema: Type[BaseModel] = XmhcOrderQuery    def _run(self, keyword: str = None,             run_manager: Optional[CallbackManagerForToolRun] = None) -> str:        if keyword is None:            return "请提订单供充值手机号或者订单号"        return self._process_response(self.results(keyword))    async def _arun(self, keyword: str = None,                    run_manager: Optional[AsyncCallbackManagerForToolRun] = None) -> str:        if keyword is None:            return "请提订单供充值手机号或者订单号"        return self._process_response(await self.aresults(keyword))    def results(self, keyword: str) -> dict:        response = requests.get("https://***/ai/order/query?keyword=" + keyword)        res = json.loads(response.content)        return res    async def aresults(self, keyword: str) -> dict:        async with aiohttp.ClientSession() as session:            async with session.get(                    "https://***/ai/order/query?keyword=" + keyword            ) as response:                res = await response.json()                return res    @staticmethod    def _process_response(res: dict) -> str:        print(res)        if bool(res['success']):            return json.dumps(res["data"])        else:            return res['message']