文心一言结合langchain实现function call调用

  
  

   
   

    import json
   
   
   
   

    
   
   
   
   

    from langchain.chat_models import ChatOpenAI, QianfanChatEndpoint
   
   
   
   

    from langchain.schema import HumanMessage, AIMessage, ChatMessage
   
   
   
   

    
   
   
   
   

    
   
   
   
   

    def find_function(function_description):
   
   
   
   

        """
   
   
   
   

        根据函数描述信息查找本地项目中对应的函数代码
   
   
   
   

        :param function_description: 用户的输入
   
   
   
   

        """
   
   
   
   

        if not isinstance(function_description, dict):
   
   
   
   

            #if '```json' in function_description:
   
   
   
   

            #    function_description = extract_json(function_description)
   
   
   
   

            function_description = json.loads(function_description)
   
   
   
   

        return function_description.get('input')
   
   
   
   

    
   
   
   
   

    function_mapping = {"find_function": find_function}
   
   
   
   

    
   
   
   
   

    functions = [
   
   
   
   

        {
   
   
   
   

            "name": "find_function",
   
   
   
   

            "description": "根据函数描述信息查找本地项目中对应的函数代码",
   
   
   
   

            "parameters": {
   
   
   
   

                "type": "object",
   
   
   
   

                "properties": {
   
   
   
   

                    "function_description": {
   
   
   
   

                        "type": "string",
   
   
   
   

                        'description': '一句针对接口或函数简明扼要的功能说明。例如{"input": "获取指定SessionID对应的聊天记录"}'
   
   
   
   

                    },
   
   
   
   

                },
   
   
   
   

                "required": ["function_description"],
   
   
   
   

            }
   
   
   
   

        }
   
   
   
   

    ]
   
   
   
   

    
   
   
   
   

    
   
   
   
   

    def chat(query):
   
   
   
   

        prompt = '''你是一个资深的测试工程师，请根据函数描述信息查找本地项目中对应的函数代码。
   
   
   
   

    
   
   
   
   

        ## 函数定义：
   
   
   
   

        %s''' % query.strip('\n')
   
   
   
   

        print(f"prompt: {prompt}")
   
   
   
   

        llm = QianfanChatEndpoint(model="ERNIE-Bot-4")
   
   
   
   

        # llm = ChatOpenAI(model='gpt-4-1106-preview')
   
   
   
   

        message = llm.predict_messages(
   
   
   
   

            [HumanMessage(content=prompt)], functions=functions
   
   
   
   

        )
   
   
   
   

        print('message: ', message, type(message))
   
   
   
   

    
   
   
   
   

        function_call_info = message.additional_kwargs.get("function_call", None)
   
   
   
   

        if not function_call_info:
   
   
   
   

            return message
   
   
   
   

        else:
   
   
   
   

            function_name = function_call_info["name"]
   
   
   
   

            arg_name = json.loads(function_call_info["arguments"])
   
   
   
   

            print(f"函数名:{function_name}, 参数:{arg_name}, 类型:{type(arg_name)} ")
   
   
   
   

            function_response = find_function(**arg_name)
   
   
   
   

            return function_response
   
   
   
   

    
   
   
   
   

    meta_info = '''
   
   
   
   

        def chat_check(self, text):
   
   
   
   

            """
   
   
   
   

            检查聊天内容是否符合预期
   
   
   
   

            :params: text: 消息内容
   
   
   
   

            :return:
   
   
   
   

            {
   
   
   
   

                "code": 0,
   
   
   
   

                "msg": "success",
   
   
   
   

                "data": true,
   
   
   
   

            }
   
   
   
   

    '''
   
   
   
   

    
   
   
   
   

    
   
   
   
   

    res = chat(meta_info)
   
   
   
   

    print("answer:", res, end='\n\n')
   
   
   
   

    print(find_function(res))